You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "morhidi (via GitHub)" <gi...@apache.org> on 2025/04/18 11:49:51 UTC
[PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
morhidi opened a new pull request, #12839:
URL: https://github.com/apache/iceberg/pull/12839
##Description
The `INCREMENTAL_FROM_LATEST_SNAPSHOT` starting strategy is inclusive in the `IcebergSource`. This means at the moment there is no way to start the `IcebergSource` in a non-empty table without reading the records from at least the latest snapshot. Users can't have the behavior they got used to with using `scan.startup.mode: latest-offset` in Kafka Source.
This change introduces a new `StreamingStartingStrategy` called `INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE` to address that.
Closes: [#9955](https://github.com/apache/iceberg/issues/9955)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu merged PR #12839:
URL: https://github.com/apache/iceberg/pull/12839
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on code in PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#discussion_r2051301525
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -173,50 +190,10 @@ public void testTableScanThenIncrementalWithNonEmptyTable() throws Exception {
}
}
- @Test
+ @TestTemplate
Review Comment:
@rodmeneses asked `@TestTemplate` hence I tried that, I thought this was some sort of convention in Iceberg, changed now to `@ParameterizedTest`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#issuecomment-2815455742
Would you mind taking a quick look at this folks? @pvary @stevenzwu @mxm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on code in PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#discussion_r2051301180
##########
docs/docs/flink-configuration.md:
##########
@@ -102,30 +102,30 @@ env.getConfig()
`Read option` has the highest priority, followed by `Flink configuration` and then `Table property`.
-| Read option | Flink configuration | Table property | Default | Description
|
Review Comment:
The original table used spaces to align cell width, I followed that pattern. Added the change without reformatting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#discussion_r2051255683
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -173,50 +190,10 @@ public void testTableScanThenIncrementalWithNonEmptyTable() throws Exception {
}
}
- @Test
+ @TestTemplate
Review Comment:
why not adding the args directly here?
```
@ParameterizedTest
@EnumSource(value = StreamingStartingStrategy.class, names = { "INCREMENTAL_FROM_LATEST_SNAPSHOT", "INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE" })
```
##########
docs/docs/flink-configuration.md:
##########
@@ -102,30 +102,30 @@ env.getConfig()
`Read option` has the highest priority, followed by `Flink configuration` and then `Table property`.
-| Read option | Flink configuration | Table property | Default | Description
|
Review Comment:
IDE probably did the auto reformatting. can you disable that and only change the relevant cell?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on code in PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#discussion_r2050973783
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -210,6 +210,43 @@ public void testIncrementalFromLatestSnapshotWithEmptyTable() throws Exception {
}
}
+ @Test
+ public void testIncrementalFromLatestSnapshotExclusiveWithEmptyTable() throws Exception {
Review Comment:
I was thinking about that too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#issuecomment-2819094014
I am going to merge this PR so that back port PR can be proceeded. if there are more comments, they can be followed up separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#issuecomment-2815856582
maybe also update the doc on starting strategy
https://iceberg.apache.org/docs/nightly/flink-configuration/#read-options
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#discussion_r2050916538
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java:
##########
@@ -180,6 +180,15 @@ private ContinuousEnumerationResult discoverInitialSplits() {
// For TABLE_SCAN_THEN_INCREMENTAL, incremental mode starts exclusive from the startSnapshot
toPosition =
IcebergEnumeratorPosition.of(startSnapshot.snapshotId(), startSnapshot.timestampMillis());
+ } else if (scanContext.streamingStartingStrategy()
+ == StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE) {
+ splits = Collections.emptyList();
Review Comment:
Could we just initialize splits variable, like
```
List<IcebergSourceSplit> splits = Collections.emptyList()
```
and simplify the code by removing the two assignments?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#discussion_r2050853542
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/StreamingStartingStrategy.java:
##########
@@ -34,6 +34,13 @@ public enum StreamingStartingStrategy {
*/
INCREMENTAL_FROM_LATEST_SNAPSHOT,
+ /**
+ * Start incremental mode from the latest snapshot exclusive.
+ *
+ * <p>If it is an empty map, all future append snapshots should be discovered.
Review Comment:
maybe it is a typo in the Javadoc above too. `empty map` probably should be `empty table`.
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -210,6 +210,43 @@ public void testIncrementalFromLatestSnapshotWithEmptyTable() throws Exception {
}
}
+ @Test
+ public void testIncrementalFromLatestSnapshotExclusiveWithEmptyTable() throws Exception {
+ ScanContext scanContext =
+ ScanContext.builder()
+ .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE)
+ .splitSize(1L)
+ .build();
+ ContinuousSplitPlannerImpl splitPlanner =
+ new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);
+
+ ContinuousEnumerationResult emptyTableInitialDiscoveryResult = splitPlanner.planSplits(null);
+ assertThat(emptyTableInitialDiscoveryResult.splits()).isEmpty();
+ assertThat(emptyTableInitialDiscoveryResult.fromPosition()).isNull();
Review Comment:
for consistency, maybe add this assertion after this line.
```
assertThat(emptyTableInitialDiscoveryResult.fromPosition().snapshotTimestampMs()).isNull();
```
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImplStartStrategy.java:
##########
@@ -88,6 +88,23 @@ public void testTableScanThenIncrementalStrategy() throws IOException {
assertThat(startSnapshot.snapshotId()).isEqualTo(snapshot3.snapshotId());
}
+ @Test
+ public void testForLatestSnapshotStrategyExclusive() throws IOException {
Review Comment:
we should add another test method for non-empty table during start.
I realized this is also missing for the `testForLatestSnapshotStrategy` test below.
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -256,6 +293,46 @@ public void testIncrementalFromLatestSnapshotWithNonEmptyTable() throws Exceptio
}
}
+ @Test
+ public void testIncrementalFromLatestSnapshotExclusiveWithNonEmptyTable() throws Exception {
+ appendTwoSnapshots();
+
+ ScanContext scanContext =
+ ScanContext.builder()
+ .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE)
+ .build();
+ ContinuousSplitPlannerImpl splitPlanner =
+ new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);
+
+ ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
+ assertThat(initialResult.fromPosition()).isNull();
+ // For exclusive behavior, the initial result should point to snapshot2
+
Review Comment:
nit: remove this empty line
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -256,6 +293,46 @@ public void testIncrementalFromLatestSnapshotWithNonEmptyTable() throws Exceptio
}
}
+ @Test
+ public void testIncrementalFromLatestSnapshotExclusiveWithNonEmptyTable() throws Exception {
+ appendTwoSnapshots();
+
+ ScanContext scanContext =
+ ScanContext.builder()
+ .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE)
+ .build();
+ ContinuousSplitPlannerImpl splitPlanner =
+ new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);
+
+ ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
+ assertThat(initialResult.fromPosition()).isNull();
+ // For exclusive behavior, the initial result should point to snapshot2
+
+ assertThat(initialResult.toPosition().snapshotId().longValue())
+ .isEqualTo(snapshot2.snapshotId());
+ assertThat(initialResult.toPosition().snapshotTimestampMs().longValue())
+ .isEqualTo(snapshot2.timestampMillis());
+ assertThat(initialResult.splits()).isEmpty();
Review Comment:
nit: move this assertion to be the first assertion (after line 307)
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -256,6 +293,46 @@ public void testIncrementalFromLatestSnapshotWithNonEmptyTable() throws Exceptio
}
}
+ @Test
+ public void testIncrementalFromLatestSnapshotExclusiveWithNonEmptyTable() throws Exception {
+ appendTwoSnapshots();
+
+ ScanContext scanContext =
+ ScanContext.builder()
+ .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE)
+ .build();
+ ContinuousSplitPlannerImpl splitPlanner =
+ new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);
+
+ ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
+ assertThat(initialResult.fromPosition()).isNull();
+ // For exclusive behavior, the initial result should point to snapshot2
+
+ assertThat(initialResult.toPosition().snapshotId().longValue())
+ .isEqualTo(snapshot2.snapshotId());
+ assertThat(initialResult.toPosition().snapshotTimestampMs().longValue())
+ .isEqualTo(snapshot2.timestampMillis());
+ assertThat(initialResult.splits()).isEmpty();
+
+ // Then the next incremental scan shall discover no files
+ ContinuousEnumerationResult secondResult = splitPlanner.planSplits(initialResult.toPosition());
+ assertThat(secondResult.fromPosition().snapshotId().longValue())
+ .isEqualTo(snapshot2.snapshotId());
+ assertThat(secondResult.fromPosition().snapshotTimestampMs().longValue())
+ .isEqualTo(snapshot2.timestampMillis());
+ assertThat(secondResult.toPosition().snapshotId().longValue())
+ .isEqualTo(snapshot2.snapshotId());
+ assertThat(secondResult.toPosition().snapshotTimestampMs().longValue())
+ .isEqualTo(snapshot2.timestampMillis());
+
+ assertThat(initialResult.splits()).isEmpty();
Review Comment:
similarly, move this assertion after line 318
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE [iceberg]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #12839:
URL: https://github.com/apache/iceberg/pull/12839#discussion_r2050952065
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java:
##########
@@ -210,6 +210,43 @@ public void testIncrementalFromLatestSnapshotWithEmptyTable() throws Exception {
}
}
+ @Test
+ public void testIncrementalFromLatestSnapshotExclusiveWithEmptyTable() throws Exception {
Review Comment:
nit: Looks like this test is identical to testIncrementalFromLatestSnapshotWithEmptyTable, except for the StreamingStartingStrategy. If it is, we can refactor it with @TestTemplate to avoid maintaining the 2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org