hudi-agent commented on code in PR #18696:
URL: https://github.com/apache/hudi/pull/18696#discussion_r3235414391
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java:
##########
@@ -894,7 +862,88 @@ public void restoreToSavepoint(String savepointTime) {
SavepointHelpers.validateSavepointRestore(table, savepointTime);
}
- @Deprecated
+ /**
+ * 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 MDT's penultimate completed
compaction (when at least
+ * two compactions exist). The restore would otherwise succeed for the
data table but
+ * {@code finishRestore} would fail to sync rollbacks into an MDT with
no base file at or
+ * before the target time.</li>
+ * <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 (nothing to delete or
worry about).
+ * Wraps any IOException reading the MDT in a {@link HoodieException} so
genuine permission /
+ * network failures surface to the caller instead of being silently
swallowed.
+ */
+ 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();
+ if (completedCompactions.size() >= 2) {
+ String penultimate =
completedCompactions.get(completedCompactions.size() - 2).requestedTime();
+ if (LESSER_THAN_OR_EQUALS.test(targetInstant, penultimate)) {
+ log.warn("Deleting MDT before restore to {}: target is at or before
penultimate MDT compaction {}",
+ targetInstant, penultimate);
+ return true;
+ }
+ }
+ 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) {
Review Comment:
🤖 If the MDT directory exists (so the line 886 check passes) but
`.hoodie/hoodie.properties` is missing or unreadable — e.g. from a partial MDT
init or an interrupted prior `deleteMetadataTable` — then
`HoodieTableMetaClient.builder().build()` on line 889 throws
`TableNotFoundException`, which extends `HoodieException` (RuntimeException),
not `IOException`. The old `catch (Exception e)` covered that case and let the
restore proceed; this narrowed `catch (IOException e)` won't, so a
corrupt-but-present MDT now hard-fails the restore. Is that the intended new
behavior, or should `TableNotFoundException` also be handled here (e.g. log and
treat as 'no usable MDT, fall through')?
<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/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
penultimate/oldest MDT compaction,
+ // pre-emptively delete the MDT so restoreToInstant does not leave it
inconsistent.
+ // deleteMetadataTableIfNecessaryBeforeRestore returns false when the
MDT was deleted
+ // (caller must not re-initialize it), true otherwise.
+ val shouldInitMdt = if (enableMetadata) {
+ client.deleteMetadataTableIfNecessaryBeforeRestore(targetInstant)
+ } else true
+ val restoreMetadata = client.restoreToInstant(targetInstant,
shouldInitMdt && enableMetadata)
Review Comment:
🤖 The MDT pre-check is now split across two call sites —
`restoreToSavepoint` invokes `shouldDeleteMdtBeforeRestore` internally, but
`restoreToInstant` does not, so the procedure has to remember to call
`deleteMetadataTableIfNecessaryBeforeRestore` here. The contract is documented
in the helper's Javadoc but not enforced, so any future Java-API caller of
`restoreToInstant` could forget and reintroduce the very bug this PR fixes. The
PR description also says `restoreToInstant` is "extended to invoke the helper,"
but the diff doesn't change `restoreToInstant`. Would it be cleaner to move the
pre-check inside `restoreToInstant` itself (gated on
`initialMetadataTableIfNecessary`), so all callers benefit uniformly?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java:
##########
@@ -894,7 +862,88 @@ public void restoreToSavepoint(String savepointTime) {
SavepointHelpers.validateSavepointRestore(table, savepointTime);
}
- @Deprecated
+ /**
+ * 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 MDT's penultimate completed
compaction (when at least
+ * two compactions exist). The restore would otherwise succeed for the
data table but
+ * {@code finishRestore} would fail to sync rollbacks into an MDT with
no base file at or
+ * before the target time.</li>
+ * <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 (nothing to delete or
worry about).
+ * Wraps any IOException reading the MDT in a {@link HoodieException} so
genuine permission /
+ * network failures surface to the caller instead of being silently
swallowed.
+ */
+ 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();
+ if (completedCompactions.size() >= 2) {
+ String penultimate =
completedCompactions.get(completedCompactions.size() - 2).requestedTime();
+ if (LESSER_THAN_OR_EQUALS.test(targetInstant, penultimate)) {
+ log.warn("Deleting MDT before restore to {}: target is at or before
penultimate MDT compaction {}",
+ targetInstant, penultimate);
+ return true;
+ }
+ }
+ 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);
+ }
+ }
+
+ /**
+ * 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 should be initialized
after the restore.
+ *
+ * <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 pass the return value as {@code
initialMetadataTableIfNecessary}:
+ *
+ * <pre>{@code
+ * boolean initMdt =
client.deleteMetadataTableIfNecessaryBeforeRestore(targetInstant);
+ * client.restoreToInstant(targetInstant, initMdt && enableMetadata);
+ * }</pre>
+ *
+ * @param targetInstant the instant the data table will be restored to
+ * @return {@code false} if the MDT was deleted (caller must not initialize
it post-restore);
+ * {@code true} otherwise (MDT either does not need deletion or does
not exist)
+ */
+ public boolean deleteMetadataTableIfNecessaryBeforeRestore(String
targetInstant) {
Review Comment:
🤖 nit: the return convention here is inverted from what a `delete*` method
name implies — `false` means "MDT was deleted, don't init it" and `true` means
"MDT is fine, init it as usual". The Scala caller already needed a three-line
comment to decode this. Could you either flip the return value (so `true =
deleted, skip init` reads naturally) and update callers, or rename the method
to something that hints at what the boolean means — e.g.
`shouldInitMetadataTableAfterRestore` (which deletes the MDT as a side-effect
when needed)?
<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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
