This is an automated email from the ASF dual-hosted git repository.
danny0405 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 a1dbb5620bd1 feat: Support table version 10 to 9 downgrade (#19199)
a1dbb5620bd1 is described below
commit a1dbb5620bd126f46fb48a51287b59bbdc3899d8
Author: Shuo Cheng <[email protected]>
AuthorDate: Tue Jul 7 19:17:56 2026 +0800
feat: Support table version 10 to 9 downgrade (#19199)
* feat: Support table version 10 to 9 downgrade
---
.../table/upgrade/TenToNineDowngradeHandler.java | 42 +++++
.../hudi/table/upgrade/UpgradeDowngrade.java | 3 +
.../upgrade/TestTenToNineDowngradeHandler.java | 50 ++++++
.../java/org/apache/hudi/common/fs/FSUtils.java | 4 +
.../sink/compact/ITTestHoodieFlinkCompactor.java | 3 +-
.../hudi/table/upgrade/TestUpgradeDowngrade.java | 94 +++++++++-
.../hudi/functional/ColumnStatIndexTestBase.scala | 15 ++
.../hudi/functional/TestColumnStatsIndex.scala | 1 +
.../hudi/functional/TestEightToNineUpgrade.scala | 1 -
.../functional/TestPayloadDeprecationFlow.scala | 32 ++--
.../hudi/functional/TestSevenToEightUpgrade.scala | 8 +-
.../others/TestMergeModeCommitTimeOrdering.scala | 3 +-
.../others/TestMergeModeEventTimeOrdering.scala | 10 +-
.../dml/others/TestPartialUpdateForMergeInto.scala | 7 +-
.../hudi/feature/index/TestSecondaryIndex.scala | 191 ++++++++++-----------
.../TestUpgradeOrDowngradeProcedure.scala | 4 +-
16 files changed, 331 insertions(+), 137 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/TenToNineDowngradeHandler.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/TenToNineDowngradeHandler.java
new file mode 100644
index 000000000000..393dd6986d82
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/TenToNineDowngradeHandler.java
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.table.upgrade;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.Collections;
+
+/**
+ * Version 10 writes native log files by default. Downgrading to version 9
requires
+ * full compaction of native data/delete logs before the downgrade completes.
+ */
+public class TenToNineDowngradeHandler implements DowngradeHandler {
+ @Override
+ public UpgradeDowngrade.TableConfigChangeSet downgrade(
+ HoodieWriteConfig config,
+ HoodieEngineContext context,
+ String instantTime,
+ SupportsUpgradeDowngrade upgradeDowngradeHelper) {
+ return new UpgradeDowngrade.TableConfigChangeSet(
+ Collections.emptyMap(),
+ Collections.singleton(HoodieTableConfig.TABLE_STORAGE_LAYOUT));
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java
index a550fbe706b1..734d9a09e98f 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java
@@ -62,6 +62,7 @@ public class UpgradeDowngrade {
));
private static final Set<Pair<Integer, Integer>>
DOWNGRADE_HANDLERS_REQUIRING_ROLLBACK_ANDCOMPACT = new HashSet<>(Arrays.asList(
+ Pair.of(10, 9), // TenToNineDowngradeHandler
Pair.of(8, 7), // EightToSevenDowngradeHandler
Pair.of(9, 8), // NineToEightDowngradeHandler
Pair.of(6, 5) // SixToFiveDowngradeHandler
@@ -328,6 +329,8 @@ public class UpgradeDowngrade {
return new EightToSevenDowngradeHandler().downgrade(config, context,
instantTime, upgradeDowngradeHelper);
} else if (fromVersion == HoodieTableVersion.NINE && toVersion ==
HoodieTableVersion.EIGHT) {
return new NineToEightDowngradeHandler().downgrade(config, context,
instantTime, upgradeDowngradeHelper);
+ } else if (fromVersion == HoodieTableVersion.TEN && toVersion ==
HoodieTableVersion.NINE) {
+ return new TenToNineDowngradeHandler().downgrade(config, context,
instantTime, upgradeDowngradeHelper);
} else {
throw new HoodieUpgradeDowngradeException(fromVersion.versionCode(),
toVersion.versionCode(), false);
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestTenToNineDowngradeHandler.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestTenToNineDowngradeHandler.java
new file mode 100644
index 000000000000..12faa9736930
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestTenToNineDowngradeHandler.java
@@ -0,0 +1,50 @@
+/*
+ * 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.hudi.table.upgrade;
+
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableVersion;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestTenToNineDowngradeHandler {
+
+ @Test
+ void testDowngradeRemovesStorageLayoutOnly() {
+ UpgradeDowngrade.TableConfigChangeSet changeSet =
+ new TenToNineDowngradeHandler().downgrade(null, null, null, null);
+
+ assertTrue(changeSet.propertiesToUpdate().isEmpty());
+ assertEquals(1, changeSet.propertiesToDelete().size());
+
assertTrue(changeSet.propertiesToDelete().contains(HoodieTableConfig.TABLE_STORAGE_LAYOUT));
+ }
+
+ @Test
+ void testTenToNineDowngradeRouteIsSupported() {
+ UpgradeDowngrade.TableConfigChangeSet changeSet =
+ new UpgradeDowngrade(null, null, null, null)
+ .downgrade(HoodieTableVersion.TEN, HoodieTableVersion.NINE, "001");
+
+ assertEquals(1, changeSet.propertiesToDelete().size());
+
assertTrue(changeSet.propertiesToDelete().contains(HoodieTableConfig.TABLE_STORAGE_LAYOUT));
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
index 16aeb1074247..3102e3c42f99 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
@@ -548,6 +548,10 @@ public class FSUtils {
return matchNativeLogFile(fileName).map(matcher ->
"deletes".equals(matcher.group(8))).orElse(false);
}
+ public static boolean isNativeCDCLogFile(String fileName) {
+ return matchNativeLogFile(fileName).map(matcher ->
"cdc".equals(matcher.group(8))).orElse(false);
+ }
+
public static boolean isCDCLogFile(String fileName) {
if (StringUtils.isNullOrEmpty(fileName)) {
return false;
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
index a4e85e17dfd7..3ae8ea520f2c 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
@@ -195,9 +195,8 @@ public class ITTestHoodieFlinkCompactor {
}
}
- // todo: enable test for downgrade after this issue solved:
https://github.com/apache/hudi/issues/19090
@ParameterizedTest
- @ValueSource(booleans = {true})
+ @ValueSource(booleans = {true, false})
void testHoodieFlinkCompactorWithUpgradeAndDowngrade(boolean upgrade) throws
Exception {
// Create hoodie table and insert into data.
EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
index a2d327b7f624..da6b4ac2c12b 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
@@ -18,6 +18,7 @@
package org.apache.hudi.table.upgrade;
+import org.apache.hudi.DataSourceWriteOptions;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteClientTestUtils;
import org.apache.hudi.common.config.HoodieMetadataConfig;
@@ -35,6 +36,7 @@ import
org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieUpgradeDowngradeException;
import org.apache.hudi.keygen.constant.KeyGeneratorType;
@@ -410,8 +412,6 @@ public class TestUpgradeDowngrade extends
SparkClientFunctionalTestHarness {
Properties props = new Properties();
props.put(HoodieTableConfig.TYPE.key(), tableType.name());
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
- props.put(HoodieWriteConfig.WRITE_TABLE_VERSION.key(),
String.valueOf(fromVersion.versionCode()));
HoodieTableMetaClient metaClient =
getHoodieMetaClient(storageConf(), URI.create(basePath()).getPath(),
props);
@@ -474,6 +474,91 @@ public class TestUpgradeDowngrade extends
SparkClientFunctionalTestHarness {
assertTrue(resultMetaClient.getTableConfig().getMetadataPartitions().contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath()));
}
+ @Test
+ public void
testTenToNineDowngradeCompactsNativeLogsAndRemovesStorageLayout() throws
Exception {
+ String tablePath = URI.create(basePath()).getPath();
+ String tableName = "ten_to_nine_native_logs";
+
+ Dataset<Row> inserts = sqlContext().range(0, 100).selectExpr(
+ "cast(id as string) as id",
+ "cast(id as int) as ts",
+ "concat('rider-', cast(id as string)) as rider",
+ "concat('driver-', cast(id as string)) as driver",
+ "cast(id as double) as fare",
+ "'2021/09/11' as partition");
+
+ inserts.write()
+ .format("hudi")
+ .option(HoodieWriteConfig.TBL_NAME.key(), tableName)
+ .option(DataSourceWriteOptions.TABLE_TYPE().key(),
HoodieTableType.MERGE_ON_READ.name())
+ .option(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL())
+ .option(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "id")
+ .option(DataSourceWriteOptions.PRECOMBINE_FIELD().key(), "ts")
+ .option(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(),
"partition")
+ .option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(),
String.valueOf(HoodieTableVersion.TEN.versionCode()))
+ .option(HoodieWriteConfig.AUTO_UPGRADE_VERSION.key(), "false")
+ .option(HoodieMetadataConfig.ENABLE.key(), "false")
+ .option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false")
+ .option(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue())
+ .mode(SaveMode.Overwrite)
+ .save(tablePath);
+
+ Dataset<Row> updates = sqlContext().range(0, 50).selectExpr(
+ "cast(id as string) as id",
+ "cast(id + 200 as int) as ts",
+ "concat('rider-updated-', cast(id as string)) as rider",
+ "concat('driver-updated-', cast(id as string)) as driver",
+ "cast(id + 1000 as double) as fare",
+ "'2021/09/11' as partition");
+
+ updates.write()
+ .format("hudi")
+ .option(HoodieWriteConfig.TBL_NAME.key(), tableName)
+ .option(DataSourceWriteOptions.TABLE_TYPE().key(),
HoodieTableType.MERGE_ON_READ.name())
+ .option(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL())
+ .option(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "id")
+ .option(DataSourceWriteOptions.PRECOMBINE_FIELD().key(), "ts")
+ .option(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(),
"partition")
+ .option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(),
String.valueOf(HoodieTableVersion.TEN.versionCode()))
+ .option(HoodieWriteConfig.AUTO_UPGRADE_VERSION.key(), "false")
+ .option(HoodieMetadataConfig.ENABLE.key(), "false")
+ .option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false")
+ .mode(SaveMode.Append)
+ .save(tablePath);
+
+ HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf().newInstance())
+ .setBasePath(tablePath)
+ .build();
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(HoodieTableVersion.TEN,
metaClient.getTableConfig().getTableVersion());
+ validateLogFilesCount(metaClient, "before 10->9 downgrade", true);
+
+ HoodieWriteConfig downgradeConfig = HoodieWriteConfig.newBuilder()
+ .withPath(metaClient.getBasePath())
+ .withWriteTableVersion(HoodieTableVersion.NINE.versionCode())
+ .withAutoUpgradeVersion(true)
+
.withCompactionConfig(HoodieCompactionConfig.newBuilder().withInlineCompaction(false).build())
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
+ .build();
+
+ new UpgradeDowngrade(metaClient, downgradeConfig, context(),
SparkUpgradeDowngradeHelper.getInstance())
+ .run(HoodieTableVersion.NINE, null);
+
+ HoodieTableMetaClient resultMetaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf().newInstance())
+ .setBasePath(metaClient.getBasePath())
+ .build();
+ assertEquals(HoodieTableVersion.NINE,
resultMetaClient.getTableConfig().getTableVersion());
+
assertFalse(resultMetaClient.getTableConfig().contains(HoodieTableConfig.TABLE_STORAGE_LAYOUT));
+ validateLogFilesCount(resultMetaClient, "after 10->9 downgrade", false);
+ Dataset<Row> downgradedData = readTableData(resultMetaClient, "after 10->9
downgrade")
+ .select("id", "rider", "driver", "fare");
+ assertEquals(100, downgradedData.count());
+ assertEquals(100, downgradedData.select("id").distinct().count());
+ }
+
/**
* Load a fixture table from resources and copy it to a temporary location
for testing.
*/
@@ -815,6 +900,9 @@ public class TestUpgradeDowngrade extends
SparkClientFunctionalTestHarness {
if (fromVersion == HoodieTableVersion.NINE && toVersion ==
HoodieTableVersion.EIGHT) {
return true; // NineToEightDowngradeHandler
}
+ if (fromVersion == HoodieTableVersion.TEN && toVersion ==
HoodieTableVersion.NINE) {
+ return true; // TenToNineDowngradeHandler
+ }
return false; // All other transitions don't perform rollbacks
}
@@ -1049,8 +1137,6 @@ public class TestUpgradeDowngrade extends
SparkClientFunctionalTestHarness {
newRecordData.write()
.format("hudi")
.option(HoodieWriteConfig.TBL_NAME.key(),
metaClientV9.getTableConfig().getTableName())
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
- .option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(),
String.valueOf(HoodieTableVersion.NINE.versionCode()))
.mode(SaveMode.Append)
.save(metaClientV9.getBasePath().toString());
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
index 50686448b408..58bd5e9af0ef 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/ColumnStatIndexTestBase.scala
@@ -459,6 +459,7 @@ object ColumnStatIndexTestBase {
def testMetadataColumnStatsIndexParams(testV6: Boolean):
java.util.stream.Stream[Arguments] = {
val currentVersionCode = HoodieTableVersion.current().versionCode()
+ val v9VersionCode = HoodieTableVersion.NINE.versionCode()
java.util.stream.Stream.of(
HoodieTableType.values().toStream.flatMap { tableType =>
val v6Seq = if (testV6) {
@@ -473,6 +474,8 @@ object ColumnStatIndexTestBase {
v6Seq ++ Seq(
Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = true, tableVersion = 8)),
Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = false, tableVersion = 8)),
+ Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = true, tableVersion = v9VersionCode)),
+ Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = false, tableVersion = v9VersionCode)),
Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = true, tableVersion = currentVersionCode)),
Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = false, tableVersion = currentVersionCode))
)
@@ -491,6 +494,7 @@ object ColumnStatIndexTestBase {
def testMetadataColumnStatsIndexParamsInMemory(testV6: Boolean):
java.util.stream.Stream[Arguments] = {
val currentVersionCode = HoodieTableVersion.current().versionCode()
+ val v9VersionCode = HoodieTableVersion.NINE.versionCode()
java.util.stream.Stream.of(
HoodieTableType.values().toStream.flatMap { tableType =>
val v6Seq = if (testV6) {
@@ -501,6 +505,7 @@ object ColumnStatIndexTestBase {
v6Seq ++ Seq(
Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = true, tableVersion = 8)),
+ Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = true, tableVersion = v9VersionCode)),
Arguments.arguments(ColumnStatsTestCase(tableType,
shouldReadInMemory = true, tableVersion = currentVersionCode))
)
}: _*
@@ -519,6 +524,7 @@ object ColumnStatIndexTestBase {
def testMetadataColumnStatsIndexParamsForMOR(testV6: Boolean):
java.util.stream.Stream[Arguments] = {
val currentVersionCode = HoodieTableVersion.current().versionCode()
+ val v9VersionCode = HoodieTableVersion.NINE.versionCode()
java.util.stream.Stream.of(
(if (testV6) Seq(
Arguments.arguments(ColumnStatsTestCase(HoodieTableType.MERGE_ON_READ,
shouldReadInMemory = true, tableVersion = 6)),
@@ -526,6 +532,8 @@ object ColumnStatIndexTestBase {
) else Seq.empty) ++ Seq(
Arguments.arguments(ColumnStatsTestCase(HoodieTableType.MERGE_ON_READ,
shouldReadInMemory = true, tableVersion = 8)),
Arguments.arguments(ColumnStatsTestCase(HoodieTableType.MERGE_ON_READ,
shouldReadInMemory = false, tableVersion = 8)),
+ Arguments.arguments(ColumnStatsTestCase(HoodieTableType.MERGE_ON_READ,
shouldReadInMemory = true, tableVersion = v9VersionCode)),
+ Arguments.arguments(ColumnStatsTestCase(HoodieTableType.MERGE_ON_READ,
shouldReadInMemory = false, tableVersion = v9VersionCode)),
Arguments.arguments(ColumnStatsTestCase(HoodieTableType.MERGE_ON_READ,
shouldReadInMemory = true, tableVersion = currentVersionCode)),
Arguments.arguments(ColumnStatsTestCase(HoodieTableType.MERGE_ON_READ,
shouldReadInMemory = false, tableVersion = currentVersionCode))
): _*
@@ -544,6 +552,7 @@ object ColumnStatIndexTestBase {
def testTableTypePartitionTypeParams(testV6: Boolean):
java.util.stream.Stream[Arguments] = {
val currentVersionCode =
HoodieTableVersion.current().versionCode().toString
+ val v9VersionCode = HoodieTableVersion.NINE.versionCode().toString
val v6Seq = if (testV6) {
Seq(
Arguments.arguments(HoodieTableType.COPY_ON_WRITE, "c8", "6"),
@@ -563,6 +572,12 @@ object ColumnStatIndexTestBase {
Arguments.arguments(HoodieTableType.MERGE_ON_READ, "c8", "8"),
Arguments.arguments(HoodieTableType.MERGE_ON_READ, "", "8"),
+ // Table version 9
+ Arguments.arguments(HoodieTableType.COPY_ON_WRITE, "c8",
v9VersionCode),
+ Arguments.arguments(HoodieTableType.COPY_ON_WRITE, "", v9VersionCode),
+ Arguments.arguments(HoodieTableType.MERGE_ON_READ, "c8",
v9VersionCode),
+ Arguments.arguments(HoodieTableType.MERGE_ON_READ, "", v9VersionCode),
+
// Table version current
Arguments.arguments(HoodieTableType.COPY_ON_WRITE, "c8",
currentVersionCode),
Arguments.arguments(HoodieTableType.COPY_ON_WRITE, "",
currentVersionCode),
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
index 42627e48656d..7eb6df797dbc 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
@@ -1505,6 +1505,7 @@ object TestColumnStatsIndex {
java.util.stream.Stream.of(Seq(
Arguments.arguments("6"),
Arguments.arguments("8"),
+ Arguments.arguments("9"),
Arguments.arguments(currentVersionCode)
)
: _*)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestEightToNineUpgrade.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestEightToNineUpgrade.scala
index 5bb76d6ef7c7..09d233cdfd8f 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestEightToNineUpgrade.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestEightToNineUpgrade.scala
@@ -169,7 +169,6 @@ class TestEightToNineUpgrade extends
RecordLevelIndexTestBase {
update.write.format("hudi").
option(OPERATION.key(), "upsert").
option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "9").
mode(SaveMode.Append).
save(basePath)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPayloadDeprecationFlow.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPayloadDeprecationFlow.scala
index 7669f6027ee0..39e5a08dd2a5 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPayloadDeprecationFlow.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPayloadDeprecationFlow.scala
@@ -41,10 +41,6 @@ import org.scalatest.Assertions.assertThrows
import scala.jdk.CollectionConverters._
class TestPayloadDeprecationFlow extends SparkClientFunctionalTestHarness {
- // TODO(https://github.com/apache/hudi/issues/19090): Remove this after 10
-> 9 downgrade is supported.
- private val writeTableVersionNineOpts: Map[String, String] = Map(
- HoodieWriteConfig.WRITE_TABLE_VERSION.key() ->
HoodieTableVersion.NINE.versionCode().toString)
-
/**
* Test if the payload based read have the same behavior for different table
versions.
*/
@@ -195,9 +191,9 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
option(OPERATION.key(), "upsert").
option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(), "1").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "9").
option(HoodieTableConfig.PAYLOAD_CLASS_NAME.key(),
classOf[MySqlDebeziumAvroPayload].getName).
- options(writeTableVersionNineOpts).
mode(SaveMode.Append).
save(basePath)
}
@@ -212,8 +208,8 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
option(OPERATION.key(), "delete").
option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(),
"1").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "9").
options(serviceOpts).
- options(writeTableVersionNineOpts).
mode(SaveMode.Append).
save(basePath)
@@ -221,7 +217,8 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
val insertData = Seq(
(13, 6L, "rider-G", "driver-G", 25.50, "i", "13.1", 13, 1, "i"),
(13, 7L, "rider-H", "driver-H", 30.25, "i", "13.1", 13, 1, "i"))
- performInsert(insertData, columns, serviceOpts, basePath,
writeTableVersionNineOpts)
+ performInsert(insertData, columns, serviceOpts, basePath,
+ Map(HoodieWriteConfig.WRITE_TABLE_VERSION.key() -> "9"))
// Final validation of table management operations after all writes
metaClient = HoodieTableMetaClient.builder()
@@ -330,11 +327,11 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
option(DataSourceWriteOptions.TABLE_NAME.key(), "test_table").
option(OPERATION.key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL).
option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "9").
option(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE.key(), "2048").
option(HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), "1024").
options(serviceOpts).
options(opts).
- options(writeTableVersionNineOpts).
mode(SaveMode.Overwrite).
save(basePath)
// Verify table was created successfully
@@ -360,7 +357,8 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
(11, 1L, "rider-X", "driver-X", 19.10, "i", "11.1", 11, 1, "i"),
(12, 1L, "rider-X", "driver-X", 20.10, "D", "12.1", 12, 1, "d"),
(11, 2L, "rider-Y", "driver-Y", 27.70, "u", "11.1", 11, 1, "u"))
- performUpsert(firstUpdateData, columns, serviceOpts,
writeTableVersionNineOpts, basePath)
+ performUpsert(firstUpdateData, columns, serviceOpts, Map.empty, basePath,
+ tableVersion = Some("9"))
// Validate table version.
metaClient = HoodieTableMetaClient.builder()
.setBasePath(basePath)
@@ -383,7 +381,8 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
(8, 3L, "rider-CC", "driver-CC", 30.00, "u", "8.1", 8, 1, "u"),
// Delete rider-E with LOWER ordering - should be IGNORED (rider-E has
ts=10 originally)
(9, 5L, "rider-EE", "driver-EE", 17.85, "D", "9.1", 9, 1, "d"))
- performUpsert(mixedOrderingData, columns, serviceOpts, opts ++
writeTableVersionNineOpts, basePath)
+ performUpsert(mixedOrderingData, columns, serviceOpts, opts, basePath,
+ tableVersion = Some("9"))
// Validate table version is still 9 after mixed ordering batch
metaClient = HoodieTableMetaClient.builder()
.setBasePath(basePath)
@@ -403,8 +402,8 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
// so that the test will fail if _event_seq is still used for ordering
(9, 4L, "rider-DD", "driver-DD", 34.15, "i", "9.1", 12, 1, "i"),
(12, 5L, "rider-EE", "driver-EE", 17.85, "i", "12.1", 12, 1, "i"))
- performUpsert(secondUpdateData, columns, serviceOpts,
writeTableVersionNineOpts, basePath,
- compactionEnabled = compactionEnabled)
+ performUpsert(secondUpdateData, columns, serviceOpts, Map.empty, basePath,
+ tableVersion = Some("9"), compactionEnabled = compactionEnabled)
// Validate table version as 9.
metaClient = HoodieTableMetaClient.builder()
.setBasePath(basePath)
@@ -423,9 +422,9 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
option(OPERATION.key(), "upsert").
option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(), "1").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "9").
option(HoodieTableConfig.PAYLOAD_CLASS_NAME.key(),
classOf[MySqlDebeziumAvroPayload].getName).
- options(writeTableVersionNineOpts).
mode(SaveMode.Append).
save(basePath)
}
@@ -440,8 +439,8 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
option(OPERATION.key(), "delete").
option(HoodieCompactionConfig.INLINE_COMPACT.key(), "false").
option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(),
"1").
+ option(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "9").
options(serviceOpts).
- options(writeTableVersionNineOpts).
mode(SaveMode.Append).
save(basePath)
@@ -449,7 +448,8 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
val insertData = Seq(
(13, 6L, "rider-G", "driver-G", 25.50, "i", "13.1", 13, 1, "i"),
(13, 7L, "rider-H", "driver-H", 30.25, "i", "13.1", 13, 1, "i"))
- performInsert(insertData, columns, serviceOpts, basePath,
writeTableVersionNineOpts)
+ performInsert(insertData, columns, serviceOpts, basePath,
+ Map(HoodieWriteConfig.WRITE_TABLE_VERSION.key() -> "9"))
// Final validation of table management operations after all writes
metaClient = HoodieTableMetaClient.builder()
@@ -595,7 +595,7 @@ class TestPayloadDeprecationFlow extends
SparkClientFunctionalTestHarness {
// 10. Add post-downgrade upsert to verify table functionality
performUpsert(
Seq((14, 10L, "rider-Z", "driver-Z", 45.50, "i", "14.1", 14, 1, "i")),
- columns, serviceOpts, Map.empty, basePath)
+ columns, serviceOpts, Map.empty, basePath, tableVersion = Some("8"))
// Validate data consistency after downgrade including new row
val downgradeDf = spark.read.format("hudi").load(basePath)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSevenToEightUpgrade.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSevenToEightUpgrade.scala
index b2f23eb4f10d..a4a79bdc1b6f 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSevenToEightUpgrade.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSevenToEightUpgrade.scala
@@ -63,9 +63,8 @@ class TestSevenToEightUpgrade extends
RecordLevelIndexTestBase {
"hoodie.metadata.enable" -> "false",
// "OverwriteWithLatestAvroPayload" is used to trigger merge mode
upgrade/downgrade.
PAYLOAD_CLASS_NAME.key ->
classOf[OverwriteWithLatestAvroPayload].getName,
- RECORD_MERGE_MODE.key -> RecordMergeMode.COMMIT_TIME_ORDERING.name,
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
- HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
HoodieTableVersion.NINE.versionCode().toString)
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
HoodieTableVersion.NINE.versionCode().toString,
+ RECORD_MERGE_MODE.key -> RecordMergeMode.COMMIT_TIME_ORDERING.name)
var hudiOpts = if (!lockProviderClass.equals("null")) {
hudiOptsWithoutLockConfigs ++
Map(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key() -> lockProviderClass)
@@ -146,8 +145,6 @@ class TestSevenToEightUpgrade extends
RecordLevelIndexTestBase {
HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key -> "true",
HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key -> "price",
HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key -> "true",
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
- HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
HoodieTableVersion.EIGHT.versionCode().toString,
// Ensure MDT compaction does not run before downgrade.
HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key -> "100"
)
@@ -344,7 +341,6 @@ class TestSevenToEightUpgrade extends
RecordLevelIndexTestBase {
.build()
val hudiOptsUpgrade = hudiOptsV6 ++ Map(
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
HoodieTableVersion.NINE.versionCode().toString,
HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key ->
"org.apache.hudi.client.transaction.lock.InProcessLockProvider",
HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key ->
"OPTIMISTIC_CONCURRENCY_CONTROL"
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeCommitTimeOrdering.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeCommitTimeOrdering.scala
index 7514f8a8e156..8220a58afa43 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeCommitTimeOrdering.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeCommitTimeOrdering.scala
@@ -34,7 +34,8 @@ class TestMergeModeCommitTimeOrdering extends
HoodieSparkSqlTestBase {
"cow,current,false,false", "cow,current,false,true",
"cow,current,true,false",
"mor,current,false,false", "mor,current,false,true",
"mor,current,true,false",
"cow,6,true,false", "cow,6,true,true", "mor,6,true,true",
- "cow,8,true,false", "cow,8,true,true", "mor,8,true,true").foreach { args =>
+ "cow,8,true,false", "cow,8,true,true", "mor,8,true,true",
+ "cow,9,true,false", "cow,9,true,true", "mor,9,true,true").foreach { args =>
val argList = args.split(',')
val tableType = argList(0)
val tableVersion = if (argList(1).equals("current")) {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeEventTimeOrdering.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeEventTimeOrdering.scala
index 15414603a892..1c61a85fcf81 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeEventTimeOrdering.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeModeEventTimeOrdering.scala
@@ -38,6 +38,8 @@ class TestMergeModeEventTimeOrdering extends
HoodieSparkSqlTestBase {
"mor,current,false",
"cow,8,true",
"mor,8,true",
+ "cow,9,true",
+ "mor,9,true",
"cow,6,true",
"cow,6,false",
"mor,6,true",
@@ -63,10 +65,10 @@ class TestMergeModeEventTimeOrdering extends
HoodieSparkSqlTestBase {
} else {
""
}
- val writeTableVersionClause = tableVersion.toInt match {
- case 6 => s"hoodie.write.table.version = $tableVersion,"
- case 8 => s"hoodie.write.table.version = $tableVersion,"
- case _ => ""
+ val writeTableVersionClause = if (tableVersion.toInt <
HoodieTableVersion.current().versionCode()) {
+ s"hoodie.write.table.version = $tableVersion,"
+ } else {
+ ""
}
val expectedMergeConfigs: Map[String, String] = tableVersion.toInt match {
case 6 =>
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestPartialUpdateForMergeInto.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestPartialUpdateForMergeInto.scala
index 8da68cc4ceac..f7e666ca0962 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestPartialUpdateForMergeInto.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestPartialUpdateForMergeInto.scala
@@ -608,8 +608,11 @@ class TestPartialUpdateForMergeInto extends
HoodieSparkSqlTestBase {
|)""".stripMargin)
}
- test("Partial updates for table version 6 and 8 handled gracefully in MIT") {
- Seq(HoodieTableVersion.SIX.versionCode(),
HoodieTableVersion.EIGHT.versionCode()).foreach(
+ test("Partial updates for table version 6, 8 and 9 handled gracefully in
MIT") {
+ Seq(
+ HoodieTableVersion.SIX.versionCode(),
+ HoodieTableVersion.EIGHT.versionCode(),
+ HoodieTableVersion.NINE.versionCode()).foreach(
tableVersion => withTempDir { tmp =>
val tableName = generateTableName
spark.sql(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/index/TestSecondaryIndex.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/index/TestSecondaryIndex.scala
index 81c8127496af..9c1886ccb062 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/index/TestSecondaryIndex.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/index/TestSecondaryIndex.scala
@@ -24,7 +24,7 @@ import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.config.{HoodieMetadataConfig, RecordMergeMode}
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.WriteOperationType
-import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
TableSchemaResolver}
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
HoodieTableVersion, TableSchemaResolver}
import org.apache.hudi.common.testutils.{HoodieTestDataGenerator,
HoodieTestUtils}
import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
import org.apache.hudi.config.{HoodieClusteringConfig, HoodieCompactionConfig,
HoodieWriteConfig}
@@ -197,11 +197,11 @@ class TestSecondaryIndex extends HoodieSparkSqlTestBase {
/**
* Test case to verify that secondary indexes are automatically dropped when
a table is upgraded
- * from version 8 to version 9. This test:
+ * from version 8 to the current table version. This test:
* 1. Creates a table with version 8
* 2. Creates secondary indexes on 'name' and 'price' columns
* 3. Verifies the indexes are created successfully
- * 4. Upgrades the table to version 9
+ * 4. Upgrades the table to the current table version
* 5. Verifies that the secondary indexes are retained
* 6. Tests this behavior for both COW and MOR table types
*/
@@ -311,9 +311,9 @@ class TestSecondaryIndex extends HoodieSparkSqlTestBase {
Seq(3, "a3", 30, 1000)
))
- // Upgrade table to version 9 and verify secondary indexes are
dropped
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
- withSparkSqlSessionConfig(s"hoodie.write.table.version" -> "9") {
+ val currentVersion = HoodieTableVersion.current().versionCode()
+ // Upgrade table to current version and verify secondary indexes
are retained
+ withSparkSqlSessionConfig("hoodie.write.table.version" ->
currentVersion.toString) {
// Update a record to trigger version upgrade
spark.sql(s"insert into $tableName values(1, 'a1', 11, 1001)")
// Both indexes should be shown
@@ -329,10 +329,10 @@ class TestSecondaryIndex extends HoodieSparkSqlTestBase {
Seq(3, "a3", 30, 1000)
)
verifyData(tableName, expected)
- verifyIndexVersion(basePath, 9, 1)
+ verifyIndexVersion(basePath, currentVersion, 1)
- // Verify that secondary indexes are dropped after upgrade
- dropRecreateIdxAndValidate(tableName, basePath, 9, 2,
dropRecreate = true, expected)
+ // Verify that recreated secondary indexes use the current index
version after upgrade
+ dropRecreateIdxAndValidate(tableName, basePath, currentVersion,
2, dropRecreate = true, expected)
}
}
}
@@ -401,94 +401,91 @@ class TestSecondaryIndex extends HoodieSparkSqlTestBase {
test("Test Secondary Index With Updates Compaction Clustering Deletes") {
withTempDir { tmp =>
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
- withSparkSqlSessionConfig(s"hoodie.write.table.version" -> "9") {
- val tableName = generateTableName
- val basePath = s"${tmp.getCanonicalPath}/$tableName"
- // Step 1: Initial Insertion of Records
- val dataGen = new HoodieTestDataGenerator()
- val hudiOpts: Map[String, String] =
loadInitialBatchAndCreateSecondaryIndex(tableName, basePath, dataGen)
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ // Step 1: Initial Insertion of Records
+ val dataGen = new HoodieTestDataGenerator()
+ val hudiOpts: Map[String, String] =
loadInitialBatchAndCreateSecondaryIndex(tableName, basePath, dataGen)
- // Verify initial state of secondary index
- val initialKeys = spark.sql(s"select _row_key from $tableName limit
5").collect().map(_.getString(0))
- validateSecondaryIndex(basePath, tableName, initialKeys)
- val initialRecordsCount = spark.sql(s"select _row_key from
$tableName").count()
+ // Verify initial state of secondary index
+ val initialKeys = spark.sql(s"select _row_key from $tableName limit
5").collect().map(_.getString(0))
+ validateSecondaryIndex(basePath, tableName, initialKeys)
+ val initialRecordsCount = spark.sql(s"select _row_key from
$tableName").count()
- // Step 3: Perform Update Operations on Subset of Records
- var updateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
- var updateDf =
spark.read.json(spark.sparkContext.parallelize(updateRecords.toSeq, 2))
- updateDf.write.format("hudi")
- .options(hudiOpts)
- .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
- .mode(SaveMode.Append)
- .save(basePath)
- // Verify secondary index after updates
- var updateKeys =
updateDf.select("_row_key").collect().map(_.getString(0))
- validateSecondaryIndex(basePath, tableName, updateKeys)
-
- // Step 4: Trigger Compaction with this update as the compaction
frequency is set to 3 commits
- updateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
- updateDf =
spark.read.json(spark.sparkContext.parallelize(updateRecords.toSeq, 2))
- updateDf.write.format("hudi")
- .options(hudiOpts)
- .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
- .mode(SaveMode.Append)
- .save(basePath)
- // Verify compaction
- var metaClient = HoodieTableMetaClient.builder()
- .setBasePath(basePath)
- .setConf(HoodieTestUtils.getDefaultStorageConf)
- .build()
-
assertTrue(metaClient.getActiveTimeline.getCommitTimeline.filterCompletedInstants.lastInstant.isPresent)
- // Verify secondary index after compaction
- updateKeys = updateDf.select("_row_key").collect().map(_.getString(0))
- validateSecondaryIndex(basePath, tableName, updateKeys)
- // Verify count of records
- assertEquals(initialRecordsCount, spark.sql(s"select _row_key from
$tableName").count())
-
- // Step 5: Trigger Clustering with this update as the clustering
frequency is set to 4 commits
- updateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
- updateDf =
spark.read.json(spark.sparkContext.parallelize(updateRecords.toSeq, 2))
- updateDf.write.format("hudi")
- .options(hudiOpts)
- .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
- .mode(SaveMode.Append)
- .save(basePath)
- // Verify clustering
- metaClient = HoodieTableMetaClient.reload(metaClient)
-
assertTrue(metaClient.getActiveTimeline.getCompletedReplaceTimeline.lastInstant.isPresent)
- // Verify secondary index after clustering
- updateKeys = updateDf.select("_row_key").collect().map(_.getString(0))
- validateSecondaryIndex(basePath, tableName, updateKeys)
-
- // Step 6: Perform Deletes on Records and Validate Secondary Index
- val deleteKeys = initialKeys.take(1) // pick a subset of keys to delete
- val deleteDf =
spark.read.format("hudi").load(basePath).filter(s"_row_key in
('${deleteKeys.mkString("','")}')")
- deleteDf.write.format("hudi")
- .options(hudiOpts)
- .option(OPERATION.key, DELETE_OPERATION_OPT_VAL)
- .mode(SaveMode.Append)
- .save(basePath)
- // Verify secondary index for deletes
- validateSecondaryIndex(basePath, tableName, deleteKeys, hasDeleteKeys
= true)
- // Verify for non deleted keys
- val nonDeletedKeys = initialKeys.diff(deleteKeys)
- validateSecondaryIndex(basePath, tableName, nonDeletedKeys)
-
- // Step 7: Final Update and Validation
- val finalUpdateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
- val finalUpdateDf =
spark.read.json(spark.sparkContext.parallelize(finalUpdateRecords.toSeq, 2))
- finalUpdateDf.write.format("hudi")
- .options(hudiOpts)
- .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
- .mode(SaveMode.Append)
- .save(basePath)
- // Verify secondary index after final updates
- val finalUpdateKeys =
finalUpdateDf.select("_row_key").collect().map(_.getString(0))
- validateSecondaryIndex(basePath, tableName, nonDeletedKeys)
- validateSecondaryIndex(basePath, tableName, finalUpdateKeys)
- dataGen.close()
- }
+ // Step 3: Perform Update Operations on Subset of Records
+ var updateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
+ var updateDf =
spark.read.json(spark.sparkContext.parallelize(updateRecords.toSeq, 2))
+ updateDf.write.format("hudi")
+ .options(hudiOpts)
+ .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ // Verify secondary index after updates
+ var updateKeys =
updateDf.select("_row_key").collect().map(_.getString(0))
+ validateSecondaryIndex(basePath, tableName, updateKeys)
+
+ // Step 4: Trigger Compaction with this update as the compaction
frequency is set to 3 commits
+ updateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
+ updateDf =
spark.read.json(spark.sparkContext.parallelize(updateRecords.toSeq, 2))
+ updateDf.write.format("hudi")
+ .options(hudiOpts)
+ .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ // Verify compaction
+ var metaClient = HoodieTableMetaClient.builder()
+ .setBasePath(basePath)
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .build()
+
assertTrue(metaClient.getActiveTimeline.getCommitTimeline.filterCompletedInstants.lastInstant.isPresent)
+ // Verify secondary index after compaction
+ updateKeys = updateDf.select("_row_key").collect().map(_.getString(0))
+ validateSecondaryIndex(basePath, tableName, updateKeys)
+ // Verify count of records
+ assertEquals(initialRecordsCount, spark.sql(s"select _row_key from
$tableName").count())
+
+ // Step 5: Trigger Clustering with this update as the clustering
frequency is set to 4 commits
+ updateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
+ updateDf =
spark.read.json(spark.sparkContext.parallelize(updateRecords.toSeq, 2))
+ updateDf.write.format("hudi")
+ .options(hudiOpts)
+ .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ // Verify clustering
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
assertTrue(metaClient.getActiveTimeline.getCompletedReplaceTimeline.lastInstant.isPresent)
+ // Verify secondary index after clustering
+ updateKeys = updateDf.select("_row_key").collect().map(_.getString(0))
+ validateSecondaryIndex(basePath, tableName, updateKeys)
+
+ // Step 6: Perform Deletes on Records and Validate Secondary Index
+ val deleteKeys = initialKeys.take(1) // pick a subset of keys to delete
+ val deleteDf =
spark.read.format("hudi").load(basePath).filter(s"_row_key in
('${deleteKeys.mkString("','")}')")
+ deleteDf.write.format("hudi")
+ .options(hudiOpts)
+ .option(OPERATION.key, DELETE_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ // Verify secondary index for deletes
+ validateSecondaryIndex(basePath, tableName, deleteKeys, hasDeleteKeys =
true)
+ // Verify for non deleted keys
+ val nonDeletedKeys = initialKeys.diff(deleteKeys)
+ validateSecondaryIndex(basePath, tableName, nonDeletedKeys)
+
+ // Step 7: Final Update and Validation
+ val finalUpdateRecords =
recordsToStrings(dataGen.generateUniqueUpdates(getInstantTime, 10,
HoodieTestDataGenerator.TRIP_FLATTENED_SCHEMA)).asScala
+ val finalUpdateDf =
spark.read.json(spark.sparkContext.parallelize(finalUpdateRecords.toSeq, 2))
+ finalUpdateDf.write.format("hudi")
+ .options(hudiOpts)
+ .option(OPERATION.key, UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ // Verify secondary index after final updates
+ val finalUpdateKeys =
finalUpdateDf.select("_row_key").collect().map(_.getString(0))
+ validateSecondaryIndex(basePath, tableName, nonDeletedKeys)
+ validateSecondaryIndex(basePath, tableName, finalUpdateKeys)
+ dataGen.close()
}
}
@@ -697,9 +694,7 @@ class TestSecondaryIndex extends HoodieSparkSqlTestBase {
val initialDf =
spark.read.json(spark.sparkContext.parallelize(initialRecords.toSeq, 2))
val hudiOpts = commonOpts ++ Map(
TABLE_TYPE.key -> "MERGE_ON_READ",
- HoodieWriteConfig.TBL_NAME.key -> tableName,
- // todo remove this option after
https://github.com/apache/hudi/issues/19090 resolved.
- HoodieWriteConfig.WRITE_TABLE_VERSION.key() -> "9")
+ HoodieWriteConfig.TBL_NAME.key -> tableName)
initialDf.write.format("hudi")
.options(hudiOpts)
.option(OPERATION.key, INSERT_OPERATION_OPT_VAL)
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 823f2cb7896d..78a100cf3150 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
@@ -157,7 +157,6 @@ class TestUpgradeOrDowngradeProcedure extends
HoodieSparkProcedureTestBase {
val tableName = generateTableName
val tablePath = s"${tmp.getCanonicalPath}/$tableName"
// create table
- // todo remove option 'hoodie.write.table.version' after
https://github.com/apache/hudi/issues/19090 resolved.
spark.sql(
s"""
|create table $tableName (
@@ -170,8 +169,7 @@ class TestUpgradeOrDowngradeProcedure extends
HoodieSparkProcedureTestBase {
| options (
| type = 'mor',
| primaryKey = 'id',
- | preCombineField = 'ts',
- | hoodie.write.table.version = '9'
+ | preCombineField = 'ts'
| )
""".stripMargin)
withSQLConf(
