This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 2460f8bed20a feat(spark): add DeltaStreamer checkpoint procedures
(#19631)
2460f8bed20a is described below
commit 2460f8bed20a8d8a5c1d922645f4345a0b425b22
Author: Danny Chan <[email protected]>
AuthorDate: Sat Aug 22 12:23:41 2026 +0800
feat(spark): add DeltaStreamer checkpoint procedures (#19631)
* feat(spark): add DeltaStreamer checkpoint procedures
---
.../DeltaStreamerCheckpointProcedure.scala | 205 +++++++++++++++++++++
.../hudi/command/procedures/HoodieProcedures.scala | 2 +
.../TestDeltaStreamerCheckpointProcedure.scala | 199 ++++++++++++++++++++
3 files changed, 406 insertions(+)
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltaStreamerCheckpointProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltaStreamerCheckpointProcedure.scala
new file mode 100644
index 000000000000..fc386f7fcdb8
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltaStreamerCheckpointProcedure.scala
@@ -0,0 +1,205 @@
+/*
+ * 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.spark.sql.hudi.command.procedures
+
+import org.apache.hudi.HoodieCLIUtils
+import org.apache.hudi.client.SparkRDDWriteClient
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.model.{HoodieFailedWritesCleaningPolicy,
HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.checkpoint.{Checkpoint, CheckpointUtils,
StreamerCheckpointV1, StreamerCheckpointV2}
+import org.apache.hudi.common.table.timeline.{HoodieTimeline, TimelineUtils}
+import org.apache.hudi.common.util.{Option => HOption}
+import org.apache.hudi.config.{HoodieCleanConfig, HoodieWriteConfig}
+import org.apache.hudi.exception.HoodieException
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField,
StructType}
+
+import java.util.function.Supplier
+
+import scala.util.control.NonFatal
+
+class GetDeltaStreamerCheckpointProcedure extends BaseProcedure with
ProcedureBuilder {
+ import DeltaStreamerCheckpointProcedureUtils._
+
+ private val PARAMETERS = Array[ProcedureParameter](
+ ProcedureParameter.optional(0, "table", DataTypes.StringType),
+ ProcedureParameter.optional(1, "path", DataTypes.StringType)
+ )
+
+ override def parameters: Array[ProcedureParameter] = PARAMETERS
+
+ override def outputType: StructType = OUTPUT_TYPE
+
+ override def call(args: ProcedureArgs): Seq[Row] = {
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
+ val metaClient = createMetaClient(jsc, getBasePath(tableName, tablePath))
+
+ val checkpoint = getLatestCheckpoint(metaClient)
+ if (checkpoint.isPresent) {
+ Seq(Row(checkpoint.get.getCheckpointKey))
+ } else {
+ Seq.empty
+ }
+ }
+
+ override def build: Procedure = new GetDeltaStreamerCheckpointProcedure
+}
+
+/**
+ * Publishes a DeltaStreamer checkpoint through an empty commit. Callers
should pause active
+ * ingestion unless the table is configured for multi-writer concurrency
control and locking.
+ * A later ingestion commit can legitimately advance the checkpoint again.
+ */
+class SetDeltaStreamerCheckpointProcedure extends BaseProcedure with
ProcedureBuilder {
+ import DeltaStreamerCheckpointProcedureUtils._
+
+ private val PARAMETERS = Array[ProcedureParameter](
+ ProcedureParameter.optional(0, "table", DataTypes.StringType),
+ ProcedureParameter.required(1, "checkpoint", DataTypes.StringType),
+ ProcedureParameter.optional(2, "path", DataTypes.StringType)
+ )
+
+ override def parameters: Array[ProcedureParameter] = PARAMETERS
+
+ override def outputType: StructType = OUTPUT_TYPE
+
+ override def call(args: ProcedureArgs): Seq[Row] = {
+ super.checkArgs(PARAMETERS, args)
+
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val checkpointValue = getArgValueOrDefault(args,
PARAMETERS(1)).get.asInstanceOf[String]
+ if (checkpointValue.trim.isEmpty) {
+ throw new IllegalArgumentException("DeltaStreamer checkpoint must not be
empty")
+ }
+ val tablePath = getArgValueOrDefault(args, PARAMETERS(2))
+ val basePath = getBasePath(tableName, tablePath)
+ val metaClient = createMetaClient(jsc, basePath)
+
+ val checkpoint = getLatestCheckpoint(metaClient)
+ .orElse(new StreamerCheckpointV1(checkpointValue))
+ checkpoint.setCheckpointKey(checkpointValue)
+ val checkpointMetadata = checkpoint.getCheckpointCommitMetadata(
+ checkpoint.getCheckpointResetKey, checkpoint.getCheckpointIgnoreKey)
+
+ val writeOptions = Map(
+ // This procedure only publishes checkpoint metadata. It must not run or
schedule table
+ // services as a side effect or clean up pending writes belonging to
another writer.
+ HoodieWriteConfig.TABLE_SERVICES_ENABLED.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key ->
HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key -> "true",
+ // Never upgrade or downgrade the table as a side effect of setting its
checkpoint.
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
metaClient.getTableConfig.getTableVersion.versionCode().toString,
+ HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> "false",
+ // A minimally configured writer must never remove metadata partitions
that are already
+ // present on disk. Available partitions are still updated based on
hoodie.properties.
+ HoodieMetadataConfig.AUTO_DELETE_PARTITIONS.key -> "false"
+ )
+
+ var client: SparkRDDWriteClient[AnyRef] = null
+ var instantTime: String = null
+ try {
+ client = HoodieCLIUtils.createHoodieWriteClient(
+ sparkSession,
+ basePath,
+ writeOptions,
+ tableName.map(_.asInstanceOf[String]))
+ .asInstanceOf[SparkRDDWriteClient[AnyRef]]
+
+ instantTime = client.startCommit(metaClient.getCommitActionType)
+ val writeStatuses = client.upsert(jsc.emptyRDD[HoodieRecord[AnyRef]],
instantTime)
+ val committed = client.commit(instantTime, writeStatuses,
HOption.of(checkpointMetadata))
+ if (!committed) {
+ throw new HoodieException(s"Failed to set DeltaStreamer checkpoint for
table at $basePath")
+ }
+ Seq(Row(checkpointValue))
+ } catch {
+ case NonFatal(failure) =>
+ // Roll back only if our instant is still pending. A commit can throw
after completing;
+ // rolling back that completed instant would discard a successfully
published checkpoint.
+ if (client != null && instantTime != null) {
+ try {
+ val stillPending = metaClient.reloadActiveTimeline()
+ .filterInflightsAndRequested()
+ .containsInstant(instantTime)
+ if (stillPending && !client.rollback(instantTime)) {
+ failure.addSuppressed(new HoodieException(
+ s"Failed to rollback DeltaStreamer checkpoint instant
$instantTime"))
+ }
+ } catch {
+ case NonFatal(rollbackFailure) =>
failure.addSuppressed(rollbackFailure)
+ }
+ }
+ throw failure
+ } finally {
+ if (client != null) {
+ client.close()
+ }
+ }
+ }
+
+ override def build: Procedure = new SetDeltaStreamerCheckpointProcedure
+}
+
+private object DeltaStreamerCheckpointProcedureUtils {
+ private val CHECKPOINT_KEYS = Array(
+ StreamerCheckpointV1.STREAMER_CHECKPOINT_KEY_V1,
+ StreamerCheckpointV1.STREAMER_CHECKPOINT_RESET_KEY_V1,
+ StreamerCheckpointV2.STREAMER_CHECKPOINT_KEY_V2,
+ StreamerCheckpointV2.STREAMER_CHECKPOINT_RESET_KEY_V2
+ )
+
+ val OUTPUT_TYPE: StructType = new StructType(Array[StructField](
+ StructField("checkpoint", DataTypes.StringType, nullable = true,
Metadata.empty)
+ ))
+
+ def getLatestCheckpoint(metaClient: HoodieTableMetaClient):
HOption[Checkpoint] = {
+ val commitsTimeline = getIngestionTimeline(metaClient)
+ TimelineUtils.getLatestInstantAndCommitMetadataWithValidCheckpointInfo(
+ commitsTimeline, CHECKPOINT_KEYS: _*)
+ .map(pair => CheckpointUtils.getCheckpoint(pair.getRight))
+ }
+
+ private def getIngestionTimeline(metaClient: HoodieTableMetaClient):
HoodieTimeline = {
+ val commitsTimeline =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ val deltaCommitTimeline = commitsTimeline.filter(
+ instant => instant.getAction == HoodieTimeline.DELTA_COMMIT_ACTION)
+
+ // Match Streamer's resume behavior: once a MOR table has delta commits,
checkpoints from
+ // older COW commits are no longer considered.
+ if (metaClient.getTableType == HoodieTableType.MERGE_ON_READ &&
!deltaCommitTimeline.empty()) {
+ deltaCommitTimeline
+ } else {
+ commitsTimeline
+ }
+ }
+}
+
+object GetDeltaStreamerCheckpointProcedure {
+ val NAME = "get_deltastreamer_checkpoint"
+
+ def builder: Supplier[ProcedureBuilder] = () => new
GetDeltaStreamerCheckpointProcedure
+}
+
+object SetDeltaStreamerCheckpointProcedure {
+ val NAME = "set_deltastreamer_checkpoint"
+
+ def builder: Supplier[ProcedureBuilder] = () => new
SetDeltaStreamerCheckpointProcedure
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
index f602b6506264..e5fd5285bf3c 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
@@ -111,6 +111,8 @@ object HoodieProcedures {
,(ShowAuditLockStatusProcedure.NAME,
ShowAuditLockStatusProcedure.builder)
,(ValidateAuditLockProcedure.NAME, ValidateAuditLockProcedure.builder)
,(CleanupAuditLockProcedure.NAME, CleanupAuditLockProcedure.builder)
+ ,(GetDeltaStreamerCheckpointProcedure.NAME,
GetDeltaStreamerCheckpointProcedure.builder)
+ ,(SetDeltaStreamerCheckpointProcedure.NAME,
SetDeltaStreamerCheckpointProcedure.builder)
)
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestDeltaStreamerCheckpointProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestDeltaStreamerCheckpointProcedure.scala
new file mode 100644
index 000000000000..2016bd311f02
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestDeltaStreamerCheckpointProcedure.scala
@@ -0,0 +1,199 @@
+/*
+ * 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.spark.sql.hudi.procedure
+
+import org.apache.hudi.DataSourceWriteOptions.{ORDERING_FIELDS,
RECORDKEY_FIELD, TABLE_TYPE}
+import org.apache.hudi.common.table.{HoodieTableMetaClient, HoodieTableVersion}
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.{Option => HOption}
+import org.apache.hudi.config.{HoodieCompactionConfig, HoodieWriteConfig}
+import org.apache.hudi.hadoop.fs.HadoopFSUtils
+
+import org.apache.spark.sql.SaveMode
+
+class TestDeltaStreamerCheckpointProcedure extends
HoodieSparkProcedureTestBase {
+
+ Seq("cow", "mor").foreach { tableType =>
+ test(s"get and set deltastreamer checkpoint for $tableType table") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'a1', 1000)")
+
+ assertResult(0) {
+ spark.sql(s"call get_deltastreamer_checkpoint(table =>
'$tableName')").count()
+ }
+
+ checkAnswer(
+ s"call set_deltastreamer_checkpoint(table => '$tableName',
checkpoint => 'checkpoint-1')")(
+ Seq("checkpoint-1"))
+ checkAnswer(s"call get_deltastreamer_checkpoint(table =>
'$tableName')")(
+ Seq("checkpoint-1"))
+
+ checkAnswer(s"select id, name, ts from $tableName")(Seq(1, "a1",
1000L))
+
+ // Both procedures can address a table by path when it is not
registered in the catalog.
+ checkAnswer(
+ s"call set_deltastreamer_checkpoint(path => '$tablePath', checkpoint
=> 'checkpoint-2')")(
+ Seq("checkpoint-2"))
+ checkAnswer(s"call get_deltastreamer_checkpoint(path =>
'$tablePath')")(
+ Seq("checkpoint-2"))
+
+ checkExceptionContain(
+ s"call set_deltastreamer_checkpoint(table => '$tableName',
checkpoint => ' ')")(
+ "DeltaStreamer checkpoint must not be empty")
+ checkAnswer(s"call get_deltastreamer_checkpoint(table =>
'$tableName')")(
+ Seq("checkpoint-2"))
+ }
+ }
+ }
+
+ test("setting checkpoint does not remove metadata table partitions") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | orderingFields = 'ts',
+ | hoodie.metadata.enable = 'true',
+ | hoodie.metadata.index.column.stats.enable = 'true'
+ | )
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'a1', 1000)")
+
+ val metadataPartitionsBefore =
loadMetaClient(tablePath).getTableConfig.getMetadataPartitions
+ assert(metadataPartitionsBefore.contains("files"))
+ assert(metadataPartitionsBefore.contains("column_stats"))
+
+ checkAnswer(
+ s"call set_deltastreamer_checkpoint(table => '$tableName', checkpoint
=> 'checkpoint-1')")(
+ Seq("checkpoint-1"))
+
+ assertResult(metadataPartitionsBefore) {
+ loadMetaClient(tablePath).getTableConfig.getMetadataPartitions
+ }
+ }
+ }
+
+ test("setting checkpoint does not schedule table services") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | type = 'mor',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'a1', 1000)")
+
+ withSQLConf(
+ HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT.key -> "true",
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key -> "1") {
+ checkAnswer(
+ s"call set_deltastreamer_checkpoint(table => '$tableName',
checkpoint => 'checkpoint-1')")(
+ Seq("checkpoint-1"))
+ }
+
+
assert(loadMetaClient(tablePath).getActiveTimeline.filterPendingCompactionTimeline.empty())
+ }
+ }
+
+ test("setting checkpoint preserves table version and pending instants") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val hoodieTableName = tableName.split('.').last
+ val tablePath = s"${tmp.getCanonicalPath}/$hoodieTableName"
+ import spark.implicits._
+ Seq((1, "a1", 1000L)).toDF("id", "name", "ts")
+ .write.format("hudi")
+ .option(HoodieWriteConfig.TBL_NAME.key, hoodieTableName)
+ .option(TABLE_TYPE.key, "COPY_ON_WRITE")
+ .option(RECORDKEY_FIELD.key, "id")
+ .option(ORDERING_FIELDS.key, "ts")
+ .option(HoodieWriteConfig.WRITE_TABLE_VERSION.key,
HoodieTableVersion.EIGHT.versionCode().toString)
+ .option(HoodieWriteConfig.AUTO_UPGRADE_VERSION.key, "false")
+ .option("hoodie.metadata.enable", "false")
+ .mode(SaveMode.Overwrite)
+ .save(tablePath)
+ spark.sql(s"create table $tableName using hudi location '$tablePath'")
+
+ val metaClient = loadMetaClient(tablePath)
+
assertResult(HoodieTableVersion.EIGHT)(metaClient.getTableConfig.getTableVersion)
+ val pendingInstantTime = metaClient.createNewInstantTime(false)
+ val requested =
metaClient.getTimelineLayout.getInstantGenerator.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.COMMIT_ACTION,
pendingInstantTime)
+ metaClient.getActiveTimeline.createNewInstant(requested)
+ metaClient.getActiveTimeline.transitionRequestedToInflight(
+ requested, HOption.empty[Array[Byte]]())
+
+ checkAnswer(
+ s"call set_deltastreamer_checkpoint(table => '$tableName', checkpoint
=> 'checkpoint-1')")(
+ Seq("checkpoint-1"))
+
+ val updatedMetaClient = loadMetaClient(tablePath)
+
assertResult(HoodieTableVersion.EIGHT)(updatedMetaClient.getTableConfig.getTableVersion)
+ val preservedPendingInstant = updatedMetaClient.getActiveTimeline
+ .filterInflightsAndRequested()
+ .getInstantsAsStream
+ .filter(instant => instant.requestedTime == pendingInstantTime)
+ .findFirst()
+ assert(preservedPendingInstant.isPresent)
+ assert(preservedPendingInstant.get.isInflight)
+ checkAnswer(s"call get_deltastreamer_checkpoint(table => '$tableName')")(
+ Seq("checkpoint-1"))
+ }
+ }
+
+ private def loadMetaClient(tablePath: String): HoodieTableMetaClient = {
+ HoodieTableMetaClient.builder()
+
.setConf(HadoopFSUtils.getStorageConfWithCopy(spark.sparkContext.hadoopConfiguration))
+ .setBasePath(tablePath)
+ .build()
+ }
+}
