This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-0.x by this push:
new abbb5d4c1c46 fix(core): record the ordering field when upgrading a
table from version 1 (#19739)
abbb5d4c1c46 is described below
commit abbb5d4c1c4636e00f604844c4ceb950e19cf8dd
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Aug 31 21:32:45 2026 -0700
fix(core): record the ordering field when upgrading a table from version 1
(#19739)
---
.../hudi/table/upgrade/OneToTwoUpgradeHandler.java | 76 +++++++++++++
.../table/upgrade/TestOneToTwoUpgradeHandler.java | 117 +++++++++++++++++++++
.../hudi/table/upgrade/TestUpgradeDowngrade.java | 103 ++++++++++++++++++
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 50 ++++++++-
4 files changed, 345 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 dbf4d6159dcb..0e5cc2e69e0c 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
@@ -18,12 +18,21 @@
package org.apache.hudi.table.upgrade;
+import org.apache.hudi.avro.AvroSchemaUtils;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.engine.HoodieEngineContext;
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.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.Hashtable;
import java.util.Map;
@@ -32,6 +41,9 @@ import java.util.Map;
*/
public class OneToTwoUpgradeHandler implements UpgradeHandler {
+ private static final Logger LOG =
LoggerFactory.getLogger(OneToTwoUpgradeHandler.class);
+ private static final String NESTED_FIELD_SEPARATOR = ".";
+
@Override
public Map<ConfigProperty, String> upgrade(
HoodieWriteConfig config, HoodieEngineContext context, String
instantTime,
@@ -40,6 +52,70 @@ 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));
+ HoodieTableMetaClient metaClient = upgradeDowngradeHelper.getTable(config,
context).getMetaClient();
+ getPreCombineFieldToPersist(config, metaClient)
+ .ifPresent(preCombineField ->
tablePropsToAdd.put(HoodieTableConfig.PRECOMBINE_FIELD, preCombineField));
return tablePropsToAdd;
}
+
+ /**
+ * Returns the ordering field to record in {@code hoodie.properties}, if one
can be established.
+ *
+ * <p>{@link HoodieTableConfig#PRECOMBINE_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 the "ts" default
that every write
+ * config materializes; a field nested under dot notation is recorded as
configured, since a
+ * default is never nested.
+ */
+ private static Option<String> getPreCombineFieldToPersist(HoodieWriteConfig
config, HoodieTableMetaClient metaClient) {
+ if
(StringUtils.nonEmpty(metaClient.getTableConfig().getPreCombineField())) {
+ // the table already records one, and the upgrade only fills in a
missing ordering field
+ return Option.empty();
+ }
+ String preCombineField = config.getPreCombineField();
+ if (StringUtils.isNullOrEmpty(preCombineField)) {
+ return Option.empty();
+ }
+ if (preCombineField.contains(NESTED_FIELD_SEPARATOR)) {
+ // only an explicit config can name a nested field, so take the writer
at its word
+ return Option.of(preCombineField);
+ }
+ Option<Schema> schema = resolveSchema(config, metaClient);
+ if (!schema.isPresent()) {
+ LOG.warn("Skipping the ordering field {} while upgrading {} to table
version two: no schema is available to "
+ + "resolve it against", preCombineField, config.getBasePath());
+ return Option.empty();
+ }
+ if (!AvroSchemaUtils.containsFieldInSchema(schema.get(), preCombineField))
{
+ LOG.warn("Skipping the ordering field {} while upgrading {} to table
version two: the schema has no such "
+ + "top level field", preCombineField, config.getBasePath());
+ return Option.empty();
+ }
+ return Option.of(preCombineField);
+ }
+
+ /**
+ * 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<Schema> resolveSchema(HoodieWriteConfig config,
HoodieTableMetaClient metaClient) {
+ try {
+ Option<Schema> tableSchema = new
TableSchemaResolver(metaClient).getTableAvroSchemaIfPresent(false);
+ if (tableSchema.isPresent()) {
+ return tableSchema;
+ }
+ String writeSchema = config.getWriteSchema();
+ return StringUtils.isNullOrEmpty(writeSchema)
+ ? Option.empty()
+ : Option.of(new Schema.Parser().parse(writeSchema));
+ } catch (Exception e) {
+ // the upgrade gates every write, so a schema that cannot be read or
parsed leaves the field
+ // 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 field
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..ce606d394053
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestOneToTwoUpgradeHandler.java
@@ -0,0 +1,117 @@
+/*
+ * 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 the ordering field {@link OneToTwoUpgradeHandler} records,
exercising the handler
+ * directly so each way of resolving the field against the schema can be
covered on its own.
+ */
+class TestOneToTwoUpgradeHandler extends HoodieClientTestBase {
+
+ @Test
+ void testRecordsKeySchemaAlongsideOrderingField() {
+ 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));
+ }
+
+ /**
+ * An ordering field the schema has no top level column for is left
unrecorded, so the table
+ * config never ends up with one no reader can resolve. That covers the "ts"
default every write
+ * config materializes whether or not the user asked for it.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"ts", "not_a_column"})
+ 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));
+ }
+
+ /**
+ * A nested ordering field is recorded as configured. The schema check
exists to catch the
+ * materialized "ts" 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));
+ }
+
+ @Test
+ void testSkipsEmptyOrderingField() {
+ assertNull(upgrade(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
"").get(HoodieTableConfig.PRECOMBINE_FIELD));
+ }
+
+ /** A table that already records an ordering field keeps it, even if the
writer configures another. */
+ @Test
+ void testLeavesRecordedOrderingFieldAlone() {
+ 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));
+ }
+
+ /** 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());
+ 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());
+ }
+
+ 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/TestUpgradeDowngrade.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
index c57e2151bf00..4d7f829a7e43 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
@@ -102,6 +102,7 @@ import static
org.apache.hudi.metadata.MetadataPartitionType.RECORD_INDEX;
import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -250,6 +251,7 @@ public class TestUpgradeDowngrade 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);
@@ -261,6 +263,8 @@ public class TestUpgradeDowngrade extends
HoodieClientTestBase {
// downgrade table props
downgradeTableConfigsFromTwoToOne(cfg);
+ // a table written before 0.8.0 records no ordering field at all
+ assertNull(metaClient.getTableConfig().getPreCombineField());
// perform upgrade
new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
@@ -274,6 +278,90 @@ public class TestUpgradeDowngrade 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("timestamp",
metaClient.getTableConfig().getPreCombineField());
+ }
+
+ /**
+ * 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.
+ */
+ @Test
+ void testUpgradeOneToTwoRecordsOrderingFieldFromTableSchema() throws
IOException {
+ Map<String, String> params = new HashMap<>();
+ addNewTableParamsToProps(params);
+ params.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), "timestamp");
+ HoodieWriteConfig cfg =
getConfigBuilder().withAutoCommit(true).withRollbackUsingMarkers(false).withProps(params).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ doInsert(client);
+
+ downgradeTableConfigsFromTwoToOne(cfg);
+ assertNull(metaClient.getTableConfig().getPreCombineField());
+
+ new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
+ .run(HoodieTableVersion.TWO, null);
+
+ metaClient = HoodieTableMetaClient.builder()
+
.setConf(context.getStorageConf().newInstance()).setBasePath(cfg.getBasePath())
+ .setLayoutVersion(Option.of(new
TimelineLayoutVersion(cfg.getTimelineLayoutVersion()))).build();
+ assertTableVersionOnDataAndMetadataTable(metaClient,
HoodieTableVersion.TWO);
+ assertEquals("timestamp",
metaClient.getTableConfig().getPreCombineField());
+ }
+
+ /**
+ * A table that already recorded an ordering field before 0.8.0 stopped
being the norm keeps the
+ * one it has, rather than having the writer's config written over it.
+ */
+ @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().withAutoCommit(true).withRollbackUsingMarkers(false).withProps(params).build();
+ doInsert(getHoodieWriteClient(cfg));
+
+ downgradeTableConfigsFromTwoToOne(cfg, "timestamp");
+ assertEquals("timestamp",
metaClient.getTableConfig().getPreCombineField());
+
+ new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
+ .run(HoodieTableVersion.TWO, null);
+
+ metaClient = HoodieTableMetaClient.builder()
+
.setConf(context.getStorageConf().newInstance()).setBasePath(cfg.getBasePath())
+ .setLayoutVersion(Option.of(new
TimelineLayoutVersion(cfg.getTimelineLayoutVersion()))).build();
+ assertTableVersionOnDataAndMetadataTable(metaClient,
HoodieTableVersion.TWO);
+ assertEquals("timestamp",
metaClient.getTableConfig().getPreCombineField());
+ }
+
+ /**
+ * {@link HoodieWriteConfig#PRECOMBINE_FIELD_NAME} defaults to "ts", which
every write config
+ * materializes whether or not the user asked for it. Recording that default
on a table without
+ * such a field would leave an ordering field no reader can resolve, and
would fail table config
+ * validation for the next writer that configures a real one.
+ */
+ @Test
+ void testUpgradeOneToTwoSkipsOrderingFieldMissingFromSchema() throws
IOException {
+ Map<String, String> params = new HashMap<>();
+ addNewTableParamsToProps(params);
+ HoodieWriteConfig cfg =
getConfigBuilder().withAutoCommit(false).withRollbackUsingMarkers(false).withProps(params).build();
+ // no ordering field was configured, so the write config carries the "ts"
default, which the
+ // test data generator's schema does not have
+ assertEquals("ts", cfg.getPreCombineField());
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ doInsert(client);
+
+ downgradeTableConfigsFromTwoToOne(cfg);
+
+ new UpgradeDowngrade(metaClient, cfg, context,
SparkUpgradeDowngradeHelper.getInstance())
+ .run(HoodieTableVersion.TWO, null);
+
+ metaClient = HoodieTableMetaClient.builder()
+
.setConf(context.getStorageConf().newInstance()).setBasePath(cfg.getBasePath())
+ .setLayoutVersion(Option.of(new
TimelineLayoutVersion(cfg.getTimelineLayoutVersion()))).build();
+ assertTableVersionOnDataAndMetadataTable(metaClient,
HoodieTableVersion.TWO);
+ assertNull(metaClient.getTableConfig().getPreCombineField());
}
@ParameterizedTest
@@ -476,13 +564,28 @@ public class TestUpgradeDowngrade extends
HoodieClientTestBase {
client.insert(writeRecords, commit1).collect();
}
+ 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 9ec852c20b40..c1cf5a98456d 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
@@ -18,7 +18,7 @@
package org.apache.hudi
import org.apache.hudi.client.SparkRDDWriteClient
-import org.apache.hudi.common.model.{HoodieFileFormat, HoodieRecord,
HoodieRecordPayload, HoodieTableType, WriteOperationType}
+import org.apache.hudi.common.model.{DefaultHoodieRecordPayload,
HoodieFileFormat, HoodieRecord, HoodieRecordPayload, HoodieTableType,
WriteOperationType}
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
TableSchemaResolver}
import org.apache.hudi.common.testutils.HoodieTestDataGenerator
import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieIndexConfig,
HoodieWriteConfig}
@@ -58,6 +58,54 @@ 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
+ * recorded field it falls back to the "ts" default and an older record
overwrites a newer one.
+ */
+ @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("version", createMetaClient(spark,
tempBasePath).getTableConfig.getPreCombineField)
+
+ // 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, "1")
+ 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.
*
