Skip to content

Commit 94e680f

Browse files
mutianfmutianf
authored
fix(bigtable): truncate client generated timestamps on the emulator c… (#14234)
…hannel Auto generated cell timestamps now carry true microsecond precision and are tagged as CLIENT_AUTO_GENERATED. The service truncates those to the target table's granularity, so writes to a MILLIS table keep working. The emulator predates timestamp_origin and instead rejects any timestamp that isn't a multiple of 1000, which broke every auto timestamped write. Install a MillisTimestampInterceptor on the emulator channel that performs the truncation the emulator is missing, so it behaves like a MILLIS granularity table. Explicitly specified timestamps are left alone, since a MILLIS table rejects those in production too. This lets the five integration tests that were skipped on the emulator run again.
1 parent 67719c0 commit 94e680f

8 files changed

Lines changed: 315 additions & 27 deletions

File tree

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataSettings.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
2626
import com.google.api.gax.rpc.UnaryCallSettings;
2727
import com.google.auth.Credentials;
28+
import com.google.cloud.bigtable.data.v2.internal.channels.MillisTimestampInterceptor;
2829
import com.google.cloud.bigtable.data.v2.models.Query;
2930
import com.google.cloud.bigtable.data.v2.models.Row;
3031
import com.google.cloud.bigtable.data.v2.stub.BigtableBatchingCallSettings;
@@ -33,7 +34,6 @@
3334
import com.google.cloud.bigtable.data.v2.stub.metrics.NoopMetricsProvider;
3435
import com.google.common.base.MoreObjects;
3536
import com.google.common.base.Strings;
36-
import io.grpc.ManagedChannelBuilder;
3737
import java.io.IOException;
3838
import java.util.List;
3939
import java.util.logging.Logger;
@@ -138,7 +138,13 @@ public static Builder newBuilderForEmulator(String hostname, int port) {
138138
InstantiatingGrpcChannelProvider.newBuilder()
139139
.setMaxInboundMessageSize(256 * 1024 * 1024)
140140
.setChannelPoolSettings(ChannelPoolSettings.staticallySized(1))
141-
.setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
141+
// The emulator only supports millisecond timestamp granularity, so round
142+
// client generated timestamps down the way the service would.
143+
.setChannelConfigurator(
144+
channelBuilder ->
145+
channelBuilder
146+
.usePlaintext()
147+
.intercept(new MillisTimestampInterceptor()))
142148
.setKeepAliveTimeDuration(
143149
java.time.Duration.ofSeconds(61)) // sends ping in this interval
144150
.setKeepAliveTimeoutDuration(

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/MillisTimestampInterceptor.java‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.bigtable.data.v2.internal.channels;
18+
19+
import com.google.bigtable.v2.CheckAndMutateRowRequest;
20+
import com.google.bigtable.v2.MutateRowRequest;
21+
import com.google.bigtable.v2.MutateRowsRequest;
22+
import com.google.bigtable.v2.Mutation;
23+
import com.google.bigtable.v2.Mutation.SetCell;
24+
import com.google.bigtable.v2.Mutation.TimestampOrigin;
25+
import io.grpc.CallOptions;
26+
import io.grpc.Channel;
27+
import io.grpc.ClientCall;
28+
import io.grpc.ClientInterceptor;
29+
import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
30+
import io.grpc.MethodDescriptor;
31+
import java.util.ArrayList;
32+
import java.util.List;
33+
34+
/**
35+
* Rounds client generated cell timestamps down to millisecond granularity.
36+
*
37+
* <p>{@link com.google.cloud.bigtable.data.v2.models.Mutation#setCell(String, String, String)} and
38+
* its siblings that don't take an explicit timestamp stamp the cell with the current time in
39+
* microseconds and tag it as {@link TimestampOrigin#CLIENT_AUTO_GENERATED}. The Bigtable service
40+
* truncates such timestamps to the granularity of the target table, so writing to a table with the
41+
* default {@code MILLIS} granularity keeps working. The Bigtable emulator predates {@code
42+
* timestamp_origin} and instead rejects any timestamp that is not a multiple of 1000, which breaks
43+
* every auto timestamped write.
44+
*
45+
* <p>This interceptor performs the truncation that the emulator is missing, so that it behaves like
46+
* a production table with millisecond granularity. Timestamps that the caller specified explicitly
47+
* are left untouched: those are rejected by a millisecond granularity table in production too.
48+
*/
49+
public class MillisTimestampInterceptor implements ClientInterceptor {
50+
private static final long MICROS_PER_MILLI = 1_000L;
51+
52+
@Override
53+
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
54+
MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
55+
return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
56+
@SuppressWarnings("unchecked")
57+
@Override
58+
public void sendMessage(ReqT message) {
59+
super.sendMessage((ReqT) truncateRequest(message));
60+
}
61+
};
62+
}
63+
64+
private static Object truncateRequest(Object message) {
65+
if (message instanceof MutateRowRequest) {
66+
MutateRowRequest request = (MutateRowRequest) message;
67+
return request.toBuilder()
68+
.clearMutations()
69+
.addAllMutations(truncateAll(request.getMutationsList()))
70+
.build();
71+
} else if (message instanceof MutateRowsRequest) {
72+
MutateRowsRequest request = (MutateRowsRequest) message;
73+
MutateRowsRequest.Builder builder = request.toBuilder().clearEntries();
74+
for (MutateRowsRequest.Entry entry : request.getEntriesList()) {
75+
builder.addEntries(
76+
entry.toBuilder()
77+
.clearMutations()
78+
.addAllMutations(truncateAll(entry.getMutationsList())));
79+
}
80+
return builder.build();
81+
} else if (message instanceof CheckAndMutateRowRequest) {
82+
CheckAndMutateRowRequest request = (CheckAndMutateRowRequest) message;
83+
return request.toBuilder()
84+
.clearTrueMutations()
85+
.addAllTrueMutations(truncateAll(request.getTrueMutationsList()))
86+
.clearFalseMutations()
87+
.addAllFalseMutations(truncateAll(request.getFalseMutationsList()))
88+
.build();
89+
}
90+
91+
return message;
92+
}
93+
94+
private static List<Mutation> truncateAll(List<Mutation> mutations) {
95+
List<Mutation> truncated = new ArrayList<>(mutations.size());
96+
for (Mutation mutation : mutations) {
97+
truncated.add(truncate(mutation));
98+
}
99+
return truncated;
100+
}
101+
102+
/** Returns the mutation with its timestamp truncated, or as is if it didn't need truncating. */
103+
private static Mutation truncate(Mutation mutation) {
104+
if (mutation.getTimestampOrigin() != TimestampOrigin.CLIENT_AUTO_GENERATED
105+
|| !mutation.hasSetCell()) {
106+
return mutation;
107+
}
108+
109+
SetCell setCell = mutation.getSetCell();
110+
long micros = setCell.getTimestampMicros();
111+
long millisAligned = micros - Math.floorMod(micros, MICROS_PER_MILLI);
112+
if (millisAligned == micros) {
113+
return mutation;
114+
}
115+
116+
return mutation.toBuilder()
117+
.setSetCell(setCell.toBuilder().setTimestampMicros(millisAligned))
118+
.build();
119+
}
120+
}

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/channels/MillisTimestampInterceptorTest.java‎

Lines changed: 187 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,187 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.bigtable.data.v2.internal.channels;
18+
19+
import static com.google.common.truth.Truth.assertThat;
20+
21+
import com.google.bigtable.v2.BigtableGrpc;
22+
import com.google.bigtable.v2.CheckAndMutateRowRequest;
23+
import com.google.bigtable.v2.MutateRowRequest;
24+
import com.google.bigtable.v2.MutateRowsRequest;
25+
import com.google.bigtable.v2.Mutation;
26+
import com.google.bigtable.v2.Mutation.DeleteFromRow;
27+
import com.google.bigtable.v2.Mutation.SetCell;
28+
import com.google.bigtable.v2.Mutation.TimestampOrigin;
29+
import com.google.protobuf.ByteString;
30+
import io.grpc.CallOptions;
31+
import io.grpc.Channel;
32+
import io.grpc.ClientCall;
33+
import io.grpc.MethodDescriptor;
34+
import org.junit.Before;
35+
import org.junit.Test;
36+
import org.junit.runner.RunWith;
37+
import org.junit.runners.JUnit4;
38+
39+
@RunWith(JUnit4.class)
40+
public class MillisTimestampInterceptorTest {
41+
private static final long UNALIGNED_MICROS = 1_234_567_890_123_456L;
42+
private static final long ALIGNED_MICROS = 1_234_567_890_123_000L;
43+
44+
private CapturingChannel channel;
45+
private MillisTimestampInterceptor interceptor;
46+
47+
@Before
48+
public void setUp() {
49+
channel = new CapturingChannel();
50+
interceptor = new MillisTimestampInterceptor();
51+
}
52+
53+
@Test
54+
public void mutateRowTruncatesAutoGeneratedTimestamps() {
55+
MutateRowRequest request =
56+
MutateRowRequest.newBuilder()
57+
.addMutations(setCell(UNALIGNED_MICROS, TimestampOrigin.CLIENT_AUTO_GENERATED))
58+
.build();
59+
60+
MutateRowRequest sent = send(BigtableGrpc.getMutateRowMethod(), request);
61+
62+
assertThat(sent.getMutations(0).getSetCell().getTimestampMicros()).isEqualTo(ALIGNED_MICROS);
63+
}
64+
65+
@Test
66+
public void mutateRowLeavesUserSpecifiedTimestampsAlone() {
67+
MutateRowRequest request =
68+
MutateRowRequest.newBuilder()
69+
.addMutations(setCell(UNALIGNED_MICROS, TimestampOrigin.USER_SPECIFIED))
70+
.build();
71+
72+
assertThat(send(BigtableGrpc.getMutateRowMethod(), request)).isEqualTo(request);
73+
}
74+
75+
@Test
76+
public void mutateRowLeavesNonSetCellMutationsAlone() {
77+
MutateRowRequest request =
78+
MutateRowRequest.newBuilder()
79+
.addMutations(
80+
Mutation.newBuilder()
81+
.setDeleteFromRow(DeleteFromRow.getDefaultInstance())
82+
.setTimestampOrigin(TimestampOrigin.CLIENT_AUTO_GENERATED))
83+
.build();
84+
85+
assertThat(send(BigtableGrpc.getMutateRowMethod(), request)).isEqualTo(request);
86+
}
87+
88+
@Test
89+
public void mutateRowsTruncatesEveryEntry() {
90+
MutateRowsRequest request =
91+
MutateRowsRequest.newBuilder()
92+
.addEntries(
93+
MutateRowsRequest.Entry.newBuilder()
94+
.setRowKey(ByteString.copyFromUtf8("row1"))
95+
.addMutations(setCell(ALIGNED_MICROS, TimestampOrigin.CLIENT_AUTO_GENERATED)))
96+
.addEntries(
97+
MutateRowsRequest.Entry.newBuilder()
98+
.setRowKey(ByteString.copyFromUtf8("row2"))
99+
.addMutations(setCell(UNALIGNED_MICROS, TimestampOrigin.CLIENT_AUTO_GENERATED))
100+
.addMutations(setCell(UNALIGNED_MICROS, TimestampOrigin.USER_SPECIFIED)))
101+
.build();
102+
103+
MutateRowsRequest sent = send(BigtableGrpc.getMutateRowsMethod(), request);
104+
105+
assertThat(sent.getEntries(0)).isEqualTo(request.getEntries(0));
106+
assertThat(sent.getEntries(1).getRowKey()).isEqualTo(ByteString.copyFromUtf8("row2"));
107+
assertThat(sent.getEntries(1).getMutations(0).getSetCell().getTimestampMicros())
108+
.isEqualTo(ALIGNED_MICROS);
109+
assertThat(sent.getEntries(1).getMutations(1).getSetCell().getTimestampMicros())
110+
.isEqualTo(UNALIGNED_MICROS);
111+
}
112+
113+
@Test
114+
public void checkAndMutateRowTruncatesBothBranches() {
115+
CheckAndMutateRowRequest request =
116+
CheckAndMutateRowRequest.newBuilder()
117+
.addTrueMutations(setCell(UNALIGNED_MICROS, TimestampOrigin.CLIENT_AUTO_GENERATED))
118+
.addFalseMutations(setCell(UNALIGNED_MICROS, TimestampOrigin.CLIENT_AUTO_GENERATED))
119+
.build();
120+
121+
CheckAndMutateRowRequest sent = send(BigtableGrpc.getCheckAndMutateRowMethod(), request);
122+
123+
assertThat(sent.getTrueMutations(0).getSetCell().getTimestampMicros())
124+
.isEqualTo(ALIGNED_MICROS);
125+
assertThat(sent.getFalseMutations(0).getSetCell().getTimestampMicros())
126+
.isEqualTo(ALIGNED_MICROS);
127+
}
128+
129+
@Test
130+
public void otherRequestsPassThrough() {
131+
com.google.bigtable.v2.ReadRowsRequest request =
132+
com.google.bigtable.v2.ReadRowsRequest.newBuilder().setTableName("my-table").build();
133+
134+
assertThat(send(BigtableGrpc.getReadRowsMethod(), request)).isEqualTo(request);
135+
}
136+
137+
private static Mutation setCell(long timestampMicros, TimestampOrigin timestampOrigin) {
138+
return Mutation.newBuilder()
139+
.setSetCell(
140+
SetCell.newBuilder()
141+
.setFamilyName("cf")
142+
.setColumnQualifier(ByteString.copyFromUtf8("q"))
143+
.setTimestampMicros(timestampMicros)
144+
.setValue(ByteString.copyFromUtf8("v")))
145+
.setTimestampOrigin(timestampOrigin)
146+
.build();
147+
}
148+
149+
@SuppressWarnings("unchecked")
150+
private <ReqT, RespT> ReqT send(MethodDescriptor<ReqT, RespT> method, ReqT request) {
151+
interceptor.interceptCall(method, CallOptions.DEFAULT, channel).sendMessage(request);
152+
return (ReqT) channel.lastMessage;
153+
}
154+
155+
/** A channel whose calls do nothing except record the last message that was sent. */
156+
private static class CapturingChannel extends Channel {
157+
private Object lastMessage;
158+
159+
@Override
160+
public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
161+
MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
162+
return new ClientCall<ReqT, RespT>() {
163+
@Override
164+
public void start(Listener<RespT> responseListener, io.grpc.Metadata headers) {}
165+
166+
@Override
167+
public void request(int numMessages) {}
168+
169+
@Override
170+
public void cancel(String message, Throwable cause) {}
171+
172+
@Override
173+
public void halfClose() {}
174+
175+
@Override
176+
public void sendMessage(ReqT message) {
177+
lastMessage = message;
178+
}
179+
};
180+
}
181+
182+
@Override
183+
public String authority() {
184+
return "fake-authority";
185+
}
186+
}
187+
}

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/BulkMutateIT.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -53,11 +53,6 @@ public class BulkMutateIT {
5353

5454
@Test(timeout = 60 * 1000)
5555
public void test() throws IOException, InterruptedException {
56-
assume()
57-
.withMessage("Emulator does not support microsecond timestamp granularity")
58-
.that(testEnvRule.env())
59-
.isNotInstanceOf(EmulatorEnv.class);
60-
6156
BigtableDataSettings settings = testEnvRule.env().getDataClientSettings();
6257
String rowPrefix = UUID.randomUUID().toString();
6358
// Set target latency really low so it'll trigger adjusting thresholds

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/CheckAndMutateIT.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -49,11 +49,6 @@ public class CheckAndMutateIT {
4949

5050
@Test
5151
public void test() throws Exception {
52-
assume()
53-
.withMessage("Emulator does not support microsecond timestamp granularity")
54-
.that(testEnvRule.env())
55-
.isNotInstanceOf(EmulatorEnv.class);
56-
5752
TableId tableId = testEnvRule.env().getTableId();
5853
String familyId = testEnvRule.env().getFamilyId();
5954
String rowKey = UUID.randomUUID().toString();

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/MutateRowIT.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -44,11 +44,6 @@ public class MutateRowIT {
4444

4545
@Test
4646
public void test() throws Exception {
47-
assume()
48-
.withMessage("Emulator does not support microsecond timestamp granularity")
49-
.that(testEnvRule.env())
50-
.isNotInstanceOf(EmulatorEnv.class);
51-
5247
String rowKey = UUID.randomUUID().toString();
5348
String familyId = testEnvRule.env().getFamilyId();
5449

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/ReadIT.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -92,11 +92,6 @@ public void setUp() {
9292

9393
@Test
9494
public void isRowExists() throws Exception {
95-
assume()
96-
.withMessage("Emulator does not support microsecond timestamp granularity")
97-
.that(testEnvRule.env())
98-
.isNotInstanceOf(EmulatorEnv.class);
99-
10095
String rowKey = prefix + "-test-row-key";
10196
TableId tableId = testEnvRule.env().getTableId();
10297
testEnvRule

‎java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/SampleRowsIT.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -50,11 +50,6 @@ public class SampleRowsIT {
5050

5151
@Test
5252
public void test() throws InterruptedException, ExecutionException, TimeoutException {
53-
assume()
54-
.withMessage("Emulator does not support microsecond timestamp granularity")
55-
.that(testEnvRule.env())
56-
.isNotInstanceOf(EmulatorEnv.class);
57-
5853
BigtableDataClient client = testEnvRule.env().getDataClient();
5954
String rowPrefix = UUID.randomUUID().toString();
6055

0 commit comments

Comments
 (0)