[HUDI-8726] s3/gcs incremental source should stick to checkpoint v1 #12688

Davis-Zhang-Onehouse · 2025-01-22T00:23:53Z

Reviewers please review commit by commit, checking the commit messages would save you great amount of time

Change Logs

Fixed and bump up massive test coverage for checkpoint and some missing coverage when related code is introduced

How the checkpoint is consumed

  public Pair<InputBatch, Boolean> readFromSource(String instantTime, HoodieTableMetaClient metaClient) throws IOException {
    // Retrieve the previous round checkpoints, if any. Consumption of checkpoint
    Option<Checkpoint> checkpointToResume = StreamerCheckpointUtils.getCheckpointToResumeFrom(commitsTimelineOpt, cfg, props);


  public static Option<Checkpoint> getCheckpointToResumeFrom(Option<HoodieTimeline> commitsTimelineOpt,
                                                             HoodieStreamer.Config streamerConfig,
                                                             TypedProperties props) throws IOException {
    Option<Checkpoint> checkpoint = Option.empty();
    if (commitsTimelineOpt.isPresent()) {
      checkpoint = getCheckpointToResumeString(commitsTimelineOpt, streamerConfig, props);
    }

    LOG.debug("Checkpoint from config: " + streamerConfig.checkpoint);
    if (!checkpoint.isPresent() && streamerConfig.checkpoint != null) {
      int writeTableVersion = ConfigUtils.getIntWithAltKeys(props, HoodieWriteConfig.WRITE_TABLE_VERSION);
      // TODO: we need to add data source class! Similarily, check all other places where we use this func as well!
      checkpoint = Option.of(CheckpointUtils.targetCheckpointV2(writeTableVersion, streamerConfig.sourceClassName)
          ? new StreamerCheckpointV2(streamerConfig.checkpoint) : new StreamerCheckpointV1(streamerConfig.checkpoint));
    }
    return checkpoint;
  }

How it is produced (from the deepest call stack to the caller at a higher level)

public class HoodieIncrSource extends RowSource {
  @Override
  public Pair<Option<Dataset<Row>>, Checkpoint> fetchNextBatch(Option<Checkpoint> lastCheckpoint, long sourceLimit) {
    if (CheckpointUtils.targetCheckpointV2(writeTableVersion, getClass().getName())) {
      return fetchNextBatchBasedOnCompletionTime(lastCheckpoint, sourceLimit);
    } else {
      return fetchNextBatchBasedOnRequestedTime(lastCheckpoint, sourceLimit); <== as long as the branch condition is good, we should be good.
    }
  }

  @Override
  protected final InputBatch<Dataset<Row>> fetchNewData(Option<String> lastCkptStr, long sourceLimit) {
    throw new UnsupportedOperationException("RowSource#fetchNewData should not be called");
  }

public abstract class RowSource extends Source<Dataset<Row>> {
  @Override
  protected final InputBatch<Dataset<Row>> fetchNewData(Option<String> lastCkptStr, long sourceLimit) {
    throw new UnsupportedOperationException("RowSource#fetchNewData should not be called");
  }

  @Override
  protected final InputBatch<Dataset<Row>> readFromCheckpoint(Option<Checkpoint> lastCheckpoint, long sourceLimit) {
    Pair<Option<Dataset<Row>>, Checkpoint> res = fetchNextBatch(lastCheckpoint, sourceLimit);
    return res.getKey().map(dsr -> {
      Dataset<Row> sanitizedRows = SanitizationUtils.sanitizeColumnNamesForAvro(dsr, props);
      SchemaProvider rowSchemaProvider =
          UtilHelpers.createRowBasedSchemaProvider(sanitizedRows.schema(), props, sparkContext);
      Dataset<Row> wrappedDf = HoodieSparkUtils.maybeWrapDataFrameWithException(sanitizedRows, HoodieReadFromSourceException.class.getName(),
          "Failed to read from row source", ConfigUtils.getBooleanWithAltKeys(props, ROW_THROW_EXPLICIT_EXCEPTIONS));
      return new InputBatch<>(Option.of(wrappedDf), res.getValue(), rowSchemaProvider);
    }).orElseGet(() -> new InputBatch<>(res.getKey(), res.getValue()));
  }


public abstract class Source<T> implements SourceCommitCallback, Serializable {
  public final InputBatch<T> fetchNext(Option<Checkpoint> lastCheckpoint, long sourceLimit) {
    InputBatch<T> batch = readFromCheckpoint(translateCheckpoint(lastCheckpoint), sourceLimit);
    // If overriddenSchemaProvider is passed in CLI, use it
    return overriddenSchemaProvider == null ? batch
        : new InputBatch<>(batch.getBatch(), batch.getCheckpointForNextBatch(), overriddenSchemaProvider);
  }

  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
  protected InputBatch<T> readFromCheckpoint(Option<Checkpoint> lastCheckpoint, long sourceLimit) {
    LOG.warn("In Hudi 1.0+, the checkpoint based on Hudi timeline is changed. "
        + "If your Source implementation relies on request time as the checkpoint, "
        + "you may consider migrating to completion time-based checkpoint by overriding "
        + "Source#translateCheckpoint and Source#fetchNewDataFromCheckpoint");
    return fetchNewData(
        lastCheckpoint.isPresent()
            ? Option.of(lastCheckpoint.get().getCheckpointKey()) : Option.empty(),
        sourceLimit);
  }


  public InputBatch<Dataset<Row>> fetchNewDataInRowFormat(Option<Checkpoint> lastCheckpoint, long sourceLimit) {

  Pair<InputBatch, Boolean> fetchNextBatchFromSource(Option<Checkpoint> resumeCheckpoint, HoodieTableMetaClient metaClient) {

  public Pair<InputBatch, Boolean> readFromSource(String instantTime, HoodieTableMetaClient metaClient) throws IOException {
  private Pair<Option<String>, JavaRDD<WriteStatus>> writeToSinkAndDoMetaSync(String instantTime, InputBatch inputBatch,
    if (!hasErrors || cfg.commitOnErrors) { 
      Map<String, String> checkpointCommitMetadata = <-------------- generate commit metadata
          !getBooleanWithAltKeys(props, CHECKPOINT_FORCE_SKIP)
            ?
              inputBatch.getCheckpointForNextBatch() != null
              ?
                inputBatch.getCheckpointForNextBatch().getCheckpointCommitMetadata(cfg.checkpoint, cfg.ignoreCheckpoint)
                :
                    CheckpointUtils.targetCheckpointV2(writeTableVersion, streamerConfig.sourceClassName)
                    ?
                        new StreamerCheckpointV2((String) null).getCheckpointCommitMetadata(cfg.checkpoint, cfg.ignoreCheckpoint)
                    :
                        new StreamerCheckpointV1((String) null).getCheckpointCommitMetadata(cfg.checkpoint, cfg.ignoreCheckpoint)
            :
              Collections.emptyMap();

      boolean success = writeClient.commit(instantTime, writeStatusRDD, Option.of(checkpointCommitMetadata), commitActionType, partitionToReplacedFileIds, Option.empty()); <-- write bindly of whatever kv in the checkpointCommitMetadata.

Impact

as title mentioned

Risk level (write none, low medium or high below)

low

Documentation Update

none

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

Apart from basic unit test, we need e2e functional test of the s3 / gcs source. I can only think of writing IT test coverage.

nsivabalan

Source code changes looks good to me.
while I review test code, reminding @yihua to review the logic and certify

nsivabalan

few minor comments

nsivabalan · 2025-01-27T04:50:04Z

hudi-common/src/main/java/org/apache/hudi/common/table/checkpoint/CheckpointUtils.java


 public class CheckpointUtils {

+  public static final Set<String> DATASOURCES_MUST_USE_CKP_V1 = new HashSet<>(Arrays.asList(


can we name this "DATASOURCES_NOT_SUPPORTED_WITH_CKPT_V2"

nsivabalan · 2025-01-27T04:53:37Z

hudi-common/src/test/java/org/apache/hudi/common/table/checkpoint/TestCheckpointUtils.java

+      "6, org.apache.hudi.utilities.sources.AnotherSource, false",
+      // Disallowed sources should return false even with version >= 8
+      "8, org.apache.hudi.utilities.sources.S3EventsHoodieIncrSource, false",
+      "8, org.apache.hudi.utilities.sources.GcsEventsHoodieIncrSource, false"


can we also add MockS3EventsHoodieIncrSource

nsivabalan · 2025-01-27T04:54:31Z

hudi-common/src/main/java/org/apache/hudi/common/table/checkpoint/CheckpointUtils.java

@@ -49,14 +59,15 @@ public static Checkpoint getCheckpoint(HoodieCommitMetadata commitMetadata) {
    throw new HoodieException("Checkpoint is not found in the commit metadata: " + commitMetadata.getExtraMetadata());
  }

-  public static boolean targetCheckpointV2(int writeTableVersion) {
-    return writeTableVersion >= HoodieTableVersion.EIGHT.versionCode();
+  public static boolean targetCheckpointV2(int writeTableVersion, String sourceClassName) {


lets rename this to

shouldTargetCheckpointV2

nsivabalan · 2025-01-27T04:55:33Z

hudi-common/src/main/java/org/apache/hudi/common/table/checkpoint/CheckpointUtils.java

  }

  // TODO(yihua): for checkpoint translation, handle cases where the checkpoint is not exactly the
  // instant or completion time
  public static StreamerCheckpointV2 convertToCheckpointV2ForCommitTime(
-      Checkpoint checkpoint, HoodieTableMetaClient metaClient) {
+      Checkpoint checkpoint, HoodieTableMetaClient metaClient, TimelineUtils.HollowCommitHandling handlingMode) {


minor.
hollowCommitHandlingMode

nsivabalan · 2025-01-27T04:56:01Z

...i-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamSourceV1.scala

@@ -188,7 +188,7 @@ class HoodieStreamSourceV1(sqlContext: SQLContext,
  private def translateCheckpoint(commitTime: String): String = {
    if (writeTableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT)) {
      CheckpointUtils.convertToCheckpointV2ForCommitTime(
-        new StreamerCheckpointV1(commitTime), metaClient).getCheckpointKey
+        new StreamerCheckpointV1(commitTime), metaClient, hollowCommitHandling).getCheckpointKey


same here. naming -> hollowCommitHandlingMode

nsivabalan · 2025-01-27T05:02:20Z

...-utilities/src/test/java/org/apache/hudi/utilities/sources/MockS3EventsHoodieIncrSource.java

+ * This class simulates different checkpoint and data fetch scenarios to test the checkpoint handling
+ * and data ingestion behavior of the StreamSync class.
+ */
+public class MockS3EventsHoodieIncrSource extends S3EventsHoodieIncrSource {


can we write a similar one for GCS ?

nsivabalan · 2025-01-27T05:04:39Z

hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestStreamSync.java

@@ -149,32 +155,6 @@ void testFetchNextBatchFromSource(Boolean useRowWriter, Boolean hasTransformer,
            HoodieErrorTableConfig.ERROR_ENABLE_VALIDATE_TARGET_SCHEMA.defaultValue());
  }

-  @ParameterizedTest
-  @MethodSource("getCheckpointToResumeCases")
-  void testGetCheckpointToResume(HoodieStreamer.Config cfg, HoodieCommitMetadata commitMetadata, Option<String> expectedResumeCheckpoint) throws IOException {


why removed this? did you move this elsewhere?

yes, check hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestStreamerCheckpointUtils.java

hudi-common/src/test/java/org/apache/hudi/common/table/checkpoint/TestCheckpointUtils.java

nsivabalan · 2025-01-27T22:17:53Z

@Davis-Zhang-Onehouse : there are some CI failures. can you follow up.

Davis-Zhang-Onehouse · 2025-01-27T22:52:06Z

fixed the checkstyle, will ping once CI all green and is ready to merge. Thanks!

Pure refactoring of the test classes - moving some method to the parent class so the subsequent commits can use them. Extracting methods to a new class so it can be consumed by later changes.

For testing ingestion flow e2e to ensure we consume the right checkpoint version and write out the right checkpoint version, we split the coverage into 2 parts: - Test anything interacting with S3/GCS incremental source. This is done by introducing a dummy class as an injected dependency so we can do validations and trigger the ingestion code covering the e2e behavior - Test S3/GCS incremental source itself, this is done by existing unit test against the class, they have done the testing already about the relevant code part.

Previously in order for EMPTY_ROW_SET_NONE_NULL_CKP to return different value, we create different BiFunction with hard coded value. Now it is parameterized via RETURN_CHECKPOINT_KEY. Also moved some constant member to the right util class.

Renaming & minor unit test changes. Next PR we introduce MockGcs source coverage just like MockS3 source.

hudi-bot · 2025-01-28T18:59:39Z

CI report:

e4727b5 UNKNOWN
00653bc UNKNOWN
0d26d85 UNKNOWN
914f1f7 UNKNOWN
6e47bcc Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

github-actions bot added the size:M PR with lines of changes in (100, 300] label Jan 22, 2025

Davis-Zhang-Onehouse force-pushed the HUDI-8726-s3gcs branch from 75369c9 to 65a87a7 Compare January 23, 2025 03:06

github-actions bot added size:XL PR with lines of changes > 1000 and removed size:M PR with lines of changes in (100, 300] labels Jan 23, 2025

Davis-Zhang-Onehouse force-pushed the HUDI-8726-s3gcs branch 6 times, most recently from 33e010e to 4fc962e Compare January 23, 2025 23:24

nsivabalan reviewed Jan 24, 2025

View reviewed changes

Davis-Zhang-Onehouse force-pushed the HUDI-8726-s3gcs branch 5 times, most recently from 00653bc to 0d26d85 Compare January 27, 2025 01:47

nsivabalan reviewed Jan 27, 2025

View reviewed changes

Davis-Zhang-Onehouse commented Jan 27, 2025

View reviewed changes

hudi-common/src/test/java/org/apache/hudi/common/table/checkpoint/TestCheckpointUtils.java Show resolved Hide resolved

Davis-Zhang-Onehouse force-pushed the HUDI-8726-s3gcs branch from 397ddda to d051243 Compare January 27, 2025 17:31

nsivabalan approved these changes Jan 27, 2025

View reviewed changes

Davis-Zhang-Onehouse mentioned this pull request Jan 28, 2025

[HUDI-8917] Checkpoint reset handling #12718

Merged

4 tasks

Davis-Zhang-Onehouse added 7 commits January 27, 2025 19:39

cover USE_TRANSITION_TIME for ckp conversion

2fdc806

Test util refactor

76767af

Pure refactoring of the test classes - moving some method to the parent class so the subsequent commits can use them. Extracting methods to a new class so it can be consumed by later changes.

Add checkpoint V2 validation in existing tests

bfece78

Fix gaps of using the right checkpoint version

eefcf6b

Unit test coverages on getCheckpointToResume

c6d249d

address PR comments

c463cfd

Davis-Zhang-Onehouse added 5 commits January 27, 2025 19:39

Add more test coverages

97cb8f2

Address PR comments 1

847077c

Renaming & minor unit test changes. Next PR we introduce MockGcs source coverage just like MockS3 source.

add mock GCS data source test coverage

f9f15e1

fix checkstyle

cc57705

Davis-Zhang-Onehouse force-pushed the HUDI-8726-s3gcs branch from 7cffcb8 to 2a21429 Compare January 28, 2025 03:39

fix CI failure

6e47bcc

Davis-Zhang-Onehouse force-pushed the HUDI-8726-s3gcs branch from 2a21429 to 6e47bcc Compare January 28, 2025 16:02

nsivabalan merged commit 0ae996b into apache:master Jan 28, 2025
43 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[HUDI-8726] s3/gcs incremental source should stick to checkpoint v1 #12688

[HUDI-8726] s3/gcs incremental source should stick to checkpoint v1 #12688

Davis-Zhang-Onehouse commented Jan 22, 2025 •

edited

Loading

nsivabalan left a comment

nsivabalan left a comment

nsivabalan Jan 27, 2025

Davis-Zhang-Onehouse Jan 27, 2025

nsivabalan Jan 27, 2025

Davis-Zhang-Onehouse Jan 27, 2025

nsivabalan Jan 27, 2025

Davis-Zhang-Onehouse Jan 27, 2025

nsivabalan Jan 27, 2025

Davis-Zhang-Onehouse Jan 27, 2025

nsivabalan Jan 27, 2025

Davis-Zhang-Onehouse Jan 27, 2025

nsivabalan Jan 27, 2025

Davis-Zhang-Onehouse Jan 27, 2025

nsivabalan Jan 27, 2025

Davis-Zhang-Onehouse Jan 27, 2025

nsivabalan commented Jan 27, 2025

Davis-Zhang-Onehouse commented Jan 27, 2025

hudi-bot commented Jan 28, 2025


		public class CheckpointUtils {

		public static final Set<String> DATASOURCES_MUST_USE_CKP_V1 = new HashSet<>(Arrays.asList(

[HUDI-8726] s3/gcs incremental source should stick to checkpoint v1 #12688

[HUDI-8726] s3/gcs incremental source should stick to checkpoint v1 #12688

Conversation

Davis-Zhang-Onehouse commented Jan 22, 2025 • edited Loading

Reviewers please review commit by commit, checking the commit messages would save you great amount of time

Change Logs

How the checkpoint is consumed

How it is produced (from the deepest call stack to the caller at a higher level)

Impact

Risk level (write none, low medium or high below)

Documentation Update

Contributor's checklist

nsivabalan left a comment

Choose a reason for hiding this comment

nsivabalan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nsivabalan commented Jan 27, 2025

Davis-Zhang-Onehouse commented Jan 27, 2025

hudi-bot commented Jan 28, 2025

CI report:

Davis-Zhang-Onehouse commented Jan 22, 2025 •

edited

Loading