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