This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 4cfba4ca4079 fix(HUDI-9666): read the table's timeline layout in
upgrade_table, hudi-cli upgrade and rollback_to_instant (#20063)
4cfba4ca4079 is described below
commit 4cfba4ca4079c4abd09348f0826374db78c8ab35
Author: Aditya Goenka <[email protected]>
AuthorDate: Fri Oct 9 03:03:41 2026 +0530
fix(HUDI-9666): read the table's timeline layout in upgrade_table, hudi-cli
upgrade and rollback_to_instant (#20063)
---
.../org/apache/hudi/cli/commands/SparkMain.java | 2 -
.../java/org/apache/hudi/keygen/KeyGenUtils.java | 24 +++++++++++-
.../org/apache/hudi/keygen/TestKeyGenUtils.java | 33 +++++++++++++++++
.../RollbackToInstantTimeProcedure.scala | 3 --
.../procedures/UpgradeOrDowngradeProcedure.scala | 3 --
.../sql/hudi/procedure/TestCallProcedure.scala | 32 ++++++++++++++++
.../TestUpgradeOrDowngradeProcedure.scala | 43 +++++++++++++++++++++-
7 files changed, 130 insertions(+), 10 deletions(-)
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
index b8927c84d566..5eba65dc6051 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
@@ -32,7 +32,6 @@ import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.PartitionPathEncodeUtils;
@@ -729,7 +728,6 @@ public class SparkMain {
.setConf(HadoopFSUtils.getStorageConfWithCopy(jsc.hadoopConfiguration()))
.setBasePath(config.getBasePath())
.setLoadActiveTimelineOnLoad(false).setConsistencyGuardConfig(config.getConsistencyGuardConfig())
- .setLayoutVersion(Option.of(new
TimelineLayoutVersion(config.getTimelineLayoutVersion())))
.setFileSystemRetryConfig(config.getFileSystemRetryConfig()).build();
HoodieWriteConfig updatedConfig =
HoodieWriteConfig.newBuilder().withProps(config.getProps())
.forTable(metaClient.getTableConfig().getTableName()).build();
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java
index 18bc64528121..c2087d6a56a1 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java
@@ -645,12 +645,17 @@ public class KeyGenUtils {
*
* @return {@link ComplexKeyGenEncoding#FIELD_PREFIXED} for a table that was
never written to, the encoding
* read from the first readable base or log file among the most recent
commits otherwise, or empty when none
- * of the commits inspected yields a record key
+ * of the commits inspected yields a record key, or when the timeline was
moved by an unfinished upgrade
*/
public static Option<ComplexKeyGenEncoding>
deduceComplexKeyGenEncodingFromData(HoodieTableMetaClient metaClient) {
String expectedPrefix =
metaClient.getTableConfig().getRecordKeyFields().get()[0] +
DEFAULT_COLUMN_VALUE_SEPARATOR;
HoodieTimeline completedTimeline =
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
if (completedTimeline.empty()) {
+ if (hasTimelineOfUnfinishedLayoutUpgrade(metaClient)) {
+ // The 7 to 8 hop moves the instants into the version 2 timeline
before hoodie.properties is rewritten, so a
+ // retry after a failed upgrade reads an empty version 1 timeline
although the table holds data.
+ return Option.empty();
+ }
// Nothing was ever written, so there is no stored key whose encoding
could differ from the canonical one.
return Option.of(ComplexKeyGenEncoding.FIELD_PREFIXED);
}
@@ -689,6 +694,23 @@ public class KeyGenUtils {
return keyGenProps;
}
+ /**
+ * Whether a table below version 8 already has the version 2 timeline
directory, which only the 7 to 8 hop of an
+ * upgrade that did not finish creates.
+ */
+ private static boolean
hasTimelineOfUnfinishedLayoutUpgrade(HoodieTableMetaClient metaClient) {
+ HoodieTableConfig tableConfig = metaClient.getTableConfig();
+ if
(tableConfig.getTableVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)) {
+ return false;
+ }
+ StoragePath timelinePath = new StoragePath(metaClient.getMetaPath(),
tableConfig.getStringOrDefault(HoodieTableConfig.TIMELINE_PATH));
+ try {
+ return metaClient.getStorage().exists(timelinePath);
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to check for the timeline directory
" + timelinePath, e);
+ }
+ }
+
private static List<HoodieWriteStat> getWriteStats(HoodieInstant instant,
HoodieTimeline timeline) {
try {
return TimelineUtils.getCommitMetadata(instant,
timeline).getWriteStats();
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java
index a8fd806405a9..d9ef0cb2f6f5 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -31,6 +32,7 @@ import org.apache.hudi.exception.HoodieKeyException;
import org.apache.hudi.keygen.constant.ComplexKeyGenEncoding;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.keygen.constant.KeyGeneratorType;
+import org.apache.hudi.storage.StoragePath;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
@@ -477,6 +479,37 @@ public class TestKeyGenUtils {
assertEquals(Option.of(ComplexKeyGenEncoding.VALUE_ONLY),
metaClient.getTableConfig().getComplexKeyGenEncoding());
}
+ /**
+ * A retry of an upgrade that died after the 7 to 8 hop had moved the
instants into the version 2 timeline reads an
+ * empty version 1 timeline. That must not be taken for a never-written
table: the encoding cannot be deduced.
+ */
+ @Test
+ void
testComplexKeyGenEncodingNotDeducedFromTimelineMovedByUnfinishedUpgrade(@TempDir
Path tempDir) throws IOException {
+ Properties props = new Properties();
+ props.put(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key(),
ComplexAvroKeyGenerator.class.getName());
+ props.put(HoodieTableConfig.RECORDKEY_FIELDS.key(), "id");
+ props.put(HoodieTableConfig.VERSION.key(),
String.valueOf(HoodieTableVersion.SIX.versionCode()));
+ props.put(HoodieTableConfig.TIMELINE_LAYOUT_VERSION.key(), "1");
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(
+ HoodieTestUtils.getDefaultStorageConf(), tempDir.toString(),
HoodieTableType.COPY_ON_WRITE, props);
+ HoodieTableConfig.delete(metaClient.getStorage(), metaClient.getMetaPath(),
+
Collections.singleton(HoodieTableConfig.COMPLEX_KEYGEN_ENCODING.key()));
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(HoodieTableVersion.SIX,
metaClient.getTableConfig().getTableVersion());
+ StoragePath timelinePath = new StoragePath(metaClient.getMetaPath(),
HoodieTableConfig.TIMELINE_PATH.defaultValue());
+ assertFalse(metaClient.getStorage().exists(timelinePath));
+
+ // no version 2 timeline: a table that was never written deduces the
default
+ assertEquals(Option.of(ComplexKeyGenEncoding.FIELD_PREFIXED),
KeyGenUtils.deduceComplexKeyGenEncodingFromData(metaClient));
+
+ // the version 2 timeline of an unfinished upgrade: nothing can be
deduced, so validation refuses to guess
+ metaClient.getStorage().createDirectory(timelinePath);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
assertFalse(KeyGenUtils.deduceComplexKeyGenEncodingFromData(metaClient).isPresent());
+ HoodieWriteConfig writeConfig =
HoodieWriteConfig.newBuilder().withPath(tempDir.toString()).build();
+ assertFalse(KeyGenUtils.resolveComplexKeyGenEncodingForWrite(metaClient,
writeConfig).isPresent());
+ }
+
@Test
void testWithComplexKeyGenEncoding() {
TypedProperties props = new TypedProperties();
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RollbackToInstantTimeProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RollbackToInstantTimeProcedure.scala
index 6af5412f4da6..920ed31ac7e2 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RollbackToInstantTimeProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RollbackToInstantTimeProcedure.scala
@@ -20,8 +20,6 @@ package org.apache.spark.sql.hudi.command.procedures
import org.apache.hudi.HoodieCLIUtils
import org.apache.hudi.client.SparkRDDWriteClient
import org.apache.hudi.common.table.HoodieTableMetaClient
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion
-import org.apache.hudi.common.util.Option
import org.apache.hudi.config.HoodieWriteConfig.ROLLBACK_USING_MARKERS_ENABLE
import org.apache.hudi.exception.HoodieException
import org.apache.hudi.hadoop.fs.HadoopFSUtils
@@ -62,7 +60,6 @@ class RollbackToInstantTimeProcedure extends BaseProcedure
with ProcedureBuilder
.setBasePath(config.getBasePath)
.setLoadActiveTimelineOnLoad(false)
.setConsistencyGuardConfig(config.getConsistencyGuardConfig)
- .setLayoutVersion(Option.of(new
TimelineLayoutVersion(config.getTimelineLayoutVersion)))
.build
val activeTimeline = metaClient.getActiveTimeline
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/UpgradeOrDowngradeProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/UpgradeOrDowngradeProcedure.scala
index 039c0b7650cf..83459aead397 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/UpgradeOrDowngradeProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/UpgradeOrDowngradeProcedure.scala
@@ -21,8 +21,6 @@ import org.apache.hudi.HoodieCLIUtils
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
HoodieTableVersion}
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion
-import org.apache.hudi.common.util.Option
import org.apache.hudi.config.{HoodieCleanConfig, HoodieIndexConfig,
HoodieWriteConfig}
import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.index.HoodieIndex
@@ -63,7 +61,6 @@ class UpgradeOrDowngradeProcedure extends BaseProcedure with
ProcedureBuilder wi
.setBasePath(config.getBasePath)
.setLoadActiveTimelineOnLoad(false)
.setConsistencyGuardConfig(config.getConsistencyGuardConfig)
- .setLayoutVersion(Option.of(new
TimelineLayoutVersion(config.getTimelineLayoutVersion)))
.setFileSystemRetryConfig(config.getFileSystemRetryConfig)
.build
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala
index 9cd6402df591..b84601073002 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.hudi.procedure
import org.apache.hudi.common.model.IOType
import org.apache.hudi.common.testutils.FileCreateUtilsLegacy
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
class TestCallProcedure extends HoodieSparkProcedureTestBase {
@@ -133,6 +134,37 @@ class TestCallProcedure extends
HoodieSparkProcedureTestBase {
}
}
+ test("Test Call rollback_to_instant Procedure on a version 6 table") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ // a table a 1.x writer keeps at version 6, so its timeline is still on
the version 1 layout
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | orderingFields = 'ts',
+ | 'hoodie.write.table.version' = '6'
+ | )
+ """.stripMargin)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+ spark.sql(s"insert into $tableName select 2, 'a2', 20, 1500")
+ assertResult(6)(createMetaClient(spark,
tablePath).getTableConfig.getTableVersion.versionCode())
+
+ val commits = spark.sql(s"""call show_commits(table => '$tableName',
limit => 10)""").collect()
+ assertResult(2)(commits.length)
+ checkAnswer(s"""call rollback_to_instant(table => '$tableName',
instant_time => '${commits(0).get(0)}')""")(Seq(true))
+ assertResult(1)(spark.sql(s"""call show_commits(table => '$tableName',
limit => 10)""").collect().length)
+ }
+ }
+
test("Test Call rollback_to_instant Procedure with refreshTable") {
withTempDir { tmp =>
val tableName = generateTableName
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestUpgradeOrDowngradeProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestUpgradeOrDowngradeProcedure.scala
index 78a100cf3150..dc9ce8f2da77 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestUpgradeOrDowngradeProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestUpgradeOrDowngradeProcedure.scala
@@ -17,14 +17,20 @@
package org.apache.spark.sql.hudi.procedure
+import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.common.config.HoodieConfig
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
HoodieTableVersion}
+import org.apache.hudi.common.testutils.HoodieTestUtils
import
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME_GENERATOR
-import org.apache.hudi.common.util.{BinaryUtil, ConfigUtils, StringUtils}
+import org.apache.hudi.common.util.{BinaryUtil, ConfigUtils, Option =>
HOption, StringUtils}
+import org.apache.hudi.functional.ComplexKeyGenFixtures._
+import org.apache.hudi.keygen.constant.ComplexKeyGenEncoding
import org.apache.hudi.metadata.MetadataPartitionType
import org.apache.hudi.storage.StoragePath
+import org.apache.hudi.table.upgrade.TestUpgradeDowngrade.getFixtureName
import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.NAME_FORMAT_0_X
import org.junit.jupiter.api.Assertions.assertTrue
@@ -152,6 +158,41 @@ class TestUpgradeOrDowngradeProcedure extends
HoodieSparkProcedureTestBase {
}
}
+ /**
+ * The version 6 complex keygen fixtures predate the encoding property, so
`upgrade_table` has to read the
+ * table's data to record it. A table below version 8 is still on the
version 1 timeline layout.
+ */
+ test("Test upgrade_table on a version 6 single-field complex keygen table
records its key encoding") {
+ Seq((COMPLEX_KEYGEN_FIXTURE_SUFFIX, ComplexKeyGenEncoding.FIELD_PREFIXED),
+ (COMPLEX_KEYGEN_BARE_FIXTURE_SUFFIX,
ComplexKeyGenEncoding.VALUE_ONLY)).foreach { case (suffix, expectedEncoding) =>
+ withTempDir { tmp =>
+ val fixtureName = getFixtureName(HoodieTableVersion.SIX, suffix)
+ HoodieTestUtils.extractZipToDirectory(COMPLEX_KEYGEN_FIXTURES_PATH +
fixtureName, tmp.toPath, getClass)
+ val fixtureTableName = fixtureName.replace(".zip", "")
+ val tablePath = tmp.toPath.resolve(fixtureTableName).toString
+ val tableName = generateTableName
+ spark.sql(s"create table $tableName using hudi location '$tablePath'")
+ val storedKeys =
spark.read.format("hudi").load(tablePath).select("_hoodie_record_key").collect().map(_.getString(0)).sorted
+
+ checkAnswer(s"call upgrade_table(table => '$tableName', to_version =>
'${HoodieTableVersion.current().name()}')")(Seq(true))
+ val metaClient = createMetaClient(spark, tablePath)
+
assertResult(HoodieTableVersion.current())(metaClient.getTableConfig.getTableVersion)
+
assertResult(HOption.of(expectedEncoding))(metaClient.getTableConfig.getComplexKeyGenEncoding)
+
+ // upserting the existing records must update them in place, with the
keys they are stored under
+ val ts = 10000L // above every ordering value the fixture holds
+ fixtureRows(spark, FIXTURE_IDS, ts).write.format("hudi")
+ .options(fixtureWriteOpts(fixtureTableName))
+ .option(DataSourceWriteOptions.OPERATION.key,
DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(tablePath)
+ val afterUpsert = spark.read.format("hudi").load(tablePath)
+
assertResult(storedKeys.toSeq)(afterUpsert.select("_hoodie_record_key").collect().map(_.getString(0)).sorted.toSeq)
+ assertResult(FIXTURE_IDS.size)(afterUpsert.filter(s"ts = $ts").count())
+ }
+ }
+ }
+
test("Test downgrade table to version six") {
withTempDir { tmp =>
val tableName = generateTableName
