This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 630af04bc7 [Improve][Connector-V2] Add row-level resume for Bigtable
source chec… (#11359)
630af04bc7 is described below
commit 630af04bc7e392d90e621a9225672386bba6c46a
Author: 雷炯 <[email protected]>
AuthorDate: Wed Jul 8 22:44:02 2026 +0800
[Improve][Connector-V2] Add row-level resume for Bigtable source chec…
(#11359)
Co-authored-by: Cursor <[email protected]>
---
.../bigtable/source/BigtableSourceReader.java | 6 +-
.../bigtable/source/BigtableSourceSplit.java | 55 +++++++++++++--
.../bigtable/source/BigtableSourceReaderTest.java | 71 +++++++++++++++++++
.../bigtable/source/BigtableSourceSplitTest.java | 81 ++++++++++++++++++++++
4 files changed, 208 insertions(+), 5 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReader.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReader.java
index 17e41107aa..3dcfa310b2 100644
---
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReader.java
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReader.java
@@ -137,6 +137,9 @@ public class BigtableSourceReader implements
SourceReader<SeaTunnelRow, Bigtable
SeaTunnelRow seaTunnelRow =
convertRow(bigtableRow);
synchronized (output.getCheckpointLock()) {
output.collect(seaTunnelRow);
+ // Update progress under the same lock as
collect for consistent
+ // snapshots
+
split.setLastReadRowKey(bigtableRow.getKey().toStringUtf8());
}
});
}
@@ -144,7 +147,8 @@ public class BigtableSourceReader implements
SourceReader<SeaTunnelRow, Bigtable
private Query buildQuery(BigtableSourceSplit split) {
Query query = Query.create(parameters.getTable());
- String startKey = split.getStartRowKey();
+ // Resume from lastReadRowKey after checkpoint restore when present
+ String startKey = split.getResumeStartRowKey();
String endKey = split.getEndRowKey();
if (!startKey.isEmpty() && !endKey.isEmpty()) {
query.range(startKey, endKey);
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplit.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplit.java
index 99e6782960..5276dde38c 100644
---
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplit.java
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplit.java
@@ -21,21 +21,42 @@ import org.apache.seatunnel.api.source.SourceSplit;
import java.io.Serializable;
+/**
+ * Describes a bounded Bigtable scan split and optional in-split resume
progress.
+ *
+ * <p>When {@link #lastReadRowKey} is present, checkpoint restore resumes from
that row key
+ * (inclusive) instead of re-scanning from {@link #startRowKey}. Semantics are
at-least-once: the
+ * resume row may be read again after failover.
+ */
public class BigtableSourceSplit implements SourceSplit, Serializable {
+ /** Kept at {@code 1L} so existing checkpoint/savepoint bytes remain
deserializable. */
private static final long serialVersionUID = 1L;
+
public static final String SPLIT_PREFIX = "bigtable_source_split_";
private final String splitId;
- /** Inclusive start row key (empty means the table start). */
+ /** Inclusive start row key (empty means table start). */
private final String startRowKey;
- /** Exclusive end row key (empty means the table end). */
+ /** Exclusive end row key (empty means table end). */
private final String endRowKey;
+ /**
+ * Last successfully emitted row key (UTF-8), persisted in checkpoint
state. {@code null} for
+ * legacy checkpoints written before row-level resume was added.
+ */
+ private volatile String lastReadRowKey;
+
public BigtableSourceSplit(int splitIndex, String startRowKey, String
endRowKey) {
+ this(splitIndex, startRowKey, endRowKey, null);
+ }
+
+ public BigtableSourceSplit(
+ int splitIndex, String startRowKey, String endRowKey, String
lastReadRowKey) {
this.splitId = SPLIT_PREFIX + splitIndex;
this.startRowKey = startRowKey;
this.endRowKey = endRowKey;
+ this.lastReadRowKey = lastReadRowKey;
}
@Override
@@ -51,10 +72,36 @@ public class BigtableSourceSplit implements SourceSplit,
Serializable {
return endRowKey;
}
+ public String getLastReadRowKey() {
+ return lastReadRowKey;
+ }
+
+ /**
+ * Updates in-split progress; called under the same lock as {@code
collect()} so snapshots see
+ * the last emitted row.
+ *
+ * @param rowKeyUtf8 current row key (UTF-8)
+ */
+ void setLastReadRowKey(String rowKeyUtf8) {
+ this.lastReadRowKey = rowKeyUtf8;
+ }
+
+ /**
+ * Row key used as the inclusive scan start when building a Bigtable
{@code Query}.
+ *
+ * <p>Returns {@link #lastReadRowKey} when set, otherwise {@link
#startRowKey}.
+ */
+ public String getResumeStartRowKey() {
+ if (lastReadRowKey != null && !lastReadRowKey.isEmpty()) {
+ return lastReadRowKey;
+ }
+ return startRowKey;
+ }
+
@Override
public String toString() {
return String.format(
- "{\"split_id\":\"%s\", \"start\":\"%s\", \"end\":\"%s\"}",
- splitId, startRowKey, endRowKey);
+ "{\"split_id\":\"%s\", \"start\":\"%s\", \"end\":\"%s\",
\"last_read\":\"%s\"}",
+ splitId, startRowKey, endRowKey, lastReadRowKey);
}
}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReaderTest.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReaderTest.java
index d1e8852028..e331ec593b 100644
---
a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReaderTest.java
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReaderTest.java
@@ -137,6 +137,77 @@ class BigtableSourceReaderTest {
"snapshotState taken during readSplit() must include the
in-flight split");
}
+ /**
+ * Checkpoint taken mid-scan must persist the last emitted rowkey so
failover can resume inside
+ * the split instead of re-scanning from split start.
+ */
+ @SuppressWarnings("unchecked")
+ @Test
+ void testSnapshotStateCapturesLastReadRowKeyDuringScan() throws Exception {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "a", "z");
+ final List<BigtableSourceSplit>[] capturedState = new List[1];
+
+ Row fakeRow = mock(Row.class);
+ RowCell cell = mock(RowCell.class);
+ when(cell.getFamily()).thenReturn("cf");
+ when(cell.getQualifier()).thenReturn(ByteString.copyFromUtf8("name"));
+ when(cell.getValue()).thenReturn(ByteString.copyFromUtf8("bob"));
+ when(fakeRow.getCells()).thenReturn(Collections.singletonList(cell));
+ when(fakeRow.getKey()).thenReturn(ByteString.copyFromUtf8("row-490"));
+
+ BigtableSourceReader reader = createOpenedReader(parameters, rowType);
+ reader.addSplits(Collections.singletonList(split));
+
+ ServerStream<Row> fakeStream = mock(ServerStream.class);
+ Mockito.doAnswer(
+ invocation -> {
+ Consumer<Row> action = invocation.getArgument(0);
+ action.accept(fakeRow);
+ capturedState[0] = reader.snapshotState(99L);
+ return null;
+ })
+ .when(fakeStream)
+ .forEach(any());
+ when(mockDataClient.readRows(any(Query.class))).thenReturn(fakeStream);
+
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ when(collector.getCheckpointLock()).thenReturn(new Object());
+
+ reader.pollNext(collector);
+
+ assertEquals(1, capturedState[0].size());
+ assertEquals("row-490", capturedState[0].get(0).getLastReadRowKey());
+ assertEquals("row-490",
capturedState[0].get(0).getResumeStartRowKey());
+ }
+
+ /**
+ * A split restored from checkpoint with {@code lastReadRowKey} must use
that key as the scan
+ * start when {@link BigtableSourceReader} issues the read request.
+ */
+ @SuppressWarnings("unchecked")
+ @Test
+ void testRestoredSplitResumesFromLastReadRowKey() throws Exception {
+ BigtableSourceSplit restoredSplit = new BigtableSourceSplit(0, "a",
"z", "row-490");
+
+ ServerStream<Row> fakeStream = mock(ServerStream.class);
+ Mockito.doAnswer(invocation -> null).when(fakeStream).forEach(any());
+ ArgumentCaptor<Query> queryCaptor =
ArgumentCaptor.forClass(Query.class);
+
when(mockDataClient.readRows(queryCaptor.capture())).thenReturn(fakeStream);
+
+ BigtableSourceReader reader = createOpenedReader(parameters, rowType);
+ reader.addSplits(Collections.singletonList(restoredSplit));
+
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ when(collector.getCheckpointLock()).thenReturn(new Object());
+
+ reader.pollNext(collector);
+
+ assertEquals("row-490", restoredSplit.getResumeStartRowKey());
+ assertTrue(
+ queryCaptor.getValue().toString().contains("row-490"),
+ "Query should scan from restored lastReadRowKey");
+ }
+
/**
* After readSplit() completes, currentSplit is cleared. A snapshot taken
after that must NOT
* re-include the already-finished split.
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitTest.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitTest.java
new file mode 100644
index 0000000000..3c911c62c6
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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 org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.apache.seatunnel.common.utils.SerializationUtils;
+
+import org.apache.commons.codec.binary.Base64;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/** Tests for {@link BigtableSourceSplit} resume progress and checkpoint
serialization. */
+class BigtableSourceSplitTest {
+
+ /**
+ * Bytes produced by {@code BigtableSourceSplit(1, "x", "y")} on dev
before {@code
+ * lastReadRowKey} existed (serialVersionUID = 1L, three fields only).
+ */
+ private static final String LEGACY_SPLIT_V1_BASE64 =
+
"rO0ABXNyAE1vcmcuYXBhY2hlLnNlYXR1bm5lbC5jb25uZWN0b3JzLnNlYXR1bm5lbC5iaWd0YWJsZS5zb3VyY2UuQmlndGFibGVTb3VyY2VTcGxpdAAAAAAAAAABAgADTAAJZW5kUm93S2V5dAASTGphdmEvbGFuZy9TdHJpbmc7TAAHc3BsaXRJZHEAfgABTAALc3RhcnRSb3dLZXlxAH4AAXhwdAABeXQAF2JpZ3RhYmxlX3NvdXJjZV9zcGxpdF8xdAABeA==";
+
+ @Test
+ void resumeStartRowKeyFallsBackToSplitStartWhenNoProgress() {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "a", "z");
+ assertEquals("a", split.getResumeStartRowKey());
+ }
+
+ @Test
+ void resumeStartRowKeyUsesLastReadRowKeyWhenPresent() {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "a", "z",
"row-490");
+ assertEquals("row-490", split.getResumeStartRowKey());
+ }
+
+ @Test
+ void lastReadRowKeyCanBeUpdatedDuringRead() {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "", "");
+ split.setLastReadRowKey("key-42");
+ assertEquals("key-42", split.getLastReadRowKey());
+ assertEquals("key-42", split.getResumeStartRowKey());
+ }
+
+ @Test
+ void checkpointRoundTripPreservesLastReadRowKey() {
+ BigtableSourceSplit original = new BigtableSourceSplit(0, "start",
"end", "resume-here");
+ byte[] bytes = SerializationUtils.serialize(original);
+ BigtableSourceSplit restored = SerializationUtils.deserialize(bytes);
+ assertEquals(original.splitId(), restored.splitId());
+ assertEquals("start", restored.getStartRowKey());
+ assertEquals("end", restored.getEndRowKey());
+ assertEquals("resume-here", restored.getLastReadRowKey());
+ assertEquals("resume-here", restored.getResumeStartRowKey());
+ }
+
+ @Test
+ void deserializesLegacyCheckpointBytesFromDevRelease() {
+ byte[] legacyBytes = Base64.decodeBase64(LEGACY_SPLIT_V1_BASE64);
+ BigtableSourceSplit restored =
SerializationUtils.deserialize(legacyBytes);
+ assertEquals("bigtable_source_split_1", restored.splitId());
+ assertEquals("x", restored.getStartRowKey());
+ assertEquals("y", restored.getEndRowKey());
+ assertNull(restored.getLastReadRowKey());
+ assertEquals("x", restored.getResumeStartRowKey());
+ }
+}
