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

[Davis-Zhang-Onehouse]

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

[hudi-bot]

CI report:

• e4727b5 UNKNOWN
• 0c644d8 Azure: CANCELED
• 00653bc UNKNOWN
• 0d26d85 UNKNOWN

Bot commands

@hudi-bot supports the following commands:

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