Flink: FLIP-27 source enumerator help classes by stevenzwu · Pull Request #4329 · apache/iceberg

stevenzwu · 2022-03-15T05:10:26Z

It mainly contains classes for streaming read (continuous split discovery) along with config and enumeration position classes.

stevenzwu · 2022-03-15T18:38:24Z

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java

 */
-class ScanContext implements Serializable {
+@Internal
+public class ScanContext implements Serializable {


This is made public because it is accessed by ContinuousSplitPlannerImpl class in this PR.

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java

yittg

Thanks @stevenzwu, i have some question about this change, hoping you can help

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java

yittg · 2022-03-16T06:25:03Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+  private void validate() {
+    Preconditions.checkArgument(scanContext.snapshotId() == null,
+        "Can't set snapshotId in ScanContext for continuous enumerator");
+    Preconditions.checkArgument(scanContext.asOfTimestamp() == null,
+        "Can't set asOfTimestamp in ScanContext for continuous enumerator");
+    Preconditions.checkArgument(scanContext.startSnapshotId() == null,
+        "Can't set startSnapshotId in ScanContext for continuous enumerator");
+    Preconditions.checkArgument(scanContext.endSnapshotId() == null,
+        "Can't set endSnapshotId in ScanContext for continuous enumerator");


Does that mean scanContext#monitorInterval should be null, because it is configured in IcebergEnumeratorConfig?

that is a good question. I am also pondering maybe we should merge the IcebergEnumeratorConfig with ScanContext. As you pointed out, there are some overlapped configs, like monitorInterval, startSnapshotId. Please let me know your preference. Will also get some input from Ryan.

Right now, I am leaning toward merging them.

merged the IcebergEnumeratorConfig into ScanContext.

yittg · 2022-03-16T06:29:14Z

.../flink/src/main/java/org/apache/iceberg/flink/source/enumerator/IcebergEnumeratorConfig.java

+    public Builder startingStrategy(StartingStrategy strategy) {
+      this.startingStrategy = strategy;
+      return this;
+    }


Would it be better to use named strategy method accepting required options? like

Suggested change

public Builder startingStrategy(StartingStrategy strategy) {

this.startingStrategy = strategy;

return this;

}

public Builder startingWithSpecificSnapshot(long startId) {

this.startingStrategy = StartingStrategy.SPECIFIC_START_SNAPSHOT_ID;

this.startSnapshotId = startId;

return this;

}

agree, I like your suggestion.

actually while working on merging the configs to ScanContext, I think we need to stay in the simple POJO style for ScanContext.Builder.fromProperties

yittg · 2022-03-16T06:49:43Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+  }
+
+  @VisibleForTesting
+  static HistoryEntry getStartSnapshot(Table table, IcebergEnumeratorConfig enumeratorConfig) {


it's a little confused whether the start snapshot is counted or not? it is exclusive based on TableScan#appendsBetween. However, it does not match the intuition for strategy EARLIEST_SNAPSHOT, even SPECIFIC_START_SNAPSHOT_ID and SPECIFIC_START_SNAPSHOT_TIMESTAMP

start snapshot is exclusive. I am thinking that we can document the behavior better for StartingStrategy enum class.

IIUC, these strategies mean semantic with AFTER?

EARLIEST_SNAPSHOT actually means START_AFTER_ EARLIEST_SNAPSHOT;
SPECIFIC_START_SNAPSHOT_ID actually means START_AFTER_SNAPSHOT_ID;
SPECIFIC_START_SNAPSHOT_TIMESTAMP actually means START_AFTER_SNAPSHOT_TIMESTAMP;

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

rdblue · 2022-03-18T16:39:41Z

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java

+    if (isStreaming) {
+      if (startingStrategy == StreamingStartingStrategy.SPECIFIC_START_SNAPSHOT_ID) {
+        Preconditions.checkArgument(startSnapshotId != null,
+            "startSnapshotId cannot be null for SPECIFIC_START_SNAPSHOT_ID starting strategy");


We probably want to remove the specific variable names from the error messages. And we also phrase error messages more directly: (Problem): (context)

This should probably be: Invalid starting snapshot for SPECIFIC_START_SNAPSHOT_ID strategy: null

will update

rdblue · 2022-03-18T16:41:57Z

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/StreamingStartingStrategy.java

+   * Start incremental mode from a specific startTimestamp.
+   * Starting snapshot has a timestamp lower than or equal to the specified timestamp.
+   */
+  SPECIFIC_START_SNAPSHOT_TIMESTAMP


What happens if table history doesn't go back that far? It should probably fail because the user's request can't be satisfied.

switched to SnapshotUtil#snapshotIdAsOfTime, which handles this scenario internally (throwing exception)

INCREMENTAL_AFTER_TIMESTAMP?

rdblue · 2022-03-18T16:45:39Z

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/StreamingStartingStrategy.java

+  /**
+   * Start incremental mode from a specific startSnapshotId
+   */
+  SPECIFIC_START_SNAPSHOT_ID,


Inclusive of the changes in this snapshot?

mentioned the exclusive behave at the class-level Javadoc

rdblue · 2022-03-18T16:51:26Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+    this.scanContext = scanContext;
+    // Within a JVM, table name should be unique across sources.
+    // Hence it is used as worker pool thread name prefix.
+    this.workerPool = ThreadPools.newWorkerPool("iceberg-worker-pool-" + table.name(), scanContext.planParallelism());


Isn't it possible to process the table twice in the same JVM? I agree operator ID would be better.

make the thread pool name a constructor argument.

Because FLIP-27 source interface doesn't expose the operator ID, for now tableName-UUID is used to guarantee uniqueness. comments will be updated.

rdblue · 2022-03-18T16:56:07Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+  public ContinuousEnumerationResult planSplits(IcebergEnumeratorPosition lastPosition) {
+    table.refresh();
+    if (lastPosition != null) {
+      return discoverDeltaSplits(lastPosition);


I would avoid using "Delta" because that has multiple meanings. Incremental is better.

will update

rdblue · 2022-03-18T16:59:18Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+  }
+
+  @VisibleForTesting
+  static HistoryEntry getStartSnapshot(Table table, ScanContext scanContext) {


Here you probably want to use SnapshotUtil, which handles most of these cases using the current table state's ancestors.

will switch

rdblue · 2022-03-18T17:00:37Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+              "Snapshot id not found in history: {}" + scanContext.startSnapshotId());
+        }
+        break;
+      case SPECIFIC_START_SNAPSHOT_TIMESTAMP:


Probably use https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java#L112

will switch

rdblue · 2022-03-18T17:00:49Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+      case LATEST_SNAPSHOT:
+        startEntry = historyEntries.get(historyEntries.size() - 1);
+        break;
+      case EARLIEST_SNAPSHOT:


Use https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java#L89

will switch

rdblue · 2022-03-18T17:02:22Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+    switch (scanContext.startingStrategy()) {
+      case TABLE_SCAN_THEN_INCREMENTAL:
+      case LATEST_SNAPSHOT:
+        startEntry = historyEntries.get(historyEntries.size() - 1);


table.currentSnapshot()?

will switch

rdblue · 2022-03-18T17:02:35Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+        startEntry = historyEntries.get(0);
+        break;
+      case SPECIFIC_START_SNAPSHOT_ID:
+        Optional<HistoryEntry> matchedEntry = historyEntries.stream()


table.snapshot(snapshotId)

will switch

rdblue · 2022-03-18T17:03:16Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+   */
+  private ContinuousEnumerationResult discoverInitialSplits() {
+    HistoryEntry startSnapshotEntry = getStartSnapshot(table, scanContext);
+    LOG.info("get startSnapshotId {} based on starting strategy {}",


Clean up log message?

will update

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

kbendick · 2022-03-21T23:33:48Z

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java

+      }
+      if (startingStrategy == StreamingStartingStrategy.SPECIFIC_START_SNAPSHOT_TIMESTAMP) {
+        Preconditions.checkArgument(startSnapshotTimestamp != null,
+            "startSnapshotTimestamp cannot be null for SPECIFIC_START_SNAPSHOT_TIMESTAMP starting strategy");


Should we validate that startSnapshotId isn't supplied if the user has set the startingStrategy to be by timestamp.

As well as making sure there's not a provided startSnapshotTimestamp when using snapshot id starting strategy?

sure. will add them

kbendick · 2022-03-21T23:41:43Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+    this.table = table;
+    this.scanContext = scanContext;
+    this.workerPool = ThreadPools.newWorkerPool(
+        "iceberg-enumerator-pool-" + threadPoolName, scanContext.planParallelism());


Nit: Would it be more informative to mention split-planner-pool-** or something that correlates more to the class name?

if we use the current naming convention in FlinkInputFormat, we can probably change it to iceberg-plan-worker-pool-<threadPoolName>.

yittg · 2022-03-24T03:18:13Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+  }
+
+  @VisibleForTesting
+  static Snapshot getStartSnapshot(Table table, ScanContext scanContext) {


How would you deal with an empty table, i.e. no snapshots are produced?

Guess this method should return an Optional<Snapshot> ?

good point. it is a corner case. will switch to Optional and also add a unit test

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

rdblue · 2022-06-06T16:09:47Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+        if (matchedSnapshotById != null) {
+          return Optional.of(matchedSnapshotById);
+        } else {
+          throw new IllegalArgumentException(


We usually prefer Preconditions.checkArgument instead of an extra if statement.

rdblue · 2022-06-06T16:10:08Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+      case INCREMENTAL_FROM_SNAPSHOT_TIMESTAMP:
+        long snapshotIdAsOfTime = SnapshotUtil.snapshotIdAsOfTime(table, scanContext.startSnapshotTimestamp());
+        Snapshot matchedSnapshotByTimestamp = table.snapshot(snapshotIdAsOfTime);
+        if (matchedSnapshotByTimestamp != null) {


If we are guaranteed that the snapshot from snapshotIdAsOfTime is known, then we can avoid this check.

rdblue · 2022-06-06T16:14:33Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+      LOG.info("Skip incremental scan because table is empty");
+      return new ContinuousEnumerationResult(Collections.emptyList(), lastPosition, lastPosition);
+    } else {
+      if (lastPosition.snapshotId() != null && currentSnapshot.snapshotId() == lastPosition.snapshotId()) {


Nit: this could be an else if

rdblue · 2022-06-06T16:24:16Z

.../org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImplStartStrategy.java

+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestRule;
+
+public class TestContinuousSplitPlannerImplStartStrategy {


Looks good!

rdblue · 2022-06-06T16:26:41Z

...src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java

+    Assert.assertEquals(1, result.splits().size());
+    IcebergSourceSplit split = result.splits().iterator().next();
+    Assert.assertEquals(1, split.task().files().size());
+    Assert.assertEquals(dataFile.path().toString(), split.task().files().iterator().next().file().path().toString());


Minor: you can use Iterables.getOnlyElement to avoid calling iterator().next() without validation.

rdblue · 2022-06-06T16:27:49Z

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

+  public ContinuousSplitPlannerImpl(Table table, ScanContext scanContext, String threadPoolName) {
+    this.table = table;
+    this.scanContext = scanContext;
+    this.workerPool = ThreadPools.newWorkerPool(


Minor: for testing, could we set this to null so that we don't spawn a worker pool?

Also, the pool is never closed. Should this be closeable? If not, should we pass in the worker pool so that its lifecycle is attached to a closeable instance?

rdblue · 2022-06-06T16:29:30Z

...src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java

+    ContinuousEnumerationResult emptyTableInitialDiscoveryResult = splitPlanner.planSplits(null);
+    Assert.assertTrue(emptyTableInitialDiscoveryResult.splits().isEmpty());
+    Assert.assertNull(emptyTableInitialDiscoveryResult.fromPosition());
+    Assert.assertNull(emptyTableInitialDiscoveryResult.toPosition().snapshotId());


We may want to add an isEmpty method to check this rather than checking snapshotId() is null.

… also added unit test coverage for the scenario

…Impl that closes private thread pool. address other comments from Ryan too.

rdblue · 2022-06-06T22:53:48Z

Thanks, @stevenzwu!

…ache#4979)

github-actions bot added the flink label Mar 15, 2022

stevenzwu force-pushed the SplitEnumerator branch from fc10e58 to 9c1001d Compare March 15, 2022 16:20

stevenzwu commented Mar 15, 2022

View reviewed changes

yittg reviewed Mar 16, 2022

View reviewed changes

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java Show resolved Hide resolved

yittg mentioned this pull request Mar 16, 2022

Flink: fix missing copy in ScanContext #4341

Merged

yittg reviewed Mar 16, 2022

View reviewed changes

stevenzwu force-pushed the SplitEnumerator branch 3 times, most recently from c70e105 to 651affb Compare March 17, 2022 16:06

rdblue reviewed Mar 18, 2022

View reviewed changes

...ink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java Show resolved Hide resolved

kbendick reviewed Mar 21, 2022

View reviewed changes

stevenzwu force-pushed the SplitEnumerator branch from 651affb to 8112ddf Compare March 21, 2022 23:37

kbendick reviewed Mar 21, 2022

View reviewed changes

yittg reviewed Mar 24, 2022

View reviewed changes

stevenzwu force-pushed the SplitEnumerator branch from 4dcb0e2 to 54baa78 Compare March 24, 2022 05:03

stevenzwu closed this Mar 24, 2022

stevenzwu reopened this Mar 24, 2022

stevenzwu force-pushed the SplitEnumerator branch from 0b9a4ae to 3fc48b3 Compare March 24, 2022 20:03

stevenzwu closed this Mar 24, 2022

rdblue reviewed Jun 6, 2022

View reviewed changes

rdblue approved these changes Jun 6, 2022

View reviewed changes

github-actions bot added the build label Jun 6, 2022

stevenzwu added 10 commits June 6, 2022 11:24

Flink: FLIP-27 source enumerator help classes

ca36ffd

Address Yi's review comments

f4c14ce

address review comments

2503335

address Kyle's comments

2ba3f2f

address the edge case of empty table scenario that yittg pointed out.…

b5d1656

… also added unit test coverage for the scenario

Change starting strategy to be inclusive for the starting snapshot

3542b82

clean up comments

e74b25f

update FlinkSplitPlanner to use the new IncrementalAppendScan interface

14bad66

Address Ryan's comments

1acebcb

make ContinuousSplitPlanner Closeable. updated ContinuousSplitPlanner…

3ed15ce

…Impl that closes private thread pool. address other comments from Ryan too.

stevenzwu force-pushed the SplitEnumerator branch from 3b79198 to 3ed15ce Compare June 6, 2022 18:24

fix checkstyleTest

ac7086e

rdblue approved these changes Jun 6, 2022

View reviewed changes

rdblue merged commit 31dafee into apache:master Jun 6, 2022

stevenzwu mentioned this pull request Jun 6, 2022

Flink: port PR 4329 (FLIP-27 enumerator) to Flink 1.15 #4979

Merged

rdblue pushed a commit that referenced this pull request Jun 7, 2022

Flink 1.15: Port PR #4329 to add FLIP-27 enumerator classes (#4979)

5d6c6cc

namrathamyske pushed a commit to namrathamyske/iceberg that referenced this pull request Jul 10, 2022

Flink: FLIP-27 source enumerator help classes (apache#4329)

a3eb890

namrathamyske pushed a commit to namrathamyske/iceberg that referenced this pull request Jul 10, 2022

Flink 1.15: Port PR apache#4329 to add FLIP-27 enumerator classes (ap…

152394a

…ache#4979)

namrathamyske pushed a commit to namrathamyske/iceberg that referenced this pull request Jul 10, 2022

Flink: FLIP-27 source enumerator help classes (apache#4329)

2e23d40

namrathamyske pushed a commit to namrathamyske/iceberg that referenced this pull request Jul 10, 2022

Flink 1.15: Port PR apache#4329 to add FLIP-27 enumerator classes (ap…

c7d2213

…ache#4979)

Conversation

stevenzwu commented Mar 15, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

yittg left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yittg Mar 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yittg Mar 17, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevenzwu commented Mar 15, 2022 •

edited

Loading

yittg Mar 16, 2022 •

edited

Loading

yittg Mar 17, 2022 •

edited

Loading