Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import com.google.bigtable.v2.SessionCheckAndMutateRowResponse;
import com.google.bigtable.v2.SessionMutateRowRequest;
import com.google.bigtable.v2.SessionMutateRowResponse;
import com.google.bigtable.v2.SessionMutateRowsRequest;
import com.google.bigtable.v2.SessionMutateRowsResponse;
import com.google.bigtable.v2.SessionReadRowRequest;
import com.google.bigtable.v2.SessionReadRowResponse;
import com.google.cloud.bigtable.data.v2.internal.channels.ChannelPool;
Expand Down Expand Up @@ -77,6 +79,7 @@ static AuthorizedViewAsync createAndStart(
VRpcDescriptor.READ_ROW_AUTH_VIEW,
VRpcDescriptor.MUTATE_ROW_AUTH_VIEW,
VRpcDescriptor.CHECK_AND_MUTATE_ROW_AUTH_VIEW,
VRpcDescriptor.MUTATE_ROWS_AUTH_VIEW,
featureFlags,
clientInfo,
configManager,
Expand Down Expand Up @@ -120,6 +123,13 @@ public CompletableFuture<SessionCheckAndMutateRowResponse> checkAndMutateRow(
return f;
}

public CompletableFuture<SessionMutateRowsResponse> mutateRows(
SessionMutateRowsRequest req, Deadline deadline) {
UnaryResponseFuture<SessionMutateRowsResponse> f = new UnaryResponseFuture<>();
base.mutateRows(req, f, deadline);
return f;
}

@Override
public void close() {
this.base.close();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,9 @@ public static MaterializedViewAsync createAndStart(
openReq,
VRpcDescriptor.MATERIALIZED_VIEW_SESSION,
VRpcDescriptor.READ_ROW_MAT_VIEW,
// Materialized views are read-only, so mutateRow and checkAndMutateRow don't apply.
// Materialized views are read-only, so mutateRow, checkAndMutateRow, and mutateRows
// don't apply.
null,
null,
null,
featureFlags,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
import com.google.bigtable.v2.SessionCheckAndMutateRowResponse;
import com.google.bigtable.v2.SessionMutateRowRequest;
import com.google.bigtable.v2.SessionMutateRowResponse;
import com.google.bigtable.v2.SessionMutateRowsRequest;
import com.google.bigtable.v2.SessionMutateRowsResponse;
import com.google.bigtable.v2.SessionReadRowRequest;
import com.google.bigtable.v2.SessionReadRowResponse;
import com.google.cloud.bigtable.data.v2.internal.channels.ChannelPool;
Expand Down Expand Up @@ -75,6 +77,7 @@ public static TableAsync createAndStart(
VRpcDescriptor.READ_ROW,
VRpcDescriptor.MUTATE_ROW,
VRpcDescriptor.CHECK_AND_MUTATE_ROW,
VRpcDescriptor.MUTATE_ROWS,
featureFlags,
clientInfo,
configManager,
Expand Down Expand Up @@ -132,4 +135,12 @@ public CompletableFuture<SessionCheckAndMutateRowResponse> checkAndMutateRow(
base.checkAndMutateRow(req, f, deadline);
return f;
}

// TODO: get deadline from compatibility layer

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this comment can be removed. My PR is removing it in all other places.

public CompletableFuture<SessionMutateRowsResponse> mutateRows(
SessionMutateRowsRequest req, Deadline deadline) {
UnaryResponseFuture<SessionMutateRowsResponse> f = new UnaryResponseFuture<>();
base.mutateRows(req, f, deadline);
return f;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
import com.google.bigtable.v2.SessionCheckAndMutateRowResponse;
import com.google.bigtable.v2.SessionMutateRowRequest;
import com.google.bigtable.v2.SessionMutateRowResponse;
import com.google.bigtable.v2.SessionMutateRowsRequest;
import com.google.bigtable.v2.SessionMutateRowsResponse;
import com.google.bigtable.v2.SessionReadRowRequest;
import com.google.bigtable.v2.SessionReadRowResponse;
import com.google.cloud.bigtable.data.v2.internal.channels.ChannelPool;
Expand Down Expand Up @@ -55,6 +57,8 @@ class TableBase implements AutoCloseable {
mutateRowDescriptor;
private final VRpcDescriptor<?, SessionCheckAndMutateRowRequest, SessionCheckAndMutateRowResponse>
checkAndMutateRowDescriptor;
private final VRpcDescriptor<?, SessionMutateRowsRequest, SessionMutateRowsResponse>
mutateRowsDescriptor;

static <ReqT extends Message> TableBase createAndStart(
ReqT openReq,
Expand All @@ -63,6 +67,7 @@ static <ReqT extends Message> TableBase createAndStart(
VRpcDescriptor<?, SessionMutateRowRequest, SessionMutateRowResponse> mutateRowDescriptor,
VRpcDescriptor<?, SessionCheckAndMutateRowRequest, SessionCheckAndMutateRowResponse>
checkAndMutateRowDescriptor,
VRpcDescriptor<?, SessionMutateRowsRequest, SessionMutateRowsResponse> mutateRowsDescriptor,
FeatureFlags featureFlags,
ClientInfo clientInfo,
ClientConfigurationManager configManager,
Expand Down Expand Up @@ -100,6 +105,7 @@ static <ReqT extends Message> TableBase createAndStart(
readRowDescriptor,
mutateRowDescriptor,
checkAndMutateRowDescriptor,
mutateRowsDescriptor,
metrics,
timer,
userCallbackExecutor);
Expand All @@ -112,13 +118,15 @@ static <ReqT extends Message> TableBase createAndStart(
VRpcDescriptor<?, SessionMutateRowRequest, SessionMutateRowResponse> mutateRowDescriptor,
VRpcDescriptor<?, SessionCheckAndMutateRowRequest, SessionCheckAndMutateRowResponse>
checkAndMutateRowDescriptor,
VRpcDescriptor<?, SessionMutateRowsRequest, SessionMutateRowsResponse> mutateRowsDescriptor,
Metrics metrics,
BigtableTimer timer,
Executor userCallbackExecutor) {
this.sessionPool = sessionPool;
this.readRowDescriptor = readRowDescriptor;
this.mutateRowDescriptor = mutateRowDescriptor;
this.checkAndMutateRowDescriptor = checkAndMutateRowDescriptor;
this.mutateRowsDescriptor = mutateRowsDescriptor;
this.metrics = metrics;
this.timer = timer;
this.userCallbackExecutor = userCallbackExecutor;
Expand Down Expand Up @@ -174,4 +182,19 @@ public void checkAndMutateRow(
new VOperationImpl<>(retry, Context.current(), userCallbackExecutor, tracer, deadline, false)
.start(req, listener);
}

public void mutateRows(
SessionMutateRowsRequest req,
VRpcListener<SessionMutateRowsResponse> listener,
Deadline deadline) {
RetryingVRpc<SessionMutateRowsRequest, SessionMutateRowsResponse> retry =
new RetryingVRpc<>(() -> sessionPool.newCall(mutateRowsDescriptor), timer);
boolean idempotent = Util.isIdempotent(req);
VRpcTracer tracer =
metrics.newTableTracer(sessionPool.getInfo(), mutateRowsDescriptor, deadline);

new VOperationImpl<>(
retry, Context.current(), userCallbackExecutor, tracer, deadline, idempotent)
.start(req, listener);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import com.google.bigtable.v2.FeatureFlags;
import com.google.bigtable.v2.Mutation;
import com.google.bigtable.v2.SessionMutateRowsRequest;
import io.grpc.Metadata;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
Expand Down Expand Up @@ -76,6 +77,15 @@ public static boolean isIdempotent(List<Mutation> mutations) {
return true;
}

public static boolean isIdempotent(SessionMutateRowsRequest req) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we remove this method?

on line 192 in tableBase you can do something like:

ImmutableList<Mutation> mutations = request.getEntriesList().stream().flatMap(entry -> entry.getMutationsList().stream()).collect(toImmutableList());
Util.isIdempotent(mutations);

for (SessionMutateRowsRequest.Entry entry : req.getEntriesList()) {
if (!isIdempotent(entry.getMutationsList())) {
return false;
}
}
return true;
}

private static String urlEncode(String s) {
try {
return URLEncoder.encode(s, StandardCharsets.UTF_8.name());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

import com.google.api.gax.rpc.UnaryCallSettings;
import com.google.api.gax.rpc.UnaryCallable;
import com.google.cloud.bigtable.data.v2.models.BulkMutation;
import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
import com.google.cloud.bigtable.data.v2.models.Query;
import com.google.cloud.bigtable.data.v2.models.RowAdapter;
Expand Down Expand Up @@ -46,4 +47,10 @@ public UnaryCallable<ConditionalRowMutation, Boolean> decorateCheckAndMutateRow(
UnaryCallable<ConditionalRowMutation, Boolean> classic, UnaryCallSettings<?, ?> settings) {
return classic;
}

@Override
public UnaryCallable<BulkMutation, Void> decorateMutateRows(
UnaryCallable<BulkMutation, Void> classic, UnaryCallSettings<?, ?> settings) {
return classic;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

import com.google.api.gax.rpc.UnaryCallSettings;
import com.google.api.gax.rpc.UnaryCallable;
import com.google.cloud.bigtable.data.v2.models.BulkMutation;
import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
import com.google.cloud.bigtable.data.v2.models.Query;
import com.google.cloud.bigtable.data.v2.models.RowAdapter;
Expand All @@ -36,4 +37,7 @@ UnaryCallable<RowMutation, Void> decorateMutateRow(

UnaryCallable<ConditionalRowMutation, Boolean> decorateCheckAndMutateRow(
UnaryCallable<ConditionalRowMutation, Boolean> classic, UnaryCallSettings<?, ?> settings);

UnaryCallable<BulkMutation, Void> decorateMutateRows(
UnaryCallable<BulkMutation, Void> classic, UnaryCallSettings<?, ?> settings);
}
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import com.google.cloud.bigtable.data.v2.internal.compat.ops.CheckAndMutateRowShim;
import com.google.cloud.bigtable.data.v2.internal.compat.ops.DivertingUnaryCallable;
import com.google.cloud.bigtable.data.v2.internal.compat.ops.MutateRowShim;
import com.google.cloud.bigtable.data.v2.internal.compat.ops.MutateRowsShim;
import com.google.cloud.bigtable.data.v2.internal.compat.ops.ReadRowShim;
import com.google.cloud.bigtable.data.v2.internal.compat.ops.ReadRowShimInner;
import com.google.cloud.bigtable.data.v2.internal.csm.Metrics;
Expand All @@ -45,6 +46,7 @@
import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer;
import com.google.cloud.bigtable.data.v2.internal.dp.DirectAccessInvestigator;
import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
import com.google.cloud.bigtable.data.v2.models.BulkMutation;
import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
import com.google.cloud.bigtable.data.v2.models.Query;
import com.google.cloud.bigtable.data.v2.models.RowAdapter;
Expand Down Expand Up @@ -89,6 +91,7 @@ public class ShimImpl implements Shim {
private final ReadRowShimInner readRowShimInner;
private final MutateRowShim mutateRowShim;
private final CheckAndMutateRowShim checkAndMutateRowShim;
private final MutateRowsShim mutateRowsShim;

public static Shim create(
ClientInfo clientInfo,
Expand Down Expand Up @@ -205,6 +208,7 @@ public ShimImpl(
this.readRowShimInner = new ReadRowShimInner(client);
this.mutateRowShim = new MutateRowShim(client);
this.checkAndMutateRowShim = new CheckAndMutateRowShim(client);
this.mutateRowsShim = new MutateRowsShim(client);
}

/**
Expand Down Expand Up @@ -390,4 +394,11 @@ public UnaryCallable<ConditionalRowMutation, Boolean> decorateCheckAndMutateRow(
Util.extractTimeout(settings),
debugTagTracer);
}

@Override
public UnaryCallable<BulkMutation, Void> decorateMutateRows(
UnaryCallable<BulkMutation, Void> classic, UnaryCallSettings<?, ?> settings) {
return new DivertingUnaryCallable<>(
configManager, classic, mutateRowsShim, Util.extractTimeout(settings), debugTagTracer);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
/*
* Copyright 2026 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.google.cloud.bigtable.data.v2.internal.compat.ops;

import com.google.bigtable.v2.OpenAuthorizedViewRequest;
import com.google.bigtable.v2.OpenTableRequest.Permission;
import com.google.bigtable.v2.SessionMutateRowsRequest;
import com.google.cloud.bigtable.data.v2.internal.api.AuthorizedViewAsync;
import com.google.cloud.bigtable.data.v2.internal.api.Client;
import com.google.cloud.bigtable.data.v2.internal.api.TableAsync;
import com.google.cloud.bigtable.data.v2.models.AuthorizedViewId;
import com.google.cloud.bigtable.data.v2.models.BulkMutation;
import com.google.cloud.bigtable.data.v2.models.TableId;
import com.google.cloud.bigtable.data.v2.models.TargetId;
import io.grpc.Deadline;
Comment on lines +27 to +28

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Import com.google.common.base.Preconditions to support defensive null-checking of parameters in the constructor.

Suggested change
import com.google.cloud.bigtable.data.v2.models.TargetId;
import io.grpc.Deadline;
import com.google.cloud.bigtable.data.v2.models.TargetId;
import com.google.common.base.Preconditions;
import io.grpc.Deadline;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;

public class MutateRowsShim implements UnaryShim<BulkMutation, Void> {

private final SessionPoolMap<TableId, TableAsync> tables;
private final SessionPoolMap<AuthorizedViewId, AuthorizedViewAsync> authViews;

public MutateRowsShim(Client client) {
tables =
Comment on lines +37 to +38

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Add a defensive null check for the client parameter in the constructor to ensure the object is initialized with a valid reference.

  public MutateRowsShim(Client client) {
    Preconditions.checkNotNull(client, "client is required");
    tables =
References
  1. Defensive null checks in constructors are appropriate to ensure object integrity, provided they are not redundant with upstream validation.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should consider share the pool with MutateRow.

See my PR (although is not approved yet 😂 )
https://github.com/googleapis/google-cloud-java/pull/14585/changes#diff-e47339d2678dd2991014d93d7eaa96d999bb53d535c753e67766f0582b2212f4

new SessionPoolMap<>(
k -> client.openTableAsync(k.getTableId(), Permission.PERMISSION_WRITE));
authViews =
new SessionPoolMap<>(
k ->
client.openAuthorizedViewAsync(
k.getTableId(),
k.getAuthorizedViewId(),
OpenAuthorizedViewRequest.Permission.PERMISSION_WRITE));
}

@Override
public void close() throws IOException {
tables.invalidateAll();
authViews.invalidateAll();
}

@Override
public boolean supports(BulkMutation request) {
// TODO: enable when server side changes is rolled out
return false;
}

@Override
public CompletableFuture<Void> call(BulkMutation request, Deadline deadline) {
TargetId targetId = request.getTargetId();

SessionMutateRowsRequest innerReq = request.toSessionProto();

if (targetId instanceof TableId) {
return tables.apply(
(TableId) targetId, t -> t.mutateRows(innerReq, deadline).thenApply(r -> null));
}
if (targetId instanceof AuthorizedViewId) {
return authViews.apply(
(AuthorizedViewId) targetId, v -> v.mutateRows(innerReq, deadline).thenApply(r -> null));
}

CompletableFuture<Void> f = new CompletableFuture<>();
f.completeExceptionally(
new UnsupportedOperationException("Unsupported targetId type: " + targetId));
return f;
}
}
Loading
Loading