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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
