You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "danny0405 (via GitHub)" <gi...@apache.org> on 2026/08/14 03:55:46 UTC
[PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
danny0405 opened a new pull request, #19631:
URL: https://github.com/apache/hudi/pull/19631
### Describe the issue this Pull Request addresses
Users currently cannot inspect or override Hudi Streamer checkpoints through Spark SQL stored procedures. Setting a checkpoint requires publishing checkpoint metadata on the table timeline, but constructing a minimally configured writer for that empty commit can inadvertently remove metadata-table partitions when their enablement properties are not supplied.
This PR adds stored procedures that read and safely update Streamer checkpoints without requiring callers to repeat writer properties.
### Summary and Changelog
- Add `get_deltastreamer_checkpoint`, addressable by catalog table name or table path, to return the latest valid V1 or V2 Streamer checkpoint.
- Add `set_deltastreamer_checkpoint`, addressable by catalog table name or table path, to publish the requested checkpoint through an empty commit.
- Derive writer configuration from catalog and persisted table properties, preserve the existing checkpoint version plus reset/ignore metadata, and match Streamer checkpoint lookup semantics for MOR tables.
- Disable cleaning, archival, inline compaction/clustering, and metadata-partition auto-deletion for the metadata-only write.
- Register both procedures and add functional coverage for COW and MOR tables, table-name and path access, unchanged table data, and preservation of existing metadata-table partitions.
- No code was copied.
### Impact
This adds two user-facing Spark SQL stored procedures. Setting a checkpoint adds an empty commit to the table timeline and updates enabled metadata-table partitions when applicable. There are no storage-format, configuration-default, backward-compatibility, or expected performance changes for existing write paths.
### Risk Level
medium
The setter exercises the standard write client and table timeline to create a metadata-only commit, including metadata-table updates. Risk is mitigated by explicitly disabling unrelated table services and metadata-partition deletion, and by `TestDeltastreamerCheckpointProcedure`, which passes for COW, MOR, path-based invocation, data preservation, and metadata-partition preservation.
Validation:
- `mvn -pl hudi-spark-datasource/hudi-spark -Dscala-2.12 -Dtest=org.apache.spark.sql.hudi.procedure.TestDeltastreamerCheckpointProcedure -Dsuites=org.apache.spark.sql.hudi.procedure.TestDeltastreamerCheckpointProcedure -Dcheckstyle.skip -Drat.skip test`
- Result: 3 tests passed; Checkstyle and Scalastyle passed.
- `git diff --check`
### Documentation Update
Required. This introduces user-facing stored procedures, so the Hudi website stored-procedure reference should document their syntax and empty-commit behavior. Website sources are not part of this repository checkout and should be updated separately.
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #19631:
URL: https://github.com/apache/hudi/pull/19631#discussion_r3827028955
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltaStreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieClusteringConfig.INLINE_CLUSTERING.key -> "false",
+ HoodieClusteringConfig.SCHEDULE_INLINE_CLUSTERING.key -> "false",
+ HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key -> "true",
+ // 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
+ try {
+ client = HoodieCLIUtils.createHoodieWriteClient(
+ sparkSession,
+ basePath,
+ writeOptions,
+ tableName.map(_.asInstanceOf[String]))
+ .asInstanceOf[SparkRDDWriteClient[AnyRef]]
+
+ val instantTime = client.startCommit(metaClient.getCommitActionType)
+ val writeStatuses = client.upsert(jsc.emptyRDD[HoodieRecord[AnyRef]], instantTime)
+ val committed = client.commit(instantTime, writeStatuses, HOption.of(checkpointMetadata))
Review Comment:
Addressed in be03dc0e9acf. The setter now tracks its instant and, on any non-fatal failure, reloads the active timeline and rolls the instant back if it is still requested or inflight. Rollback failures are suppressed onto the original exception. The pending-state guard avoids rolling back a commit that completed before an exception was raised.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on code in PR #19631:
URL: https://github.com/apache/hudi/pull/19631#discussion_r3826788043
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieClusteringConfig.INLINE_CLUSTERING.key -> "false",
+ HoodieClusteringConfig.SCHEDULE_INLINE_CLUSTERING.key -> "false",
+ HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key -> "true",
+ // 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
+ try {
+ client = HoodieCLIUtils.createHoodieWriteClient(
Review Comment:
`createHoodieWriteClient` defaults the writer table version to the current version with auto-upgrade enabled; `hoodie.table.version` from hoodie.properties is a different property. Invoking this procedure on a v6/v8/v9 table can therefore upgrade it before publishing the checkpoint. The upgrade path also rolls back failed instants using EAGER policy, bypassing the `NEVER` setting above and potentially touching another writer's pending work. Please pin `WRITE_TABLE_VERSION` to the table's current version and disable `AUTO_UPGRADE_VERSION`, with a test asserting that an older table's version and pending instants remain unchanged.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
Review Comment:
This disables inline compaction execution but not all post-commit table services. If `hoodie.compact.schedule.inline=true`, this empty commit can schedule compaction; inline log compaction can execute, and partition TTL can also run. That violates the metadata-only contract. Please disable `HoodieWriteConfig.TABLE_SERVICES_ENABLED` for this client, or explicitly override every scheduling/execution flag, and add coverage with one of these options enabled.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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)
Review Comment:
An empty checkpoint is accepted and committed, but `getLatestCheckpoint` only recognizes non-empty checkpoint/reset metadata. Without a preserved reset key, the new commit is skipped and the getter returns an older checkpoint even though the setter returned success. Please reject empty checkpoint values, or define explicit clear semantics that shadow older checkpoints, and test an empty set followed by get.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19631:
URL: https://github.com/apache/hudi/pull/19631#issuecomment-5364937717
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286",
"triggerID" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"triggerType" : "PUSH"
}, {
"hash" : "be03dc0e9acf150ca2415a59ba4e682142f71382",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16437",
"triggerID" : "be03dc0e9acf150ca2415a59ba4e682142f71382",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 699e554d10e9dededb4c67d7a7601a54da9a347c Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286)
* be03dc0e9acf150ca2415a59ba4e682142f71382 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16437)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19631:
URL: https://github.com/apache/hudi/pull/19631#issuecomment-5364819915
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286",
"triggerID" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"triggerType" : "PUSH"
}, {
"hash" : "be03dc0e9acf150ca2415a59ba4e682142f71382",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "be03dc0e9acf150ca2415a59ba4e682142f71382",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 699e554d10e9dededb4c67d7a7601a54da9a347c Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286)
* be03dc0e9acf150ca2415a59ba4e682142f71382 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #19631:
URL: https://github.com/apache/hudi/pull/19631#discussion_r3827030047
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieClusteringConfig.INLINE_CLUSTERING.key -> "false",
+ HoodieClusteringConfig.SCHEDULE_INLINE_CLUSTERING.key -> "false",
+ HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key -> "true",
+ // 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
+ try {
+ client = HoodieCLIUtils.createHoodieWriteClient(
+ sparkSession,
+ basePath,
+ writeOptions,
+ tableName.map(_.asInstanceOf[String]))
+ .asInstanceOf[SparkRDDWriteClient[AnyRef]]
+
+ val instantTime = client.startCommit(metaClient.getCommitActionType)
Review Comment:
Addressed in be03dc0e9acf by documenting the operational contract on the setter: pause active ingestion unless the table is configured for multi-writer concurrency control and locking. The write client still derives table and session concurrency and lock configuration, so it can participate in that setup. A later Streamer commit can legitimately advance the checkpoint again, so paused ingestion remains the recommended way to perform an administrative reset.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+class GetDeltastreamerCheckpointProcedure extends BaseProcedure with ProcedureBuilder {
+ import DeltastreamerCheckpointProcedureUtils._
Review Comment:
Addressed in be03dc0e9acf. Renamed the file, procedure classes, utility object, registrations, and test suite to use DeltaStreamer. The SQL procedure names remain unchanged for API compatibility.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
Review Comment:
Addressed in be03dc0e9acf. The checkpoint writer now sets HoodieWriteConfig.TABLE_SERVICES_ENABLED=false instead of relying on incomplete per-service overrides. Added a MOR regression with inline compaction scheduling enabled and a one-delta-commit threshold; it verifies that the checkpoint commit leaves no pending compaction.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltaStreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieClusteringConfig.INLINE_CLUSTERING.key -> "false",
+ HoodieClusteringConfig.SCHEDULE_INLINE_CLUSTERING.key -> "false",
+ HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key -> "true",
+ // 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
+ try {
+ client = HoodieCLIUtils.createHoodieWriteClient(
Review Comment:
Addressed in be03dc0e9acf. The writer now pins WRITE_TABLE_VERSION to the version from the table config and sets AUTO_UPGRADE_VERSION=false. Added coverage using a v8 table with a pre-existing inflight commit; after setting the checkpoint, both the v8 table version and the unrelated inflight instant are preserved.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltaStreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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)
Review Comment:
Addressed in be03dc0e9acf. Blank and empty checkpoint values are rejected before the write client is created. The COW and MOR tests now attempt a blank set and verify that the previously published checkpoint remains visible.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19631:
URL: https://github.com/apache/hudi/pull/19631#issuecomment-5289355961
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286",
"triggerID" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 699e554d10e9dededb4c67d7a7601a54da9a347c Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19631:
URL: https://github.com/apache/hudi/pull/19631#issuecomment-5290052909
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286",
"triggerID" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 699e554d10e9dededb4c67d7a7601a54da9a347c Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #19631:
URL: https://github.com/apache/hudi/pull/19631#discussion_r3786635300
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieClusteringConfig.INLINE_CLUSTERING.key -> "false",
+ HoodieClusteringConfig.SCHEDULE_INLINE_CLUSTERING.key -> "false",
+ HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key -> "true",
+ // 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
+ try {
+ client = HoodieCLIUtils.createHoodieWriteClient(
+ sparkSession,
+ basePath,
+ writeOptions,
+ tableName.map(_.asInstanceOf[String]))
+ .asInstanceOf[SparkRDDWriteClient[AnyRef]]
+
+ val instantTime = client.startCommit(metaClient.getCommitActionType)
+ val writeStatuses = client.upsert(jsc.emptyRDD[HoodieRecord[AnyRef]], instantTime)
+ val committed = client.commit(instantTime, writeStatuses, HOption.of(checkpointMetadata))
Review Comment:
🤖 On the failure path, `startCommit` has already created a requested/inflight instant. If `upsert` or `commit` throws, or `commit` returns false and we throw here, that pending instant is left on the timeline with no rollback. Would it be worth calling `client.rollback(instantTime)` (or rolling back failed writes) in a catch before rethrowing, so a failed checkpoint-set doesn't leave a dangling commit for the next writer to clean up?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+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
+}
+
+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]
+ 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 table services as a
+ // side effect or clean up pending writes belonging to another writer.
+ HoodieCleanConfig.AUTO_CLEAN.key -> "false",
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key -> HoodieFailedWritesCleaningPolicy.NEVER.name,
+ HoodieArchivalConfig.AUTO_ARCHIVE.key -> "false",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieClusteringConfig.INLINE_CLUSTERING.key -> "false",
+ HoodieClusteringConfig.SCHEDULE_INLINE_CLUSTERING.key -> "false",
+ HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key -> "true",
+ // 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
+ try {
+ client = HoodieCLIUtils.createHoodieWriteClient(
+ sparkSession,
+ basePath,
+ writeOptions,
+ tableName.map(_.asInstanceOf[String]))
+ .asInstanceOf[SparkRDDWriteClient[AnyRef]]
+
+ val instantTime = client.startCommit(metaClient.getCommitActionType)
Review Comment:
🤖 Since this publishes a real empty commit through the standard write client without an OCC/lock provider configured, what happens if a Streamer (or other writer) is actively ingesting into the same table when this runs? Could the two commits race, or could the next ingest commit immediately overwrite the checkpoint we just set? It might be worth documenting that this is intended for use while ingestion is paused, or guarding against a concurrent active writer.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/DeltastreamerCheckpointProcedure.scala:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.{HoodieArchivalConfig, HoodieCleanConfig, HoodieClusteringConfig, HoodieCompactionConfig, 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
+
+class GetDeltastreamerCheckpointProcedure extends BaseProcedure with ProcedureBuilder {
+ import DeltastreamerCheckpointProcedureUtils._
Review Comment:
🤖 nit: the product name is `DeltaStreamer` (capital S) throughout the Hudi codebase — could you rename these to `GetDeltaStreamerCheckpointProcedure` / `SetDeltaStreamerCheckpointProcedure` (and the utils object / file name) for consistency?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19631:
URL: https://github.com/apache/hudi/pull/19631#issuecomment-5289366332
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/19631?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 77.44%. Comparing base ([`b2174b6`](https://app.codecov.io/gh/apache/hudi/commit/b2174b6f9ea30235cabf39d8269f179e9f670a87?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`699e554`](https://app.codecov.io/gh/apache/hudi/commit/699e554d10e9dededb4c67d7a7601a54da9a347c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 4 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19631 +/- ##
============================================
- Coverage 77.50% 77.44% -0.06%
- Complexity 32862 32886 +24
============================================
Files 2524 2524
Lines 139355 139553 +198
Branches 16747 16808 +61
============================================
+ Hits 108002 108074 +72
- Misses 23781 23883 +102
- Partials 7572 7596 +24
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.19% <ø> (-0.09%)` | :arrow_down: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.67% <ø> (-0.06%)` | :arrow_down: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.75% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.56% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.64% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `15.26% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.50% <ø> (-0.45%)` | :arrow_down: |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.58% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.47% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.44% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `64.06% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/19631/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/19631/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/19631/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.85% <ø> (+<0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/19631?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...sql/hudi/command/procedures/HoodieProcedures.scala](https://app.codecov.io/gh/apache/hudi/pull/19631?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2Fhudi%2Fcommand%2Fprocedures%2FHoodieProcedures.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9zcGFyay9zcWwvaHVkaS9jb21tYW5kL3Byb2NlZHVyZXMvSG9vZGllUHJvY2VkdXJlcy5zY2FsYQ==) | `97.61% <ø> (ø)` | |
... and [3 files with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/19631/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19631:
URL: https://github.com/apache/hudi/pull/19631#issuecomment-5289314962
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 699e554d10e9dededb4c67d7a7601a54da9a347c UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19631:
URL: https://github.com/apache/hudi/pull/19631#issuecomment-5365407821
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16286",
"triggerID" : "699e554d10e9dededb4c67d7a7601a54da9a347c",
"triggerType" : "PUSH"
}, {
"hash" : "be03dc0e9acf150ca2415a59ba4e682142f71382",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16437",
"triggerID" : "be03dc0e9acf150ca2415a59ba4e682142f71382",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* be03dc0e9acf150ca2415a59ba4e682142f71382 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16437)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): add DeltaStreamer checkpoint procedures [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 merged PR #19631:
URL: https://github.com/apache/hudi/pull/19631
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org