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 1e59c9eb6d9e fix(core): record the ordering field when upgrading a
table from version 1 (#19803)
1e59c9eb6d9e is described below
commit 1e59c9eb6d9e341d5fd10e1418c0c3deb34b6579
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Aug 31 21:30:39 2026 -0700
fix(core): record the ordering field when upgrading a table from version 1
(#19803)
---
.../hudi/table/upgrade/OneToTwoUpgradeHandler.java | 85 +++++++++++++-
.../table/upgrade/TestOneToTwoUpgradeHandler.java | 126 +++++++++++++++++++++
.../table/upgrade/TestUpgradeDowngradeLegacy.java | 98 ++++++++++++++++
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 55 +++++++++
4 files changed, 363 insertions(+), 1 deletion(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/OneToTwoUpgradeHandler.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/OneToTwoUpgradeHandler.java
index 79e29a9d2e60..0677e955dbaa 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/OneToTwoUpgradeHandler.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/OneToTwoUpgradeHandler.java
@@ -20,18 +20,32 @@ package org.apache.hudi.table.upgrade;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
import java.util.Hashtable;
+import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
/**
* Upgrade handle to assist in upgrading hoodie table from version 1 to 2.
*/
public class OneToTwoUpgradeHandler implements UpgradeHandler {
+ private static final Logger LOG =
LoggerFactory.getLogger(OneToTwoUpgradeHandler.class);
+ private static final String NESTED_FIELD_SEPARATOR = ".";
+
@Override
public UpgradeDowngrade.TableConfigChangeSet upgrade(
HoodieWriteConfig config,
@@ -42,6 +56,75 @@ public class OneToTwoUpgradeHandler implements
UpgradeHandler {
tablePropsToAdd.put(HoodieTableConfig.PARTITION_FIELDS,
upgradeDowngradeHelper.getPartitionColumns(config));
tablePropsToAdd.put(HoodieTableConfig.RECORDKEY_FIELDS,
config.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key()));
tablePropsToAdd.put(HoodieTableConfig.BASE_FILE_FORMAT,
config.getString(HoodieTableConfig.BASE_FILE_FORMAT));
- return new UpgradeDowngrade.TableConfigChangeSet();
+ HoodieTableMetaClient metaClient = upgradeDowngradeHelper.getTable(config,
context).getMetaClient();
+ getOrderingFieldsToRecord(config, metaClient)
+ .ifPresent(orderingFields ->
tablePropsToAdd.put(HoodieTableConfig.PRECOMBINE_FIELD, orderingFields));
+ return new UpgradeDowngrade.TableConfigChangeSet(tablePropsToAdd,
Collections.emptySet());
+ }
+
+ /**
+ * Returns the ordering fields to record in {@code hoodie.properties}, if
they can be established.
+ *
+ * <p>The ordering field is only written at table creation, and only since
0.8.0, so a table
+ * created before that records none even though its writer merges on one,
and everything that
+ * resolves ordering from the table config alone then falls back to no
ordering. Only a table that
+ * records none is filled in, and only from this upgrade. A top level field
is recorded once the
+ * schema has it, which is what keeps out a materialized default; a field
nested under dot
+ * notation is recorded as configured, since a default is never nested. They
are recorded under
+ * the deprecated {@link HoodieTableConfig#PRECOMBINE_FIELD}, the key a
table at this version is
+ * expected to carry and the one a 0.x reader understands; {@code
EightToNineUpgradeHandler}
+ * migrates it to {@link HoodieTableConfig#ORDERING_FIELDS} in turn.
+ */
+ private static Option<String> getOrderingFieldsToRecord(HoodieWriteConfig
config, HoodieTableMetaClient metaClient) {
+ if (!metaClient.getTableConfig().getOrderingFields().isEmpty()) {
+ // the table already records them, and the upgrade only fills in missing
ordering fields
+ return Option.empty();
+ }
+ List<String> orderingFields = config.getPreCombineFields();
+ if (orderingFields.isEmpty()) {
+ return Option.empty();
+ }
+ // only an explicit config can name a nested field, so take the writer at
its word for those
+ List<String> topLevelFields = orderingFields.stream()
+ .filter(field -> !field.contains(NESTED_FIELD_SEPARATOR))
+ .collect(Collectors.toList());
+ if (topLevelFields.isEmpty()) {
+ return Option.of(String.join(",", orderingFields));
+ }
+ Option<HoodieSchema> schema = resolveSchema(config, metaClient);
+ if (!schema.isPresent()) {
+ LOG.warn("Skipping the ordering fields {} while upgrading {} to table
version two: no schema is available to "
+ + "resolve them against", orderingFields, config.getBasePath());
+ return Option.empty();
+ }
+ if (!topLevelFields.stream().allMatch(field ->
schema.get().getField(field).isPresent())) {
+ LOG.warn("Skipping the ordering fields {} while upgrading {} to table
version two: the schema does not have "
+ + "all of them as top level fields", orderingFields,
config.getBasePath());
+ return Option.empty();
+ }
+ return Option.of(String.join(",", orderingFields));
+ }
+
+ /**
+ * The table's own schema, falling back to the writer's for a table with no
committed data, and to
+ * nothing if neither can be read.
+ */
+ private static Option<HoodieSchema> resolveSchema(HoodieWriteConfig config,
HoodieTableMetaClient metaClient) {
+ try {
+ Option<HoodieSchema> tableSchema = new
TableSchemaResolver(metaClient).getTableSchemaIfPresent(false);
+ if (tableSchema.isPresent()) {
+ return tableSchema;
+ }
+ String writeSchema = config.getWriteSchema();
+ return StringUtils.isNullOrEmpty(writeSchema)
+ ? Option.empty()
+ : Option.of(HoodieSchema.parse(writeSchema));
+ } catch (Exception e) {
+ // the upgrade gates every write, so a schema that cannot be read or
parsed leaves the fields
+ // unrecorded rather than blocking the table
+ LOG.warn("Failed to resolve the schema of " + config.getBasePath()
+ + " while upgrading to table version two, leaving the ordering
fields unrecorded", e);
+ return Option.empty();
+ }
}
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestOneToTwoUpgradeHandler.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestOneToTwoUpgradeHandler.java
new file mode 100644
index 000000000000..b595d86642e4
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestOneToTwoUpgradeHandler.java
@@ -0,0 +1,126 @@
+/*
+ * 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.config.ConfigProperty;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+import org.apache.hudi.testutils.HoodieClientTestBase;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Unit tests {@link OneToTwoUpgradeHandler}. The handler is exercised
directly rather than through
+ * {@link UpgradeDowngrade}, which rejects any table below version six.
+ */
+class TestOneToTwoUpgradeHandler extends HoodieClientTestBase {
+
+ @Test
+ void testUpgradeRecordsKeySchemaAndOrderingField() {
+ Map<ConfigProperty, String> tableProps =
upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA, "timestamp");
+ assertEquals("uuid", tableProps.get(HoodieTableConfig.RECORDKEY_FIELDS));
+ assertEquals("partition_path",
tableProps.get(HoodieTableConfig.PARTITION_FIELDS));
+ assertEquals(HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().name(),
tableProps.get(HoodieTableConfig.BASE_FILE_FORMAT));
+ assertEquals("timestamp",
tableProps.get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ /**
+ * A nested ordering field is recorded as configured. The schema check
exists to catch a
+ * materialized default, and a default is never nested, so an explicitly
configured nested field
+ * is taken at face value.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"fare.amount", "not_a_record.not_a_column"})
+ void testRecordsNestedOrderingFieldAsConfigured(String orderingField) {
+ assertEquals(orderingField,
+ upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
orderingField).get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ /** A table that already records ordering fields keeps them, even if the
writer configures another. */
+ @Test
+ void testLeavesRecordedOrderingFieldsAlone() {
+ Properties recordedProps = new Properties();
+ recordedProps.setProperty(HoodieTableConfig.PRECOMBINE_FIELD.key(),
"timestamp");
+ HoodieTableConfig.update(metaClient.getStorage(),
metaClient.getMetaPath(), recordedProps);
+
+ assertNull(upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
"_row_key").get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ /** Every field of a multi field ordering config has to resolve for any of
it to be recorded. */
+ @Test
+ void testRecordsMultipleOrderingFields() {
+ assertEquals("timestamp,_row_key",
+ upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
"timestamp,_row_key").get(HoodieTableConfig.PRECOMBINE_FIELD));
+ assertNull(upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
"timestamp,not_a_column").get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ /**
+ * An ordering field the schema cannot resolve is left unrecorded, so the
table config never ends
+ * up with one no reader can resolve.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"not_a_column", "ts"})
+ void testSkipsOrderingFieldTheSchemaCannotResolve(String orderingField) {
+ Map<ConfigProperty, String> tableProps =
upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA, orderingField);
+ assertEquals("uuid", tableProps.get(HoodieTableConfig.RECORDKEY_FIELDS));
+ assertNull(tableProps.get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ @Test
+ void testSkipsEmptyOrderingField() {
+ assertNull(upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
"").get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ /** With no committed data and no writer schema there is nothing to resolve
the field against. */
+ @Test
+ void testSkipsOrderingFieldWhenNoSchemaIsAvailable() {
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+
.withPath(basePath).forTable("test-trip-table").withProps(keySchemaParams("timestamp")).build();
+ Map<ConfigProperty, String> tableProps = new OneToTwoUpgradeHandler()
+ .upgrade(config, context, null,
SparkUpgradeDowngradeHelper.getInstance()).propertiesToUpdate();
+ assertNull(tableProps.get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ private Map<ConfigProperty, String> upgrade(String schema, String
orderingField) {
+ HoodieWriteConfig config =
getConfigBuilder(schema).withProps(keySchemaParams(orderingField)).build();
+ return new OneToTwoUpgradeHandler()
+ .upgrade(config, context, null,
SparkUpgradeDowngradeHelper.getInstance())
+ .propertiesToUpdate();
+ }
+
+ private Map<String, String> keySchemaParams(String orderingField) {
+ Map<String, String> params = new HashMap<>();
+ params.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "uuid");
+ params.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(),
"partition_path");
+ params.put(HoodieTableConfig.BASE_FILE_FORMAT.key(),
HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().name());
+ params.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), orderingField);
+ return params;
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngradeLegacy.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngradeLegacy.java
index 34f7f3c48b9c..02b95111e6c0 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngradeLegacy.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngradeLegacy.java
@@ -269,6 +269,7 @@ public class TestUpgradeDowngradeLegacy extends
HoodieClientTestBase {
// init config, table and client.
Map<String, String> params = new HashMap<>();
addNewTableParamsToProps(params);
+ params.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), "timestamp");
if (tableType == HoodieTableType.MERGE_ON_READ) {
params.put(TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
metaClient = HoodieTestUtils.init(storageConf, basePath,
HoodieTableType.MERGE_ON_READ);
@@ -280,6 +281,8 @@ public class TestUpgradeDowngradeLegacy extends
HoodieClientTestBase {
// downgrade table props
downgradeTableConfigsFromTwoToOne(cfg);
+ // a table written before 0.8.0 records no ordering field at all
+ assertTrue(metaClient.getTableConfig().getOrderingFields().isEmpty());
// perform upgrade
new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
@@ -293,6 +296,78 @@ public class TestUpgradeDowngradeLegacy extends
HoodieClientTestBase {
// verify table props
assertTableProps(cfg);
+ // the ordering field the writer merges on is now recorded for readers
that only see the table config
+ assertEquals(Collections.singletonList("timestamp"),
metaClient.getTableConfig().getOrderingFields());
+ }
+
+ /**
+ * The migration case: a table with committed data, whose ordering field
resolves against the
+ * table's own schema rather than the schema the writer brings along.
+ */
+ @Disabled("HUDI-9700")
+ @Test
+ void testUpgradeOneToTwoRecordsOrderingFieldFromTableSchema() throws
IOException {
+ Map<String, String> params = new HashMap<>();
+ addNewTableParamsToProps(params);
+ params.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), "timestamp");
+ HoodieWriteConfig cfg =
getConfigBuilder().withRollbackUsingMarkers(false).withProps(params).build();
+ doInsert(getHoodieWriteClient(cfg));
+
+ downgradeTableConfigsFromTwoToOne(cfg);
+ assertTrue(metaClient.getTableConfig().getOrderingFields().isEmpty());
+
+ new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
+ .run(HoodieTableVersion.TWO, null);
+
+ metaClient = reloadMetaClientAtVersionTwo(cfg);
+ assertEquals(Collections.singletonList("timestamp"),
metaClient.getTableConfig().getOrderingFields());
+ }
+
+ /**
+ * A table that already recorded an ordering field keeps the one it has,
rather than having the
+ * writer's config written over it.
+ */
+ @Disabled("HUDI-9700")
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testUpgradeOneToTwoKeepsRecordedOrderingField(HoodieTableType
tableType) throws IOException {
+ Map<String, String> params = new HashMap<>();
+ addNewTableParamsToProps(params);
+ params.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), "_row_key");
+ initTableOfType(tableType, params);
+ HoodieWriteConfig cfg =
getConfigBuilder().withRollbackUsingMarkers(false).withProps(params).build();
+ doInsert(getHoodieWriteClient(cfg));
+
+ downgradeTableConfigsFromTwoToOne(cfg, "timestamp");
+ assertEquals(Collections.singletonList("timestamp"),
metaClient.getTableConfig().getOrderingFields());
+
+ new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
+ .run(HoodieTableVersion.TWO, null);
+
+ metaClient = reloadMetaClientAtVersionTwo(cfg);
+ assertEquals(Collections.singletonList("timestamp"),
metaClient.getTableConfig().getOrderingFields());
+ }
+
+ /**
+ * An ordering field the schema cannot resolve is left unrecorded, so the
table config never ends
+ * up with one no reader can resolve.
+ */
+ @Disabled("HUDI-9700")
+ @Test
+ void testUpgradeOneToTwoSkipsOrderingFieldMissingFromSchema() throws
IOException {
+ Map<String, String> params = new HashMap<>();
+ addNewTableParamsToProps(params);
+ params.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), "not_a_column");
+ HoodieWriteConfig cfg =
getConfigBuilder().withRollbackUsingMarkers(false).withProps(params).build();
+ doInsert(getHoodieWriteClient(cfg));
+
+ downgradeTableConfigsFromTwoToOne(cfg);
+
+ new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
+ .run(HoodieTableVersion.TWO, null);
+
+ metaClient = reloadMetaClientAtVersionTwo(cfg);
+ assertTrue(metaClient.getTableConfig().getOrderingFields().isEmpty());
}
@Disabled("HUDI-9700")
@@ -501,13 +576,36 @@ public class TestUpgradeDowngradeLegacy extends
HoodieClientTestBase {
client.insert(writeRecords, commit1).collect();
}
+ private HoodieTableMetaClient reloadMetaClientAtVersionTwo(HoodieWriteConfig
cfg) throws IOException {
+ HoodieTableMetaClient reloaded = HoodieTableMetaClient.builder()
+
.setConf(context.getStorageConf().newInstance()).setBasePath(cfg.getBasePath())
+ .setLayoutVersion(Option.of(new
TimelineLayoutVersion(cfg.getTimelineLayoutVersion()))).build();
+ assertTableVersionOnDataAndMetadataTable(reloaded, HoodieTableVersion.TWO);
+ return reloaded;
+ }
+
+ private void initTableOfType(HoodieTableType tableType, Map<String, String>
params) throws IOException {
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ params.put(TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
+ metaClient = HoodieTestUtils.init(storageConf, basePath,
HoodieTableType.MERGE_ON_READ);
+ }
+ }
+
private void downgradeTableConfigsFromTwoToOne(HoodieWriteConfig cfg) throws
IOException {
+ downgradeTableConfigsFromTwoToOne(cfg, null);
+ }
+
+ private void downgradeTableConfigsFromTwoToOne(HoodieWriteConfig cfg, String
orderingField) throws IOException {
Properties properties = new Properties(cfg.getProps());
properties.remove(HoodieTableConfig.RECORDKEY_FIELDS.key());
properties.remove(HoodieTableConfig.PARTITION_FIELDS.key());
properties.remove(HoodieTableConfig.NAME.key());
properties.remove(BASE_FILE_FORMAT.key());
+ properties.remove(HoodieTableConfig.PRECOMBINE_FIELD.key());
properties.setProperty(HoodieTableConfig.VERSION.key(), "1");
+ if (orderingField != null) {
+ properties.setProperty(HoodieTableConfig.PRECOMBINE_FIELD.key(),
orderingField);
+ }
metaClient = HoodieTestUtils.init(storageConf, basePath, getTableType(),
properties);
// set hoodie.table.version to 1 in hoodie.properties file
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
index e4ae6bbdd086..7177e6e527bc 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
@@ -23,6 +23,7 @@ import org.apache.hudi.common.config.{HoodieConfig,
HoodieMetadataConfig, Record
import org.apache.hudi.common.model.{DefaultHoodieRecordPayload,
HoodieFileFormat, HoodieRecord, HoodieRecordPayload,
HoodieReplaceCommitMetadata, HoodieTableType, MetaFieldsMode,
WriteOperationType}
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
TableSchemaResolver}
+import org.apache.hudi.common.table.HoodieTableVersion
import org.apache.hudi.common.table.timeline.{HoodieTimeline, TimelineUtils}
import org.apache.hudi.common.testutils.HoodieTestDataGenerator
import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieIndexConfig,
HoodieWriteConfig}
@@ -40,6 +41,7 @@ import org.apache.spark.sql.{DataFrame, Row, SaveMode,
SparkSession}
import org.apache.spark.sql.functions.{expr, lit}
import org.apache.spark.sql.hudi.command.SqlKeyGenerator
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertNotNull, assertNull, assertTrue, fail}
+import org.junit.jupiter.api.Disabled
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.{Arguments, CsvSource, EnumSource,
MethodSource, ValueSource}
@@ -61,6 +63,59 @@ import scala.collection.JavaConverters._
*/
class TestHoodieSparkSqlWriter extends HoodieSparkWriterTestBase {
+ case class OrderedRecord(uuid: String, version: Long, ts: Long, value:
String)
+
+ /**
+ * A writer that configures no ordering field resolves it from the table
config, so without the
+ * field the version 1 to 2 upgrade records, the "ts" fallback lets an older
record overwrite a
+ * newer one. Disabled alongside the rest of the legacy upgrade coverage
under HUDI-9700, since
+ * UpgradeDowngrade refuses any table below version 6 and the upgrading
write throws before the
+ * backfill runs.
+ */
+ @Disabled("HUDI-9700")
+ @Test
+ def testUpgradeFromTableVersionOneRestoresOrderingOnUpdates(): Unit = {
+ val writeParams = Map("path" -> tempBasePath,
+ HoodieWriteConfig.TBL_NAME.key -> hoodieFooTableName,
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "uuid",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "",
+ DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key ->
classOf[NonpartitionedKeyGenerator].getName,
+ DataSourceWriteOptions.PAYLOAD_CLASS_NAME.key ->
classOf[DefaultHoodieRecordPayload].getName)
+ val orderingParams = writeParams +
(DataSourceWriteOptions.PRECOMBINE_FIELD.key -> "version")
+ HoodieSparkSqlWriter.write(sqlContext, SaveMode.Overwrite, orderingParams,
+ orderedRecordFrame("key1", version = 2, ts = 1, value = "new"))
+
+ // a table written before 0.8.0 records no ordering field
+ dropRecordedOrderingFieldAndSetVersionOne()
+
+ // the upgrading write records the ordering field it merges on
+ HoodieSparkSqlWriter.write(sqlContext, SaveMode.Append, orderingParams,
+ orderedRecordFrame("key2", version = 1, ts = 1, value = "other"))
+ assertEquals(Collections.singletonList("version"),
+ createMetaClient(spark, tempBasePath).getTableConfig.getOrderingFields)
+
+ // the lower version loses the merge even though its "ts" is higher
+ HoodieSparkSqlWriter.write(sqlContext, SaveMode.Append, writeParams,
+ orderedRecordFrame("key1", version = 1, ts = 5, value = "old"))
+ assertEquals("new", readValueOf("key1"))
+ }
+
+ private def orderedRecordFrame(uuid: String, version: Long, ts: Long, value:
String): DataFrame =
+ spark.createDataFrame(Seq(OrderedRecord(uuid, version, ts, value)))
+
+ private def dropRecordedOrderingFieldAndSetVersionOne(): Unit = {
+ val metaClient = createMetaClient(spark, tempBasePath)
+ HoodieTableConfig.delete(metaClient.getStorage, metaClient.getMetaPath,
+ Collections.singleton(HoodieTableConfig.PRECOMBINE_FIELD.key))
+ val versionProps = new java.util.Properties()
+ versionProps.setProperty(HoodieTableConfig.VERSION.key,
String.valueOf(HoodieTableVersion.ONE.versionCode))
+ HoodieTableConfig.update(metaClient.getStorage, metaClient.getMetaPath,
versionProps)
+ }
+
+ private def readValueOf(uuid: String): String =
+ spark.read.format("hudi").load(tempBasePath).where(s"uuid = '$uuid'")
+ .select("value").collect().head.getString(0)
+
/**
* Local utility method for performing bulk insert tests.
*
