This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit dbfafbcf5738ce7c34c844492145530359c7a8ea Author: Mahsood Ebrahim <[email protected]> AuthorDate: Mon May 18 23:43:55 2026 -0700 feat(spark): add restore_to_instant stored procedure (#18696) * feat(spark): add restore_to_instant stored procedure Adds a Spark SQL stored procedure that performs a point-in-time table restore to any instant on the active timeline, with optional post-restore file-existence audit. Unlike rollback_to_savepoint, no savepoint is required at the target instant. Centralizes the MDT pre-check that was inlined in restoreToSavepoint into a new BaseHoodieWriteClient.shouldDeleteMdtBeforeRestore helper, and extends restoreToInstant to invoke it. The helper also catches the penultimate-compaction case (target at or before the second-most-recent MDT compaction) which the previous restoreToSavepoint inline check missed. IO/permission failures now surface as HoodieException instead of being silently swallowed. The audit returns a tri-state result (PASSED / FAILED / INCONCLUSIVE) so transient cloud-storage timeouts are distinguishable from real audit failures; an audit_only mode lets users re-audit a previously completed restore by passing its restore_instant_time. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * fix(spark): address PR review: public MDT guard, variable rename, test coverage Add a public deleteMetadataTableIfNecessaryBeforeRestore method to BaseHoodieWriteClient so callers that drive restore via restoreToInstant directly (e.g. the restore_to_instant procedure) can pre-check and pre-emptively delete the MDT before calling restoreToInstant. This closes the coverage gap identified in review: the procedure previously had no MDT protection after shouldDeleteMdtBeforeRestore was removed from restoreToInstant. The procedure now calls client.deleteMetadataTableIfNecessaryBeforeRestore before restoreToInstant and passes the returned boolean as initialMetadataTableIfNecessary, ensuring the penultimate/oldest compaction and timeline-start checks fire on the procedure path. Rename internal Scala variable restoreInstantTime -> startRestoreTimeArg for consistency with the start_restore_time parameter name. Strengthen testRestoreToInstantSkipsMdtCheckWhenMetadataDisabled: the test now verifies (a) deleteMetadataTableIfNecessaryBeforeRestore returns false and deletes the MDT for a target at/before the oldest compaction, and (b) restoreToInstant(target, false) then proceeds without invoking the guard. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * fix(spark): address PR review round 2 — remove penultimate check, rename public API Remove the penultimate-compaction check from shouldDeleteMdtBeforeRestore. The check fired too aggressively: it deleted the MDT even when the MDT could restore successfully, and then the fresh re-bootstrap failed for record_index-enabled tables (testRLIWithMDTCleaning) or triggered an inline MDT compaction during bootstrap that broke the isMetadataTableRecreatedDuringRestore detection (testRestoreToSavepointDeletesMdtWhenTargetIsBeforePenultimateCompaction). Only the oldest-compaction and timeline-start checks remain, matching the original pre-PR behaviour. Add a catch clause for HoodieException (e.g. TableNotFoundException from a partially initialized MDT directory) so a corrupt-but-present MDT is treated as absent rather than hard-failing the restore. Rename deleteMetadataTableIfNecessaryBeforeRestore -> deleteMdtIfNecessaryBeforeRestore and flip the return value so that true = MDT was deleted (callers negate with ! when passing to restoreToInstant). Update RestoreToInstantProcedure and the corresponding Java test accordingly. Restore the @Deprecated annotation on the two-arg rollback overload that was accidentally dropped in a prior commit. Remove testRestoreToSavepointDeletesMdtWhenTargetIsBeforePenultimateCompaction which tested the now-removed penultimate check. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> --------- Co-authored-by: mahsoode <[email protected]> Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]> (cherry picked from commit e406e5d092d090926803946445ede3e6ac261af0) --- .../apache/hudi/client/BaseHoodieWriteClient.java | 120 +++++--- .../TestSavepointRestoreCopyOnWrite.java | 89 ++++++ .../hudi/command/procedures/HoodieProcedures.scala | 1 + .../procedures/RestoreToInstantProcedure.scala | 309 +++++++++++++++++++++ .../sql/hudi/procedure/TestRestoreProcedure.scala | 258 +++++++++++++++++ 5 files changed, 739 insertions(+), 38 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java index df06df2fbbba..76ffe51a256d 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java @@ -92,6 +92,7 @@ import org.apache.hudi.metadata.HoodieTableMetadataUtil; import org.apache.hudi.metadata.HoodieTableMetadataWriter; import org.apache.hudi.metadata.MetadataPartitionType; import org.apache.hudi.metrics.HoodieMetrics; +import org.apache.hudi.storage.StoragePath; import org.apache.hudi.table.BulkInsertPartitioner; import org.apache.hudi.table.HoodieTable; import org.apache.hudi.table.action.HoodieWriteMetadata; @@ -846,44 +847,11 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient */ public void restoreToSavepoint(String savepointTime) { boolean initializeMetadataTableIfNecessary = config.isMetadataTableEnabled(); - if (initializeMetadataTableIfNecessary) { - try { - // Delete metadata table directly when users trigger savepoint rollback if mdt existed and if the savePointTime is beforeTimelineStarts - // or before the oldest compaction on MDT. - // We cannot restore to before the oldest compaction on MDT as we don't have the basefiles before that time. - HoodieTableMetaClient mdtMetaClient = HoodieTableMetaClient.builder() - .setConf(storageConf.newInstance()) - .setBasePath(getMetadataTableBasePath(config.getBasePath())).build(); - Option<HoodieInstant> oldestMdtCompaction = mdtMetaClient.getCommitTimeline().filterCompletedInstants().firstInstant(); - boolean deleteMDT = false; - if (oldestMdtCompaction.isPresent()) { - if (LESSER_THAN_OR_EQUALS.test(savepointTime, oldestMdtCompaction.get().requestedTime())) { - log.warn("Deleting MDT during restore to {} as the savepoint is older than oldest compaction {} on MDT", - savepointTime, oldestMdtCompaction.get().requestedTime()); - deleteMDT = true; - } - } - - // The instant required to sync rollback to MDT has been archived and the mdt syncing will be failed - // So that we need to delete the whole MDT here. - if (!deleteMDT) { - HoodieInstant syncedInstant = mdtMetaClient.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, savepointTime); - if (mdtMetaClient.getCommitsTimeline().isBeforeTimelineStarts(syncedInstant.requestedTime())) { - log.warn("Deleting MDT during restore to {} as the savepoint is older than the MDT timeline {}", - savepointTime, mdtMetaClient.getCommitsTimeline().firstInstant().get().requestedTime()); - deleteMDT = true; - } - } - - if (deleteMDT) { - HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context); - // rollbackToSavepoint action will try to bootstrap MDT at first but sync to MDT will fail at the current scenario. - // so that we need to disable metadata initialized here. - initializeMetadataTableIfNecessary = false; - } - } catch (Exception e) { - // Metadata directory does not exist - } + if (initializeMetadataTableIfNecessary && shouldDeleteMdtBeforeRestore(savepointTime)) { + HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context); + // rollbackToSavepoint action will try to bootstrap MDT at first but sync to MDT will fail at the current scenario. + // so that we need to disable metadata initialized here. + initializeMetadataTableIfNecessary = false; } HoodieTable<T, I, K, O> table = initTable(WriteOperationType.UNKNOWN, Option.empty(), initializeMetadataTableIfNecessary); @@ -894,6 +862,82 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient SavepointHelpers.validateSavepointRestore(table, savepointTime); } + /** + * Decides whether the metadata table (MDT) must be deleted before restoring the data table to + * {@code targetInstant}. Returns true when restoring would leave the MDT in an inconsistent + * state, specifically when any of the following holds: + * <ol> + * <li>The target is at or before the oldest completed compaction. We cannot restore to before + * the oldest compaction because we don't have base files before that time.</li> + * <li>The target is before the MDT timeline start (the relevant history was archived away).</li> + * </ol> + * Returns false when the MDT directory does not exist or is not readable (nothing to delete or + * worry about). Wraps genuine IO failures ({@link IOException}) in a {@link HoodieException} + * so permission / network errors surface to the caller. + */ + protected boolean shouldDeleteMdtBeforeRestore(String targetInstant) { + String mdtBasePath = getMetadataTableBasePath(config.getBasePath()); + try { + // Cheap existence check first to avoid constructing an MDT meta client when there is no MDT. + if (!storage.exists(new StoragePath(mdtBasePath))) { + return false; + } + HoodieTableMetaClient mdtMetaClient = HoodieTableMetaClient.builder() + .setConf(storageConf.newInstance()) + .setBasePath(mdtBasePath).build(); + List<HoodieInstant> completedCompactions = mdtMetaClient.getCommitTimeline() + .filterCompletedInstants().getInstants(); + Option<HoodieInstant> oldestMdtCompaction = completedCompactions.isEmpty() + ? Option.empty() : Option.of(completedCompactions.get(0)); + if (oldestMdtCompaction.isPresent() + && LESSER_THAN_OR_EQUALS.test(targetInstant, oldestMdtCompaction.get().requestedTime())) { + log.warn("Deleting MDT before restore to {}: target is at or before oldest MDT compaction {}", + targetInstant, oldestMdtCompaction.get().requestedTime()); + return true; + } + if (mdtMetaClient.getCommitsTimeline().isBeforeTimelineStarts(targetInstant)) { + log.warn("Deleting MDT before restore to {}: target is before MDT timeline start", targetInstant); + return true; + } + return false; + } catch (IOException e) { + throw new HoodieException( + "Failed to inspect MDT at " + mdtBasePath + " before restore to " + targetInstant + + " - refusing to silently proceed without an MDT integrity check.", e); + } catch (HoodieException e) { + // MDT directory exists but is not usable (e.g. TableNotFoundException from a partially + // initialized MDT). Treat as absent: no deletion needed, let the restore proceed. + log.warn("MDT at {} is present but could not be read ({}); skipping pre-check.", + mdtBasePath, e.getMessage()); + return false; + } + } + + /** + * Deletes the metadata table (MDT) if it would be left in an inconsistent state by a restore + * to {@code targetInstant}, and returns whether the MDT was actually deleted. + * + * <p>Callers that drive restore via {@link #restoreToInstant} directly (e.g. the + * {@code restore_to_instant} stored procedure) should call this method before invoking + * {@code restoreToInstant} and suppress MDT initialization when it returns {@code true}: + * + * <pre>{@code + * boolean mdtDeleted = client.deleteMdtIfNecessaryBeforeRestore(targetInstant); + * client.restoreToInstant(targetInstant, !mdtDeleted && enableMetadata); + * }</pre> + * + * @param targetInstant the instant the data table will be restored to + * @return {@code true} if the MDT was deleted (caller must not re-initialize it); + * {@code false} otherwise (MDT either did not need deletion or does not exist) + */ + public boolean deleteMdtIfNecessaryBeforeRestore(String targetInstant) { + if (shouldDeleteMdtBeforeRestore(targetInstant)) { + HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context); + return true; + } + return false; + } + @Deprecated public boolean rollback(final String commitInstantTime) throws HoodieRollbackException { HoodieTable<T, I, K, O> table = initTable(WriteOperationType.UNKNOWN, Option.empty()); diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestSavepointRestoreCopyOnWrite.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestSavepointRestoreCopyOnWrite.java index 0be971630c34..5c3c83226073 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestSavepointRestoreCopyOnWrite.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestSavepointRestoreCopyOnWrite.java @@ -29,6 +29,8 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.storage.HoodieStorageUtils; +import org.apache.hudi.storage.StoragePath; import org.apache.hudi.table.HoodieSparkTable; import org.apache.hudi.testutils.HoodieClientTestBase; @@ -41,6 +43,7 @@ import java.util.Objects; import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR; import static org.apache.hudi.metadata.HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -217,4 +220,90 @@ public class TestSavepointRestoreCopyOnWrite extends HoodieClientTestBase { assertRowNumberEqualsTo(20); } } + + /** + * Two-assertion test for the MDT pre-check guard: + * <ol> + * <li>{@code deleteMdtIfNecessaryBeforeRestore(firstCommit)} returns {@code true} (and + * deletes the MDT) when the target is at or before the oldest MDT compaction, + * confirming the guard logic fires for this table setup.</li> + * <li>{@code restoreToInstant(firstCommit, false)} does NOT invoke the pre-check, so a + * caller that explicitly opts out of MDT integration is never surprised by an implicit + * MDT deletion.</li> + * </ol> + */ + @Test + void testRestoreToInstantSkipsMdtCheckWhenMetadataDisabled() throws Exception { + HoodieWriteConfig hoodieWriteConfig = getConfigBuilder(HoodieFailedWritesCleaningPolicy.LAZY) + .withRollbackUsingMarkers(true) + .withMetadataConfig(HoodieMetadataConfig.newBuilder() + .enable(true) + .withMaxNumDeltaCommitsBeforeCompaction(4) + .build()) + .build(); + try (SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig)) { + String firstCommit = null; + String prevInstant = HoodieTimeline.INIT_INSTANT_TS; + final int numRecords = 10; + // 5 inserts so the MDT goes through one compaction with maxDeltaCommits=4. + for (int i = 1; i <= 5; i++) { + String newCommitTime = WriteClientTestUtils.createNewInstantTime(); + insertBatch(hoodieWriteConfig, client, newCommitTime, prevInstant, numRecords, SparkRDDWriteClient::insert, + false, true, numRecords, numRecords * i, 1, Option.empty(), INSTANT_GENERATOR); + prevInstant = newCommitTime; + if (i == 1) { + firstCommit = newCommitTime; + } + } + assertRowNumberEqualsTo(50); + + String mdtBasePath = HoodieTableMetadata.getMetadataTableBasePath(hoodieWriteConfig.getBasePath()); + assertTrue(HoodieStorageUtils.getStorage(mdtBasePath, storageConf).exists(new StoragePath(mdtBasePath)), + "MDT directory should exist before any pre-check"); + + // Assertion 1: deleteMdtIfNecessaryBeforeRestore detects that firstCommit is at or before + // the oldest MDT compaction, deletes the MDT, and returns true. + boolean mdtDeleted = client.deleteMdtIfNecessaryBeforeRestore( + Objects.requireNonNull(firstCommit, "first commit should not be null")); + assertTrue(mdtDeleted, + "deleteMdtIfNecessaryBeforeRestore should return true when target is at or before the oldest MDT compaction"); + assertFalse(HoodieStorageUtils.getStorage(mdtBasePath, storageConf).exists(new StoragePath(mdtBasePath)), + "MDT directory should have been deleted by deleteMdtIfNecessaryBeforeRestore"); + + // Assertion 2: restoreToInstant with initialMetadataTableIfNecessary=false does NOT invoke + // the pre-check — the MDT (now absent) is not touched. The restore proceeds without MDT. + client.restoreToInstant(firstCommit, false); + assertRowNumberEqualsTo(numRecords); + } + } + + /** + * Regression coverage for the {@code restoreToSavepoint} refactor. After replacing the inline + * MDT pre-check with a call to the centralized helper, the common case (target after the + * oldest MDT compaction, no MDT delete needed) must still work end-to-end. + */ + @Test + void testRestoreToSavepointStillWorksAfterRefactor() throws Exception { + HoodieWriteConfig hoodieWriteConfig = getConfigBuilder(HoodieFailedWritesCleaningPolicy.LAZY) + .withRollbackUsingMarkers(true) + .build(); + try (SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig)) { + String savepointCommit = null; + String prevInstant = HoodieTimeline.INIT_INSTANT_TS; + final int numRecords = 10; + for (int i = 1; i <= 4; i++) { + String newCommitTime = WriteClientTestUtils.createNewInstantTime(); + insertBatch(hoodieWriteConfig, client, newCommitTime, prevInstant, numRecords, SparkRDDWriteClient::insert, + false, true, numRecords, numRecords * i, 1, Option.empty(), INSTANT_GENERATOR); + prevInstant = newCommitTime; + if (i == 2) { + savepointCommit = newCommitTime; + client.savepoint("user1", "Savepoint for 2nd commit"); + } + } + assertRowNumberEqualsTo(40); + client.restoreToSavepoint(Objects.requireNonNull(savepointCommit, "restore commit should not be null")); + assertRowNumberEqualsTo(20); + } + } } diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala index 4319ce77c1c2..027c9f8c2882 100644 --- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala +++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala @@ -39,6 +39,7 @@ object HoodieProcedures { ,(DeleteSavepointProcedure.NAME, DeleteSavepointProcedure.builder) ,(RollbackToSavepointProcedure.NAME, RollbackToSavepointProcedure.builder) ,(RollbackToInstantTimeProcedure.NAME, RollbackToInstantTimeProcedure.builder) + ,(RestoreToInstantProcedure.NAME, RestoreToInstantProcedure.builder) ,(RunClusteringProcedure.NAME, RunClusteringProcedure.builder) ,(ShowClusteringProcedure.NAME, ShowClusteringProcedure.builder) ,(ShowCommitsProcedure.NAME, ShowCommitsProcedure.builder) diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RestoreToInstantProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RestoreToInstantProcedure.scala new file mode 100644 index 000000000000..ca5a2fc924ae --- /dev/null +++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RestoreToInstantProcedure.scala @@ -0,0 +1,309 @@ +/* + * 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.avro.model.HoodieRestoreMetadata +import org.apache.hudi.client.SparkRDDWriteClient +import org.apache.hudi.common.config.HoodieMetadataConfig +import org.apache.hudi.common.fs.ConsistencyGuardConfig +import org.apache.hudi.common.table.HoodieTableMetaClient +import org.apache.hudi.common.table.timeline.HoodieInstant +import org.apache.hudi.config.HoodieWriteConfig +import org.apache.hudi.exception.HoodieException +import org.apache.hudi.hadoop.fs.HadoopFSUtils +import org.apache.hudi.storage.StoragePath + +import org.apache.hadoop.fs.Path +import org.apache.spark.internal.Logging +import org.apache.spark.sql.Row +import org.apache.spark.sql.hudi.command.procedures.RestoreToInstantProcedure._ +import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType} + +import java.util.function.Supplier + +import scala.collection.JavaConverters._ + +/** + * Stored procedure to perform a full point-in-time table restore to a given instant. + * + * Unlike [[RollbackToSavepointProcedure]] (which requires a savepoint at the target instant), + * this procedure calls restoreToInstant() directly and works on any arbitrary instant on the + * active timeline. + * + * Parameters: + * - table / path: identifies the Hudi table (one must be provided) + * - instant_time: target commit to restore to (required when audit_only=false; must be omitted + * when audit_only=true) + * - start_restore_time: the restore operation's own timeline timestamp (the start_restore_time + * value returned by a prior restore_to_instant call). Required when + * audit_only=true; must be omitted otherwise. + * - enable_metadata: whether the metadata table is enabled (default: true) + * - rollback_parallelism: Spark parallelism for rollback and audit operations (default: 4) + * - enable_consistency_guard: enable consistency guard for file existence checks (default: false) + * - audit_post_restore: after restoring, verify that all successfully-deleted files are absent (default: false) + * - audit_only: skip the restore and only audit a previously completed restore instant (default: false) + * + * Output columns: + * - restore_result: true if restore succeeded; null if audit_only=true + * - start_restore_time: the restore operation's own timeline timestamp; null if audit_only=true. + * Pass this value as start_restore_time to re-run the audit later. + * - time_taken_in_millis: restore duration; null if audit_only=true + * - instants_rolled_back: number of commits rolled back; null if audit_only=true + * - audit_result: one of "PASSED" / "FAILED" / "INCONCLUSIVE" when an audit ran; null otherwise. + * INCONCLUSIVE means at least one file existence check threw an IOException + * (e.g. transient cloud-storage timeout) — re-run audit_only=true to retry. + */ +class RestoreToInstantProcedure extends BaseProcedure with ProcedureBuilder with Logging { + + private val PARAMETERS = Array[ProcedureParameter]( + ProcedureParameter.optional(0, "table", DataTypes.StringType), + ProcedureParameter.optional(1, "instant_time", DataTypes.StringType), + ProcedureParameter.optional(2, "enable_metadata", DataTypes.BooleanType, true), + ProcedureParameter.optional(3, "rollback_parallelism", DataTypes.IntegerType, 4), + ProcedureParameter.optional(4, "enable_consistency_guard", DataTypes.BooleanType, false), + ProcedureParameter.optional(5, "audit_post_restore", DataTypes.BooleanType, false), + ProcedureParameter.optional(6, "audit_only", DataTypes.BooleanType, false), + ProcedureParameter.optional(7, "path", DataTypes.StringType), + ProcedureParameter.optional(8, "start_restore_time", DataTypes.StringType) + ) + + private val OUTPUT_TYPE = new StructType(Array[StructField]( + StructField("restore_result", DataTypes.BooleanType, nullable = true, Metadata.empty), + StructField("start_restore_time", DataTypes.StringType, nullable = true, Metadata.empty), + StructField("time_taken_in_millis", DataTypes.LongType, nullable = true, Metadata.empty), + StructField("instants_rolled_back", DataTypes.LongType, nullable = true, Metadata.empty), + StructField("audit_result", DataTypes.StringType, nullable = true, Metadata.empty) + )) + + def parameters: Array[ProcedureParameter] = PARAMETERS + + def outputType: StructType = OUTPUT_TYPE + + override def call(args: ProcedureArgs): Seq[Row] = { + super.checkArgs(PARAMETERS, args) + + val tableName = getArgValueOrDefault(args, PARAMETERS(0)) + val instantTime = getArgValueOrDefault(args, PARAMETERS(1)) + val enableMetadata = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Boolean] + val rollbackParallelism = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Int] + val enableConsistencyGuard = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[Boolean] + val shouldAuditPostRestore = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[Boolean] + val auditOnly = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[Boolean] + val tablePath = getArgValueOrDefault(args, PARAMETERS(7)) + val startRestoreTimeArg = getArgValueOrDefault(args, PARAMETERS(8)) + + // Cross-validation: each of (instant_time, start_restore_time) has one unambiguous meaning. + if (!auditOnly && instantTime.isEmpty) { + throw new HoodieException("instant_time is required when audit_only=false.") + } + if (auditOnly && startRestoreTimeArg.isEmpty) { + throw new HoodieException( + "start_restore_time is required when audit_only=true. " + + "Pass the start_restore_time value from a prior restore_to_instant call.") + } + if (!auditOnly && startRestoreTimeArg.isDefined) { + throw new HoodieException("start_restore_time may only be specified when audit_only=true.") + } + if (auditOnly && instantTime.isDefined) { + throw new HoodieException( + "instant_time may only be specified when audit_only=false. " + + "Use start_restore_time to identify a previously executed restore.") + } + if (auditOnly && shouldAuditPostRestore) { + logWarning("Both audit_only and audit_post_restore are set. Only audit_only will be honored.") + } + + val basePath = getBasePath(tableName, tablePath) + + val confs = Map( + HoodieMetadataConfig.ENABLE.key() -> enableMetadata.toString, + HoodieWriteConfig.ROLLBACK_PARALLELISM_VALUE.key() -> rollbackParallelism.toString, + HoodieWriteConfig.ROLLBACK_USING_MARKERS_ENABLE.key() -> "false" + ) ++ (if (enableConsistencyGuard) Map(ConsistencyGuardConfig.ENABLE.key() -> "true") else Map.empty) + + val metaClient = createMetaClient(jsc, basePath) + + // Nullable boxed types so Row can hold null for audit_only runs + var restoreResult: java.lang.Boolean = null + var startRestoreTime: String = null + var timeTakenInMillis: java.lang.Long = null + var instantsRolledBack: java.lang.Long = null + + if (!auditOnly) { + val targetInstant = instantTime.get.asInstanceOf[String] + var client: SparkRDDWriteClient[_] = null + try { + client = HoodieCLIUtils.createHoodieWriteClient(sparkSession, basePath, confs, + tableName.asInstanceOf[Option[String]]) + // Pre-check: if the target instant is at or before the oldest MDT compaction or before the + // MDT timeline start, pre-emptively delete the MDT so restoreToInstant does not leave it + // inconsistent. deleteMdtIfNecessaryBeforeRestore returns true when the MDT was deleted + // (caller must not re-initialize it), false otherwise. + val mdtDeleted = if (enableMetadata) { + client.deleteMdtIfNecessaryBeforeRestore(targetInstant) + } else false + val restoreMetadata = client.restoreToInstant(targetInstant, !mdtDeleted && enableMetadata) + restoreResult = true + startRestoreTime = restoreMetadata.getStartRestoreTime + timeTakenInMillis = restoreMetadata.getTimeTakenInMillis + instantsRolledBack = restoreMetadata.getInstantsToRollback.size().toLong + } finally { + if (client != null) { + client.close() + } + } + if (tableName.isDefined) { + spark.catalog.refreshTable(tableName.get.asInstanceOf[String]) + } + // getActiveTimeline is lazily cached; reload so the new .restore instant is visible to the audit. + if (shouldAuditPostRestore) { + metaClient.reloadActiveTimeline() + } + } + + var auditResult: String = null + if (auditOnly || shouldAuditPostRestore) { + val restoreInstant: HoodieInstant = if (auditOnly) { + val ts = startRestoreTimeArg.get.asInstanceOf[String] + val instants = metaClient.getActiveTimeline.getRestoreTimeline.filterCompletedInstants + .getInstants.asScala + instants.find(_.requestedTime().equals(ts)).getOrElse( + throw new HoodieException(s"No completed restore instant found for $ts. " + + "Pass the start_restore_time from a prior restore_to_instant call as start_restore_time.") + ) + } else { + // Use startRestoreTime captured from the restore metadata — more deterministic than + // lastInstant() since another concurrent restore could otherwise land in between. + val ts = startRestoreTime + val instants = metaClient.getActiveTimeline.getRestoreTimeline.filterCompletedInstants + .getInstants.asScala + instants.find(_.requestedTime().equals(ts)).getOrElse( + throw new HoodieException(s"No completed restore instant found for $ts after restore.") + ) + } + auditResult = auditPostRestore(metaClient, basePath, restoreInstant, rollbackParallelism) + } + + Seq(Row(restoreResult, startRestoreTime, timeTakenInMillis, instantsRolledBack, auditResult)) + } + + /** + * Verifies that all files expected to have been deleted by a restore operation are actually + * absent from storage. Returns one of "PASSED", "FAILED", or "INCONCLUSIVE": + * - PASSED: every file expected to be absent is in fact absent. + * - FAILED: at least one file is still present after restore. + * - INCONCLUSIVE: no file was confirmed present, but at least one existence check threw + * an IOException (e.g. transient cloud-storage timeout). Re-run with + * audit_only=true to retry. + */ + private def auditPostRestore( + metaClient: HoodieTableMetaClient, + basePath: String, + restoreInstant: HoodieInstant, + rollbackParallelism: Int): String = { + try { + val restoreMetadata: HoodieRestoreMetadata = + metaClient.getActiveTimeline.readRestoreMetadata(restoreInstant) + + val filesToCheck = new java.util.ArrayList[String]() + restoreMetadata.getHoodieRestoreMetadata.values.asScala.foreach { rollbackList => + rollbackList.asScala.foreach { rollback => + rollback.getPartitionMetadata.asScala.foreach { case (partition, pm) => + val partitionPathStr = basePath + Path.SEPARATOR + partition + val partitionStoragePath = new StoragePath(partitionPathStr) + if (!metaClient.getStorage.exists(partitionStoragePath)) { + logInfo(s"Partition path $partitionStoragePath does not exist. Skipping audit for its files.") + } else { + // Use .toString() to preserve the full absolute path. Using .getName() would strip + // the path to just the filename, making the subsequent FS.exists() check vacuously pass. + // Only check getSuccessDeleteFiles: these are the files the restore intended to delete + // and whose absence we can meaningfully verify. Files in getFailedDeleteFiles already + // failed to be deleted and are expected to still be present. + pm.getSuccessDeleteFiles.asScala.foreach(f => + filesToCheck.add(new Path(partitionPathStr, f).toString)) + } + } + } + } + + if (filesToCheck.isEmpty) { + logInfo(s"No files to audit for restore instant ${restoreInstant.requestedTime()}") + "PASSED" + } else { + // HadoopFSUtils.getStorageConfWithCopy returns a Serializable StorageConfiguration; the + // closure captures only storageConf, so Spark's ClosureCleaner can serialize it cleanly. + // HadoopFSUtils.getFs internally calls prepareHadoopConf, preserving HOODIE_ENV_* (e.g. + // S3A) injection at cloud DCs. + val storageConf = HadoopFSUtils.getStorageConfWithCopy(jsc.hadoopConfiguration()) + val outcomes = jsc.parallelize(filesToCheck, Math.max(1, rollbackParallelism)) + .map { pathStr => + val p = new Path(pathStr) + try { + val exists = HadoopFSUtils.getFs(p, storageConf.unwrap()).exists(p) + val status: AuditFileStatus = if (exists) Present else Absent + (pathStr, status) + } catch { + case e: Exception => (pathStr, IndeterminateError(e.getMessage)) + } + } + .collect() + .asScala + .toArray + + val present = outcomes.collect { case (p, Present) => p } + val indeterminate = outcomes.collect { case (p, IndeterminateError(msg)) => (p, msg) } + if (present.nonEmpty) { + logError(s"Restore audit FAILED for instant ${restoreInstant.requestedTime()}: " + + s"${present.length} file(s) still present, e.g. ${present.take(5).mkString(", ")}") + "FAILED" + } else if (indeterminate.nonEmpty) { + logWarning(s"Restore audit INCONCLUSIVE for instant ${restoreInstant.requestedTime()}: " + + s"${indeterminate.length} file(s) had IO errors, e.g. ${indeterminate.take(5).mkString(", ")}") + "INCONCLUSIVE" + } else { + logInfo(s"Restore audit PASSED for instant ${restoreInstant.requestedTime()}: " + + s"${outcomes.length} file(s) confirmed absent.") + "PASSED" + } + } + } catch { + case e: Exception => + logError(s"Exception during restore audit for instant ${restoreInstant.requestedTime()}", e) + "INCONCLUSIVE" + } + } + + override def build: Procedure = new RestoreToInstantProcedure() +} + +object RestoreToInstantProcedure { + val NAME: String = "restore_to_instant" + + def builder: Supplier[ProcedureBuilder] = new Supplier[ProcedureBuilder] { + override def get(): RestoreToInstantProcedure = new RestoreToInstantProcedure() + } + + // ADT for per-file audit outcomes. Defined at object level (NOT inside auditPostRestore) so the + // generated bytecode does not embed a synthetic outer-class reference; otherwise Spark's + // ClosureCleaner cannot serialize the closure that returns these values from executors. + private sealed trait AuditFileStatus extends Serializable + private case object Absent extends AuditFileStatus + private case object Present extends AuditFileStatus + private case class IndeterminateError(message: String) extends AuditFileStatus +} diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRestoreProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRestoreProcedure.scala new file mode 100644 index 000000000000..7674fa6abaa8 --- /dev/null +++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRestoreProcedure.scala @@ -0,0 +1,258 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.hudi.procedure + +class TestRestoreProcedure extends HoodieSparkProcedureTestBase { + + private def createTableAndInsertData(tableName: String, tablePath: String, tableType: String = "cow"): Array[String] = { + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | price double, + | ts long + |) using hudi + | location '$tablePath' + | tblproperties ( + | primaryKey = 'id', + | preCombineField = 'ts', + | type = '$tableType' + | ) + """.stripMargin) + spark.sql(s"insert into $tableName select 1, 'a1', 10.0, 1000") + spark.sql(s"insert into $tableName select 2, 'a2', 20.0, 1500") + spark.sql(s"insert into $tableName select 3, 'a3', 30.0, 2000") + spark.sql(s"insert into $tableName select 4, 'a4', 40.0, 2500") + spark.sql(s"call show_commits(table => '$tableName')").collect() + .map(_.getString(0)).sorted + } + + test("Test restore_to_instant basic CoW") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + val commits = createTableAndInsertData(tableName, tablePath) + assertResult(4)(commits.length) + + // restore to after the 2nd commit — only commits(0) and commits(1) should remain + val result = spark.sql( + s"call restore_to_instant(table => '$tableName', instant_time => '${commits(1)}')" + ).collect() + + assertResult(1)(result.length) + assertResult(true)(result(0).getBoolean(0)) // restore_result + assert(result(0).getString(1) != null) // start_restore_time (dynamic timestamp) + assert(result(0).getLong(2) >= 0) // time_taken_in_millis + assert(result(0).getLong(3) >= 0L) // instants_rolled_back + assertResult(true)(result(0).isNullAt(4)) // audit_result = null (audit not requested) + + // verify data reverted to 2 records + val count = spark.sql(s"select count(*) from $tableName").collect()(0).getLong(0) + assertResult(2)(count) + } + } + + test("Test restore_to_instant basic MoR") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + val commits = createTableAndInsertData(tableName, tablePath, tableType = "mor") + assertResult(4)(commits.length) + + val result = spark.sql( + s"call restore_to_instant(table => '$tableName', instant_time => '${commits(1)}')" + ).collect() + + assertResult(1)(result.length) + assertResult(true)(result(0).getBoolean(0)) + assert(result(0).getString(1) != null) + assert(result(0).getLong(2) >= 0) + assert(result(0).getLong(3) >= 0L) + assertResult(true)(result(0).isNullAt(4)) + + val count = spark.sql(s"select count(*) from $tableName").collect()(0).getLong(0) + assertResult(2)(count) + } + } + + test("Test restore_to_instant using path parameter") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + val commits = createTableAndInsertData(tableName, tablePath) + assertResult(4)(commits.length) + + val result = spark.sql( + s"call restore_to_instant(path => '$tablePath', instant_time => '${commits(1)}')" + ).collect() + + assertResult(1)(result.length) + assertResult(true)(result(0).getBoolean(0)) + assert(result(0).isNullAt(4)) + + val count = spark.sql(s"select count(*) from $tableName").collect()(0).getLong(0) + assertResult(2)(count) + } + } + + test("Test restore_to_instant with audit_post_restore") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + val commits = createTableAndInsertData(tableName, tablePath) + assertResult(4)(commits.length) + + val result = spark.sql( + s"""call restore_to_instant( + | table => '$tableName', + | instant_time => '${commits(1)}', + | audit_post_restore => true + |)""".stripMargin + ).collect() + + assertResult(1)(result.length) + assertResult(true)(result(0).getBoolean(0)) // restore_result + assert(result(0).getString(1) != null) // start_restore_time + // audit_result should be "PASSED": all rolled-back files are absent + assertResult(false)(result(0).isNullAt(4)) + assertResult("PASSED")(result(0).getString(4)) + + val count = spark.sql(s"select count(*) from $tableName").collect()(0).getLong(0) + assertResult(2)(count) + } + } + + test("Test restore_to_instant audit_only mode") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + val commits = createTableAndInsertData(tableName, tablePath) + assertResult(4)(commits.length) + + // Round 1: perform the restore and capture the restore operation's own timeline timestamp + val restoreRows = spark.sql( + s"call restore_to_instant(table => '$tableName', instant_time => '${commits(1)}')" + ).collect() + assertResult(true)(restoreRows(0).getBoolean(0)) + // start_restore_time is the restore instant's timeline timestamp — NOT commits(1) + val restoreInstantTs = restoreRows(0).getString(1) + assert(restoreInstantTs != null) + assert(restoreInstantTs != commits(1)) + + // Round 2: audit_only using start_restore_time (the original target commit is not needed) + val auditRows = spark.sql( + s"""call restore_to_instant( + | table => '$tableName', + | audit_only => true, + | start_restore_time => '$restoreInstantTs' + |)""".stripMargin + ).collect() + + assertResult(1)(auditRows.length) + assertResult(true)(auditRows(0).isNullAt(0)) // restore_result = null (no restore performed) + assertResult(true)(auditRows(0).isNullAt(1)) // start_restore_time = null + assertResult(true)(auditRows(0).isNullAt(2)) // time_taken_in_millis = null + assertResult(true)(auditRows(0).isNullAt(3)) // instants_rolled_back = null + assertResult(false)(auditRows(0).isNullAt(4)) // audit_result is present + assertResult("PASSED")(auditRows(0).getString(4)) + } + } + + test("Test restore_to_instant audit_only with non-existent restore instant throws") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + createTableAndInsertData(tableName, tablePath) + + // start_restore_time pointing at a timestamp that has never been a restore instant. + assertThrows[Exception] { + spark.sql( + s"""call restore_to_instant( + | table => '$tableName', + | audit_only => true, + | start_restore_time => '19700101000000000' + |)""".stripMargin + ).collect() + } + } + } + + test("Test restore_to_instant cross-validation: audit_only=true requires start_restore_time") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + createTableAndInsertData(tableName, tablePath) + + assertThrows[Exception] { + spark.sql( + s"call restore_to_instant(table => '$tableName', audit_only => true)" + ).collect() + } + } + } + + test("Test restore_to_instant cross-validation: audit_only=false rejects start_restore_time") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + val commits = createTableAndInsertData(tableName, tablePath) + + assertThrows[Exception] { + spark.sql( + s"""call restore_to_instant( + | table => '$tableName', + | instant_time => '${commits(1)}', + | start_restore_time => '20990101000000000' + |)""".stripMargin + ).collect() + } + } + } + + test("Test restore_to_instant cross-validation: audit_only=true rejects instant_time") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + val commits = createTableAndInsertData(tableName, tablePath) + + assertThrows[Exception] { + spark.sql( + s"""call restore_to_instant( + | table => '$tableName', + | audit_only => true, + | instant_time => '${commits(1)}', + | start_restore_time => '20990101000000000' + |)""".stripMargin + ).collect() + } + } + } + + test("Test restore_to_instant cross-validation: audit_only=false requires instant_time") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = tmp.getCanonicalPath + "/" + tableName + createTableAndInsertData(tableName, tablePath) + + assertThrows[Exception] { + spark.sql(s"call restore_to_instant(table => '$tableName')").collect() + } + } + } +}
