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 6ac51eb51807 fix(core): write the key generator class for tables below
version 8 (#20276)
6ac51eb51807 is described below
commit 6ac51eb518073d6a6fdccc200d38ffed45755925
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Oct 26 14:40:00 2026 -0700
fix(core): write the key generator class for tables below version 8 (#20276)
---
.../apache/hudi/client/BaseHoodieWriteClient.java | 35 ++++-
.../hudi/common/table/HoodieTableConfig.java | 18 +++
.../hudi/common/testutils/HoodieTestUtils.java | 15 +++
.../org/apache/hudi/utils/TestStreamerUtil.java | 99 ++++++++------
.../common/table/TestHoodieTableMetaClient.java | 74 +++++++++--
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 24 ----
.../hudi/TestMultipleTableVersionWriting.scala | 17 ---
.../apache/hudi/functional/TestCOWDataSource.scala | 131 +++++++++++++-----
.../apache/hudi/functional/TestMORDataSource.scala | 52 ++++----
.../spark/sql/hudi/ddl/TestCreateTable.scala | 148 +++++++++++++--------
.../deltastreamer/TestHoodieDeltaStreamer.java | 74 ++++++++++-
11 files changed, 478 insertions(+), 209 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index d93108c35b1f..26f26900b201 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -116,6 +116,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -1143,7 +1144,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
*/
@VisibleForTesting
String startCommit(Option<String> providedInstantTime, String actionType,
HoodieTableMetaClient metaClient) {
- if (needsUpgrade(metaClient)) {
+ if (needsUpgradeOrBackfill(metaClient)) {
// unclear what instant to use, since upgrade does have a given instant.
executeUsingTxnManager(Option.empty(), () -> tryUpgrade(metaClient,
Option.empty()));
}
@@ -1475,7 +1476,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
ownerInstant =
Option.of(metaClient.createNewInstant(HoodieInstant.State.INFLIGHT,
CommitUtils.getCommitActionType(operationType,
metaClient.getTableType()), instantTime.get()));
}
- boolean requiresInitTable = needsUpgrade(metaClient) ||
config.isMetadataTableEnabled();
+ boolean requiresInitTable = needsUpgradeOrBackfill(metaClient) ||
config.isMetadataTableEnabled();
if (requiresInitTable) {
executeUsingTxnManager(ownerInstant, () -> {
tryUpgrade(metaClient, instantTime);
@@ -1695,8 +1696,10 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
}
/**
- * Upgrades the hoodie table if need be when moving to a new Hudi version.
- * This method is called within a lock. Try to avoid double locking from
within this method.
+ * Upgrades the hoodie table if need be when moving to a new Hudi version,
and backfills the table configs
+ * that readers of the table version need but the table does not carry.
+ * This method is called within a lock, except when the Flink stream write
coordinator calls it on start.
+ * Try to avoid double locking from within this method.
* @param metaClient instance of {@link HoodieTableMetaClient} to use.
* @param instantTime instant time of interest if we have one.
*/
@@ -1721,11 +1724,31 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
metaClient.reloadTableConfig();
metaClient.reloadActiveTimeline();
}
+
+ backfillTableConfigs(metaClient);
+ }
+
+ private void backfillTableConfigs(HoodieTableMetaClient metaClient) {
+ if
(!HoodieTableConfig.getMissingKeyGeneratorClass(metaClient.getTableConfig()).isPresent())
{
+ return;
+ }
+ // another writer may have backfilled the table since this meta client
loaded its config
+ metaClient.reloadTableConfig();
+ Option<String> missingKeyGeneratorClass =
HoodieTableConfig.getMissingKeyGeneratorClass(metaClient.getTableConfig());
+ if (missingKeyGeneratorClass.isPresent()) {
+ log.info("Backfilling {}={} for table {}",
HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key(),
+ missingKeyGeneratorClass.get(), metaClient.getBasePath());
+ Properties updatedProps = new Properties();
+
updatedProps.setProperty(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key(),
missingKeyGeneratorClass.get());
+ HoodieTableConfig.update(metaClient.getStorage(),
metaClient.getMetaPath(), updatedProps);
+ metaClient.reloadTableConfig();
+ }
}
- private boolean needsUpgrade(HoodieTableMetaClient metaClient) {
+ private boolean needsUpgradeOrBackfill(HoodieTableMetaClient metaClient) {
UpgradeDowngrade upgradeDowngrade = new UpgradeDowngrade(metaClient,
config, context, upgradeDowngradeHelper);
- return upgradeDowngrade.needsUpgrade(config.getWriteVersion());
+ return upgradeDowngrade.needsUpgrade(config.getWriteVersion())
+ ||
HoodieTableConfig.getMissingKeyGeneratorClass(metaClient.getTableConfig()).isPresent();
}
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
index 538bc467a393..6c9e935b878e 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
@@ -701,12 +701,30 @@ public class HoodieTableConfig extends HoodieConfig {
hoodieConfig.setValue(POPULATE_META_FIELDS,
Boolean.toString(metaFieldsMode.toLegacyPopulateMetaFields()));
}
+ getMissingKeyGeneratorClass(hoodieConfig).ifPresent(
+ keyGeneratorClass -> hoodieConfig.setValue(KEY_GENERATOR_CLASS_NAME,
keyGeneratorClass));
dropInvalidConfigs(hoodieConfig);
storeProperties(hoodieConfig.getProps(), outputStream, propertyPath);
}
}
+ /**
+ * Returns the key generator class that a table below version 8 records only
through its key generator type.
+ * Hudi 0.x reads only the class, so such a table must carry it as well.
+ */
+ public static Option<String> getMissingKeyGeneratorClass(HoodieConfig
config) {
+ if (!getTableVersion(config).lesserThan(HoodieTableVersion.EIGHT)
+ || config.contains(KEY_GENERATOR_CLASS_NAME) ||
!config.contains(KEY_GENERATOR_TYPE)) {
+ return Option.empty();
+ }
+ String keyGeneratorType = config.getString(KEY_GENERATOR_TYPE);
+ return Option.fromJavaOptional(Arrays.stream(KeyGeneratorType.values())
+ .filter(type -> type != KeyGeneratorType.USER_PROVIDED &&
type.name().equals(keyGeneratorType))
+ .map(KeyGeneratorType::getClassName)
+ .findFirst());
+ }
+
public static long generateChecksum(Properties props) {
if (!props.containsKey(NAME.key())) {
throw new IllegalArgumentException(NAME.key() + " property needs to be
specified");
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
index ae9deef4f7e5..a01ea6908bf9 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
@@ -86,6 +86,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
+import java.util.TreeMap;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.zip.ZipEntry;
@@ -554,6 +555,20 @@ public class HoodieTestUtils {
}
}
+ /**
+ * Asserts that hoodie.properties holds exactly the expected table configs,
ignoring the checksum and the
+ * create schema. The table-config tests of each engine pin table creation
per table version through this
+ * check, so a table config change updates their expected configs instead of
adding a new test.
+ */
+ public static void assertTableConfigs(HoodieStorage storage, String
basePath, Map<String, String> expectedConfigs) {
+ Map<String, String> actualConfigs = new TreeMap<>();
+ HoodieTableConfig.loadFromHoodieProps(storage, basePath).getProps()
+ .forEach((key, value) -> actualConfigs.put((String) key, (String)
value));
+ actualConfigs.remove(HoodieTableConfig.TABLE_CHECKSUM.key());
+ actualConfigs.remove(HoodieTableConfig.CREATE_SCHEMA.key());
+ assertEquals(new TreeMap<>(expectedConfigs), actualConfigs);
+ }
+
public static void validateTableConfig(HoodieStorage storage,
String basePath,
Map<String, String> expectedConfigs,
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
index f917593dd8de..41e81bd8066e 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
@@ -50,9 +50,9 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.io.util.FileIOUtils;
import org.apache.hudi.keygen.ComplexAvroKeyGenerator;
-import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
import org.apache.hudi.sink.FlinkCheckpointClient;
import org.apache.hudi.sink.muttley.AthenaIngestionGateway;
+import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.streamer.FlinkStreamerConfig;
@@ -65,6 +65,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.NullSource;
import org.mockito.Mockito;
import java.io.File;
@@ -77,9 +78,9 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.assertTableConfigs;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -185,26 +186,6 @@ class TestStreamerUtil {
assertTrue(StreamerUtil.getIndexConfig(conf).getBoolean(HoodieIndexConfig.BUCKET_PARTITIONER));
}
- @Test
- void testInitTableWithSpecificVersion() throws IOException {
- Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
-
- // Test for partitioned table.
- conf.set(FlinkOptions.PARTITION_PATH_FIELD, "p0,p1");
- conf.set(FlinkOptions.WRITE_TABLE_VERSION,
HoodieTableVersion.SIX.versionCode());
- conf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
- HoodieTableConfig.TableStorageLayout.DEFAULT.configValue());
- StreamerUtil.initTableIfNotExists(conf);
-
- // Validate the partition fields & preCombineField in hoodie.properties.
- HoodieTableMetaClient metaClient1 =
HoodieTestUtils.createMetaClient(tempFile.getAbsolutePath());
- assertArrayEquals(metaClient1.getTableConfig().getPartitionFields().get(),
new String[] {"p0", "p1"});
- assertNotNull(metaClient1.getTableConfig().getKeyGeneratorClassName());
- assertEquals(HoodieTableVersion.SIX,
metaClient1.getTableConfig().getTableVersion());
- assertEquals(HoodieTableConfig.TableStorageLayout.DEFAULT.configValue(),
- conf.getString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(), null));
- }
-
@Test
void testInitLsmTreeTable() throws IOException {
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
@@ -235,32 +216,74 @@ class TestStreamerUtil {
assertTrue(metaClient.getTableConfig().isLSMTreeStorageLayout());
}
- @Test
- void testInitTableIfNotExists() throws IOException {
+ /**
+ * A null version leaves the write table version unset, so the default
version is covered as well.
+ */
+ @ParameterizedTest
+ @NullSource
+ @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "EIGHT",
"NINE", "TEN"})
+ void testInitTableIfNotExists(HoodieTableVersion version) throws IOException
{
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ if (version != null) {
+ conf.set(FlinkOptions.WRITE_TABLE_VERSION, version.versionCode());
+ }
+ HoodieTableVersion expectedVersion = version == null
+ ?
HoodieTableVersion.fromVersionCode(FlinkOptions.WRITE_TABLE_VERSION.defaultValue())
: version;
+ conf.set(FlinkOptions.ORDERING_FIELDS, "ts");
// Test for partitioned table.
- conf.set(FlinkOptions.ORDERING_FIELDS, "ts");
conf.set(FlinkOptions.PARTITION_PATH_FIELD, "p0,p1");
StreamerUtil.initTableIfNotExists(conf);
-
- // Validate the partition fields & preCombineField in hoodie.properties.
- HoodieTableMetaClient metaClient1 =
HoodieTestUtils.createMetaClient(tempFile.getAbsolutePath());
- assertTrue(metaClient1.getTableConfig().getPartitionFields().isPresent(),
- "Missing partition columns in the hoodie.properties.");
- assertArrayEquals(metaClient1.getTableConfig().getPartitionFields().get(),
new String[] {"p0", "p1"});
- assertEquals(metaClient1.getTableConfig().getOrderingFieldsStr().get(),
"ts");
- assertEquals(metaClient1.getTableConfig().getKeyGeneratorClassName(),
SimpleAvroKeyGenerator.class.getName());
- assertEquals(HoodieTableVersion.current(),
metaClient1.getTableConfig().getTableVersion());
- assertTrue(metaClient1.getTableConfig().isLSMTreeStorageLayout());
+ HoodieStorage storage =
HoodieTestUtils.createMetaClient(tempFile.getAbsolutePath()).getStorage();
+ assertTableConfigs(storage, tempFile.getAbsolutePath(),
expectedTableConfigs(expectedVersion, true));
// Test for non-partitioned table.
conf.removeConfig(FlinkOptions.PARTITION_PATH_FIELD);
FileIOUtils.deleteDirectory(tempFile);
StreamerUtil.initTableIfNotExists(conf);
- HoodieTableMetaClient metaClient2 =
HoodieTestUtils.createMetaClient(tempFile.getAbsolutePath());
- assertFalse(metaClient2.getTableConfig().getPartitionFields().isPresent());
- assertEquals(metaClient2.getTableConfig().getKeyGeneratorClassName(),
SimpleAvroKeyGenerator.class.getName());
+ assertTableConfigs(storage, tempFile.getAbsolutePath(),
expectedTableConfigs(expectedVersion, false));
+ }
+
+ private static Map<String, String> expectedTableConfigs(HoodieTableVersion
version, boolean partitioned) {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("hoodie.table.name", "TestHoodieTable");
+ configs.put("hoodie.table.type", "COPY_ON_WRITE");
+ configs.put("hoodie.table.version", String.valueOf(version.versionCode()));
+ configs.put("hoodie.table.recordkey.fields", "uuid");
+ configs.put("hoodie.table.ordering.fields", "ts");
+ configs.put("hoodie.table.keygenerator.type", "SIMPLE_AVRO");
+ configs.put("hoodie.table.format", "native");
+ configs.put("hoodie.table.cdc.enabled", "false");
+ configs.put("hoodie.meta.fields.mode", "ALL");
+ configs.put("hoodie.datasource.write.drop.partition.columns", "false");
+ configs.put("hoodie.datasource.write.hive_style_partitioning", "false");
+ configs.put("hoodie.datasource.write.partitionpath.urlencode", "false");
+ configs.put("hoodie.archivelog.folder", "history");
+ configs.put("hoodie.timeline.path", "timeline");
+ configs.put("hoodie.timeline.history.path", "history");
+ if (partitioned) {
+ configs.put("hoodie.table.partition.fields", "p0,p1");
+ }
+ if (version.lesserThan(HoodieTableVersion.EIGHT)) {
+ configs.put("hoodie.timeline.layout.version", "1");
+ configs.put("hoodie.table.keygenerator.class",
"org.apache.hudi.keygen.SimpleAvroKeyGenerator");
+ } else {
+ configs.put("hoodie.timeline.layout.version", "2");
+ configs.put("hoodie.table.initial.version",
String.valueOf(version.versionCode()));
+ configs.put("hoodie.record.merge.mode", "EVENT_TIME_ORDERING");
+ }
+ if (version.lesserThan(HoodieTableVersion.NINE)) {
+ configs.put("hoodie.compaction.payload.class",
"org.apache.hudi.common.model.EventTimeAvroPayload");
+ configs.put("hoodie.record.merge.strategy.id",
"eeb8d96f-b1e4-49fd-bbf8-28ac514178e5");
+ } else {
+ configs.put("hoodie.table.legacy.payload.class",
"org.apache.hudi.common.model.EventTimeAvroPayload");
+ }
+ if (version.lesserThan(HoodieTableVersion.TEN)) {
+ configs.put("hoodie.populate.meta.fields", "true");
+ } else {
+ configs.put("hoodie.table.storage.layout", "lsm_tree");
+ }
+ return configs;
}
@Test
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
index 6ee96663cd63..ccf8335f6e53 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
@@ -43,6 +43,9 @@ import org.apache.hadoop.fs.Path;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -58,8 +61,10 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import java.util.stream.Stream;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.assertTableConfigs;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -292,21 +297,68 @@ class TestHoodieTableMetaClient extends
HoodieCommonTestHarness {
}
}
- @Test
- void testTableVersion() throws IOException {
+ /**
+ * A null version leaves the table version unset on the builder, so the
default version is covered as well.
+ */
+ @ParameterizedTest
+ @MethodSource("tableVersionsAndKeyGeneratorSettings")
+ void testTableConfigsBasedOnVersion(HoodieTableVersion version, boolean
setKeyGeneratorType) throws IOException {
final String basePath = tempDir.toAbsolutePath() + Path.SEPARATOR + "t1";
- HoodieTableMetaClient metaClient1 = HoodieTableMetaClient.newTableBuilder()
+ HoodieTableMetaClient.TableBuilder tableBuilder =
HoodieTableMetaClient.newTableBuilder()
.setTableType(HoodieTableType.MERGE_ON_READ.name())
.setTableName("table-version-test")
- .setTableVersion(HoodieTableVersion.SIX.versionCode())
- .initTable(this.metaClient.getStorageConf(), basePath);
- assertEquals(HoodieTableVersion.SIX,
metaClient1.getTableConfig().getTableVersion());
+ .setRecordKeyFields("key")
+ .setPartitionFields("partition")
+ .setOrderingFields("ts");
+ if (version != null) {
+ tableBuilder.setTableVersion(version.versionCode());
+ }
+ if (setKeyGeneratorType) {
+ tableBuilder.setKeyGeneratorType(KeyGeneratorType.SIMPLE.name());
+ } else {
+
tableBuilder.setKeyGeneratorClassProp(KeyGeneratorType.SIMPLE.getClassName());
+ }
+ HoodieTableVersion expectedVersion = version == null ?
HoodieTableVersion.current() : version;
+ HoodieTableMetaClient createdMetaClient =
tableBuilder.initTable(this.metaClient.getStorageConf(), basePath);
+ assertEquals(expectedVersion,
createdMetaClient.getTableConfig().getTableVersion());
+ assertTableConfigs(createdMetaClient.getStorage(), basePath,
expectedTableConfigs(expectedVersion));
+ }
- HoodieTableMetaClient metaClient2 = HoodieTableMetaClient.builder()
- .setConf(this.metaClient.getStorageConf())
- .setBasePath(basePath)
- .build();
- assertEquals(HoodieTableVersion.SIX,
metaClient2.getTableConfig().getTableVersion());
+ private static Stream<Arguments> tableVersionsAndKeyGeneratorSettings() {
+ return Stream.of(null, HoodieTableVersion.SIX, HoodieTableVersion.EIGHT,
HoodieTableVersion.NINE, HoodieTableVersion.TEN)
+ .flatMap(version -> Stream.of(Arguments.of(version, false),
Arguments.of(version, true)));
+ }
+
+ private static Map<String, String> expectedTableConfigs(HoodieTableVersion
version) {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("hoodie.table.name", "table-version-test");
+ configs.put("hoodie.table.type", "MERGE_ON_READ");
+ configs.put("hoodie.table.version", String.valueOf(version.versionCode()));
+ configs.put("hoodie.table.recordkey.fields", "key");
+ configs.put("hoodie.table.partition.fields", "partition");
+ configs.put("hoodie.table.ordering.fields", "ts");
+ configs.put("hoodie.table.keygenerator.type", "SIMPLE");
+ configs.put("hoodie.meta.fields.mode", "ALL");
+ configs.put("hoodie.datasource.write.drop.partition.columns", "false");
+ configs.put("hoodie.archivelog.folder", "archived");
+ configs.put("hoodie.timeline.path", "timeline");
+ configs.put("hoodie.timeline.history.path", "history");
+ if (version.lesserThan(HoodieTableVersion.EIGHT)) {
+ configs.put("hoodie.timeline.layout.version", "1");
+ configs.put("hoodie.table.keygenerator.class",
"org.apache.hudi.keygen.SimpleKeyGenerator");
+ } else {
+ configs.put("hoodie.timeline.layout.version", "2");
+ configs.put("hoodie.table.initial.version",
String.valueOf(version.versionCode()));
+ configs.put("hoodie.record.merge.mode", "EVENT_TIME_ORDERING");
+ }
+ if (version.lesserThan(HoodieTableVersion.NINE)) {
+ configs.put("hoodie.compaction.payload.class",
"org.apache.hudi.common.model.DefaultHoodieRecordPayload");
+ configs.put("hoodie.record.merge.strategy.id",
"eeb8d96f-b1e4-49fd-bbf8-28ac514178e5");
+ }
+ if (version.lesserThan(HoodieTableVersion.TEN)) {
+ configs.put("hoodie.populate.meta.fields", "true");
+ }
+ return configs;
}
@Test
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 7177e6e527bc..7904e9b4abde 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
@@ -294,30 +294,6 @@ class TestHoodieSparkSqlWriter extends
HoodieSparkWriterTestBase {
assert(HoodieSparkSqlWriter.write(sqlContext, SaveMode.Overwrite,
tableModifier2, dataFrame2)._1)
}
- /**
- * Test case for do not let the parttitonpath field change
- */
- @Test
- def testChangeWriteTableVersion(): Unit = {
- Seq(6, 8).foreach { tableVersion =>
- val tempPath = s"$tempBasePath/${tableVersion}"
- val tableModifier1 = Map(
- "path" -> tempPath,
- HoodieWriteConfig.TBL_NAME.key -> hoodieFooTableName,
- "hoodie.write.table.version" -> s"$tableVersion",
- "hoodie.datasource.write.recordkey.field" -> "uuid",
- "hoodie.datasource.write.partitionpath.field" -> "ts"
- )
- val dataFrame =
spark.createDataFrame(Seq(StringLongTest(UUID.randomUUID().toString, new
Date().getTime)))
- HoodieSparkSqlWriter.write(sqlContext, SaveMode.Overwrite,
tableModifier1, dataFrame)
-
- // Make sure table version is adopted.
- val metaClient = HoodieTableMetaClient.builder().setBasePath(tempPath)
-
.setConf(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf())).build()
- assertEquals(metaClient.getTableConfig.getTableVersion.versionCode(),
tableVersion)
- }
- }
-
/**
* Test case for each bulk insert sort mode
*
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestMultipleTableVersionWriting.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestMultipleTableVersionWriting.scala
index 6c965a28d15e..3c59fe24ac17 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestMultipleTableVersionWriting.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestMultipleTableVersionWriting.scala
@@ -23,7 +23,6 @@ package org.apache.hudi
import org.apache.hudi.common.model.HoodieTableType
import org.apache.hudi.common.table.HoodieTableVersion
import org.apache.hudi.common.testutils.HoodieTestUtils
-import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.SaveMode
import org.junit.jupiter.api.Assertions.assertEquals
@@ -31,22 +30,6 @@ import org.junit.jupiter.api.Test
class TestMultipleTableVersionWriting extends HoodieSparkWriterTestBase {
- @Test
- def testTableVersionAndWriteVersionMatching(): Unit = {
- val basePath = s"$tempBasePath/tbl_1"
- val df = spark.range(1).selectExpr("1 as id", "1 as name", "1 as
partition")
-
- // write table with current version
- df.write.format("hudi")
- .option(HoodieWriteConfig.TBL_NAME.key, "tbl_1")
- .mode(SaveMode.Overwrite)
- .save(basePath)
-
- val metaClient = HoodieTestUtils.createMetaClient(basePath)
- assertEquals(HoodieTableVersion.current().versionCode(),
- metaClient.getTableConfig.getTableVersion.versionCode())
- }
-
@Test
def testThrowsExceptionForIncompatibleTableVersion(): Unit = {
val basePath = s"$tempBasePath/tbl_2"
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
index 38cdbb3af6c6..3bf0e029b072 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
@@ -34,7 +34,7 @@ import org.apache.hudi.common.table.{HoodieTableConfig,
HoodieTableMetaClient, H
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline,
TimelineUtils}
import org.apache.hudi.common.testutils.{HoodieTestDataGenerator,
HoodieTestUtils}
import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{deleteRecordsToStrings,
recordsToStrings}
-import
org.apache.hudi.common.testutils.HoodieTestUtils.{INSTANT_FILE_NAME_GENERATOR,
INSTANT_GENERATOR}
+import org.apache.hudi.common.testutils.HoodieTestUtils.{assertTableConfigs,
INSTANT_FILE_NAME_GENERATOR, INSTANT_GENERATOR}
import org.apache.hudi.common.util.{ClusteringUtils, Option}
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.exception.HoodieException
@@ -144,15 +144,71 @@ class TestCOWDataSource extends HoodieSparkClientTestBase
with ScalaAssertionSup
.options(writeOptions).mode(SaveMode.Overwrite).save(basePath)
metaClient =
HoodieTableMetaClient.builder.setConf(storageConf).setBasePath(basePath).build
// If no write version is specified, use current.
- if (!targetTableVersion.equals("null")) {
- assertEquals(
-
HoodieTableVersion.fromVersionCode(Integer.valueOf(targetTableVersion)),
- metaClient.getTableConfig.getTableVersion)
+ val expectedTableVersion = if (targetTableVersion.equals("null")) {
+ HoodieTableVersion.current
} else {
- // Otherwise, the table version is the target table version.
- assertEquals(HoodieTableVersion.current,
metaClient.getTableConfig.getTableVersion)
+ HoodieTableVersion.fromVersionCode(Integer.valueOf(targetTableVersion))
}
+ assertEquals(expectedTableVersion,
metaClient.getTableConfig.getTableVersion)
+ assertTableConfigs(storage, basePath,
expectedTableConfigs(expectedTableVersion).asJava)
+
+ // the next write backfills the key generator class of a table created
without it
+ HoodieTableConfig.delete(storage, metaClient.getMetaPath,
+
java.util.Collections.singleton(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key))
+ inputDF.write.format("hudi").partitionBy("partition")
+ .options(writeOptions).mode(SaveMode.Append).save(basePath)
+ assertTableConfigs(storage, basePath,
expectedTableConfigs(expectedTableVersion).asJava)
+ }
+ }
+
+ private def expectedTableConfigs(tableVersion: HoodieTableVersion):
Map[String, String] = {
+ val commonConfigs = Map(
+ "hoodie.table.name" -> "testTableCreation",
+ "hoodie.table.type" -> "COPY_ON_WRITE",
+ "hoodie.table.version" -> tableVersion.versionCode().toString,
+ "hoodie.table.partition.fields" -> "partition",
+ "hoodie.table.keygenerator.type" -> "SIMPLE",
+ "hoodie.table.base.file.format" -> "PARQUET",
+ "hoodie.table.format" -> "native",
+ "hoodie.table.cdc.enabled" -> "false",
+ "hoodie.table.timeline.timezone" -> "LOCAL",
+ "hoodie.table.metadata.partitions.inflight" -> "",
+ "hoodie.table.index.defs.path" -> ".hoodie/.index_defs/index.json",
+ "hoodie.table.partition_extractor_class" ->
"org.apache.hudi.hive.SinglePartPartitionValueExtractor",
+ "hoodie.meta.fields.mode" -> "ALL",
+ "hoodie.partition.metafile.use.base.format" -> "false",
+ "hoodie.datasource.write.drop.partition.columns" -> "false",
+ "hoodie.datasource.write.hive_style_partitioning" -> "false",
+ "hoodie.datasource.write.partitionpath.urlencode" -> "false",
+ "hoodie.archivelog.folder" -> "history",
+ "hoodie.timeline.path" -> "timeline",
+ "hoodie.timeline.history.path" -> "history")
+ val layoutConfigs = if (tableVersion.lesserThan(HoodieTableVersion.EIGHT))
{
+ Map(
+ "hoodie.timeline.layout.version" -> "1",
+ "hoodie.table.keygenerator.class" ->
"org.apache.hudi.keygen.SimpleKeyGenerator",
+ "hoodie.table.metadata.partitions" -> "column_stats,files")
+ } else {
+ Map(
+ "hoodie.timeline.layout.version" -> "2",
+ "hoodie.table.initial.version" -> tableVersion.versionCode().toString,
+ "hoodie.record.merge.mode" -> "COMMIT_TIME_ORDERING",
+ "hoodie.table.multiple.base.file.formats.enable" -> "false",
+ "hoodie.table.metadata.partitions" ->
"column_stats,files,partition_stats")
+ }
+ val mergeConfigs = if (tableVersion.lesserThan(HoodieTableVersion.NINE)) {
+ Map(
+ "hoodie.compaction.payload.class" ->
"org.apache.hudi.common.model.OverwriteWithLatestAvroPayload",
+ "hoodie.record.merge.strategy.id" ->
"ce9acb64-bde0-424c-9b91-f6ebba25356d")
+ } else {
+ Map.empty[String, String]
}
+ val metaFieldsConfigs = if
(tableVersion.lesserThan(HoodieTableVersion.TEN)) {
+ Map("hoodie.populate.meta.fields" -> "true")
+ } else {
+ Map("hoodie.table.storage.layout" -> "default")
+ }
+ commonConfigs ++ layoutConfigs ++ mergeConfigs ++ metaFieldsConfigs
}
@ParameterizedTest
@@ -2527,9 +2583,14 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
@ParameterizedTest
@MethodSource(Array("provideParamsForKeyGenTest"))
def testUserProvidedKeyGeneratorClass(keyGenClass: Option[String],
- keyGenType: Option[String]): Unit = {
+ keyGenType: Option[String],
+ tableVersion:
Option[HoodieTableVersion]): Unit = {
val recordType = HoodieRecordType.AVRO
- var opts: Map[String, String] = Map()
+ // the table is new, so the guard against a legacy single-field complex
key encoding below version 9 does not apply
+ var opts: Map[String, String] =
Map(HoodieWriteConfig.ENABLE_COMPLEX_KEYGEN_VALIDATION.key -> "false")
+ if (tableVersion.isPresent) {
+ opts = opts ++ Map(HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
tableVersion.get.versionCode.toString)
+ }
if (keyGenClass.isPresent) {
opts = opts ++ Map(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key ->
keyGenClass.get)
}
@@ -2560,14 +2621,19 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
.sort("_row_key")
var actualKeyDF =
actualDF.select("_hoodie_record_key").sort("_hoodie_record_key")
assertTrue(inputKeyDF.except(actualKeyDF).isEmpty &&
actualKeyDF.except(inputKeyDF).isEmpty)
- val metaClient = getHoodieMetaClient(storageConf, basePath)
- val actualKeyGenType = metaClient.getTableConfig
- .getProps.getString(HoodieTableConfig.KEY_GENERATOR_TYPE.key, null)
- assertEquals(expectedKeyGenType, actualKeyGenType)
- // For USER_PROVIDED type, the class should exist in table config.
- if (KeyGeneratorType.USER_PROVIDED.name == actualKeyGenType) {
- assertEquals(keyGenClass.get(),
metaClient.getTableConfig.getKeyGeneratorClassName)
+ val tableConfig = getHoodieMetaClient(storageConf, basePath).getTableConfig
+ val expectedTableVersion =
TestCOWDataSource.writeTableVersionOrDefault(tableVersion)
+ assertEquals(expectedTableVersion, tableConfig.getTableVersion)
+ assertEquals(expectedKeyGenType,
tableConfig.getProps.getString(HoodieTableConfig.KEY_GENERATOR_TYPE.key, null))
+ // A user-provided key generator is persisted as a class at every table
version, the others below version 8.
+ val expectedKeyGenClass = if (KeyGeneratorType.USER_PROVIDED.name ==
expectedKeyGenType) {
+ keyGenClass.get
+ } else if (expectedTableVersion.lesserThan(HoodieTableVersion.EIGHT)) {
+ KeyGeneratorType.valueOf(expectedKeyGenType).getClassName
+ } else {
+ null
}
+ assertEquals(expectedKeyGenClass,
tableConfig.getProps.getString(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key,
null))
// First update.
val firstUpdate =
recordsToStrings(dataGen.generateUpdatesForAllRecords("001")).asScala.toList
@@ -2950,6 +3016,14 @@ object TestCOWDataSource {
s"Top-level level='L1' should return key1 for $tableType")
}
+ /**
+ * The table version a write with the given write table version produces,
falling back to the default write
+ * version when it is unset so that a test arm without one follows a change
of the default.
+ */
+ def writeTableVersionOrDefault(tableVersion: Option[HoodieTableVersion]):
HoodieTableVersion = {
+
tableVersion.orElse(HoodieTableVersion.fromVersionCode(HoodieWriteConfig.WRITE_TABLE_VERSION.defaultValue()))
+ }
+
def convertColumnsToNullable(df: DataFrame, cols: String*): DataFrame = {
cols.foldLeft(df) { (df, c) =>
// NOTE: This is the trick to make Spark convert a non-null column "c"
into a nullable
@@ -2971,26 +3045,21 @@ object TestCOWDataSource {
def tableVersionCreationTestCases = {
val autoUpgradeValues = Array("true", "false")
- val targetVersions = Array("1", "2", "3", "4", "5", "6", "7", "8", "9",
"null")
+ val targetVersions = Array("1", "2", "3", "4", "5", "6", "7", "8", "9",
"10", "null")
autoUpgradeValues.flatMap(
(autoUpgrade: String) => targetVersions.map(
(targetVersion: String) => Arguments.of(autoUpgrade, targetVersion)))
}
def provideParamsForKeyGenTest(): java.util.List[Arguments] = {
- java.util.Arrays.asList(
- Arguments.of(
- Option.of("org.apache.hudi.keygen.MockUserProvidedKeyGenerator"),
- Option.of(KeyGeneratorType.USER_PROVIDED.name())),
- Arguments.of(
- Option.empty(),
- Option.of(KeyGeneratorType.SIMPLE.name())),
- Arguments.of(
- Option.of("org.apache.hudi.keygen.SimpleAvroKeyGenerator"),
- Option.of(KeyGeneratorType.SIMPLE.name())),
- Arguments.of(
- Option.of("org.apache.hudi.keygen.ComplexKeyGenerator"),
- Option.empty())
- )
+ val keyGenSettings = Seq(
+ (Option.of("org.apache.hudi.keygen.MockUserProvidedKeyGenerator"),
Option.of(KeyGeneratorType.USER_PROVIDED.name())),
+ (Option.empty[String](), Option.of(KeyGeneratorType.SIMPLE.name())),
+ (Option.of("org.apache.hudi.keygen.SimpleAvroKeyGenerator"),
Option.of(KeyGeneratorType.SIMPLE.name())),
+ (Option.of("org.apache.hudi.keygen.ComplexKeyGenerator"),
Option.empty[String]()))
+ val tableVersions = Seq(Option.of(HoodieTableVersion.SIX),
Option.of(HoodieTableVersion.EIGHT),
+ Option.of(HoodieTableVersion.NINE), Option.of(HoodieTableVersion.TEN),
Option.empty[HoodieTableVersion]())
+ (for ((keyGenClass, keyGenType) <- keyGenSettings; tableVersion <-
tableVersions)
+ yield Arguments.of(keyGenClass, keyGenType, tableVersion)).asJava
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
index 359e1fd54fc6..29185b6e1a12 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
@@ -2462,12 +2462,18 @@ class TestMORDataSource extends
HoodieSparkClientTestBase with SparkDatasetMixin
@ParameterizedTest
@MethodSource(Array("provideParamsForKeyGenTest"))
def testUserProvidedKeyGeneratorClass(keyGenClass: Option[String],
- keyGenType: Option[String]): Unit = {
+ keyGenType: Option[String],
+ tableVersion:
Option[HoodieTableVersion]): Unit = {
val recordType = HoodieRecordType.AVRO
+ // the table is new, so the guard against a legacy single-field complex
key encoding below version 9 does not apply
var opts: Map[String, String] = Map(
DataSourceWriteOptions.TABLE_TYPE.key ->
DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL,
- HoodieWriteConfig.MERGE_SMALL_FILE_GROUP_CANDIDATES_LIMIT.key -> "0"
+ HoodieWriteConfig.MERGE_SMALL_FILE_GROUP_CANDIDATES_LIMIT.key -> "0",
+ HoodieWriteConfig.ENABLE_COMPLEX_KEYGEN_VALIDATION.key -> "false"
)
+ if (tableVersion.isPresent) {
+ opts = opts ++ Map(HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
tableVersion.get.versionCode.toString)
+ }
if (keyGenClass.isPresent) {
opts = opts ++ Map(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key ->
keyGenClass.get)
}
@@ -2498,14 +2504,19 @@ class TestMORDataSource extends
HoodieSparkClientTestBase with SparkDatasetMixin
.sort("_row_key")
var actualKeyDF =
actualDF.select("_hoodie_record_key").sort("_hoodie_record_key")
assertTrue(inputKeyDF.except(actualKeyDF).isEmpty &&
actualKeyDF.except(inputKeyDF).isEmpty)
- val metaClient = getHoodieMetaClient(storageConf, basePath)
- val actualKeyGenType = metaClient.getTableConfig
- .getProps.getString(HoodieTableConfig.KEY_GENERATOR_TYPE.key, null)
- assertEquals(expectedKeyGenType, actualKeyGenType)
- // For USER_PROVIDED type, the class should exist in table config.
- if (KeyGeneratorType.USER_PROVIDED.name == actualKeyGenType) {
- assertEquals(keyGenClass.get(),
metaClient.getTableConfig.getKeyGeneratorClassName)
+ val tableConfig = getHoodieMetaClient(storageConf, basePath).getTableConfig
+ val expectedTableVersion =
TestCOWDataSource.writeTableVersionOrDefault(tableVersion)
+ assertEquals(expectedTableVersion, tableConfig.getTableVersion)
+ assertEquals(expectedKeyGenType,
tableConfig.getProps.getString(HoodieTableConfig.KEY_GENERATOR_TYPE.key, null))
+ // A user-provided key generator is persisted as a class at every table
version, the others below version 8.
+ val expectedKeyGenClass = if (KeyGeneratorType.USER_PROVIDED.name ==
expectedKeyGenType) {
+ keyGenClass.get
+ } else if (expectedTableVersion.lesserThan(HoodieTableVersion.EIGHT)) {
+ KeyGeneratorType.valueOf(expectedKeyGenType).getClassName
+ } else {
+ null
}
+ assertEquals(expectedKeyGenClass,
tableConfig.getProps.getString(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key,
null))
// First update.
val firstUpdate =
recordsToStrings(dataGen.generateUpdatesForAllRecords("001")).asScala.toList
@@ -2718,19 +2729,14 @@ class TestMORDataSource extends
HoodieSparkClientTestBase with SparkDatasetMixin
object TestMORDataSource {
def provideParamsForKeyGenTest(): java.util.List[Arguments] = {
- java.util.Arrays.asList(
- Arguments.of(
- Option.of("org.apache.hudi.keygen.MockUserProvidedKeyGenerator"),
- Option.of(KeyGeneratorType.USER_PROVIDED.name())),
- Arguments.of(
- Option.empty(),
- Option.of(KeyGeneratorType.SIMPLE.name())),
- Arguments.of(
- Option.of("org.apache.hudi.keygen.SimpleAvroKeyGenerator"),
- Option.of(KeyGeneratorType.SIMPLE.name())),
- Arguments.of(
- Option.of("org.apache.hudi.keygen.ComplexKeyGenerator"),
- Option.empty())
- )
+ val keyGenSettings = Seq(
+ (Option.of("org.apache.hudi.keygen.MockUserProvidedKeyGenerator"),
Option.of(KeyGeneratorType.USER_PROVIDED.name())),
+ (Option.empty[String](), Option.of(KeyGeneratorType.SIMPLE.name())),
+ (Option.of("org.apache.hudi.keygen.SimpleAvroKeyGenerator"),
Option.of(KeyGeneratorType.SIMPLE.name())),
+ (Option.of("org.apache.hudi.keygen.ComplexKeyGenerator"),
Option.empty[String]()))
+ val tableVersions = Seq(Option.of(HoodieTableVersion.SIX),
Option.of(HoodieTableVersion.EIGHT),
+ Option.of(HoodieTableVersion.NINE), Option.of(HoodieTableVersion.TEN),
Option.empty[HoodieTableVersion]())
+ (for ((keyGenClass, keyGenType) <- keyGenSettings; tableVersion <-
tableVersions)
+ yield Arguments.of(keyGenClass, keyGenType, tableVersion)).asJava
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
index a10ae3599e8d..3e3f3cf774ec 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
@@ -21,14 +21,15 @@ import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.model.{HoodieRecord, HoodieTableType,
WriteOperationType}
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaType}
-import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
HoodieTableVersion}
import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.common.testutils.HoodieTestUtils.assertTableConfigs
import org.apache.hudi.common.util.HoodieStorageUtils
import org.apache.hudi.common.util.PartitionPathEncodeUtils.escapePathName
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat
-import org.apache.hudi.keygen.constant.{ComplexKeyGenEncoding,
KeyGeneratorType}
+import org.apache.hudi.keygen.constant.ComplexKeyGenEncoding
import org.apache.hudi.storage.{HoodieStorage, StoragePath}
import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
import org.apache.hudi.testutils.Assertions
@@ -103,57 +104,100 @@ class TestCreateTable extends HoodieSparkSqlTestBase
with ExtendedParserTestHelp
spark.sql("use default")
}
- test("Test Create Hoodie Table With Options") {
- val tableName = generateTableName
- spark.sql(
- s"""
- | create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long,
- | dt string
- | ) using hudi
- | partitioned by (dt)
- | options (
- | hoodie.database.name = "databaseName",
- | hoodie.table.name = "tableName",
- | primaryKey = 'id',
- | orderingFields = 'ts',
- | hoodie.datasource.write.operation = 'upsert'
- | )
- """.stripMargin)
- val table =
spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName))
- assertResult(table.properties("type"))("cow")
- assertResult(table.properties("primaryKey"))("id")
- assertResult(table.properties("orderingFields"))("ts")
- assertResult(tableName)(table.identifier.table)
- assertResult("hudi")(table.provider.get)
- assertResult(CatalogTableType.MANAGED)(table.tableType)
- assertResult(
- HoodieRecord.HOODIE_META_COLUMNS.asScala.map(StructField(_, StringType))
- ++ Seq(
- StructField("id", IntegerType),
- StructField("name", StringType),
- StructField("price", DoubleType),
- StructField("ts", LongType),
- StructField("dt", StringType))
- )(table.schema.fields)
-
assertFalse(table.properties.contains(HoodieTableConfig.DATABASE_NAME.key()))
- assertFalse(table.properties.contains(HoodieTableConfig.NAME.key()))
- assertFalse(table.properties.contains(OPERATION.key()))
+ // the arm without a table version leaves the write table version unset, so
the default version is covered as well
+ Seq(Some(HoodieTableVersion.SIX), Some(HoodieTableVersion.EIGHT),
Some(HoodieTableVersion.NINE), Some(HoodieTableVersion.TEN), None).foreach {
tableVersion =>
+ test(s"Test Create Hoodie Table With Options
(tableVersion=${tableVersion.map(_.versionCode().toString).getOrElse("default")})")
{
+ val tableName = generateTableName
+ val tableVersionOption = tableVersion.map(v =>
s"hoodie.write.table.version = '${v.versionCode()}',").getOrElse("")
+ spark.sql(
+ s"""
+ | create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long,
+ | dt string
+ | ) using hudi
+ | partitioned by (dt)
+ | options (
+ | hoodie.database.name = "databaseName",
+ | hoodie.table.name = "tableName",
+ | $tableVersionOption
+ | primaryKey = 'id',
+ | orderingFields = 'ts',
+ | hoodie.datasource.write.operation = 'upsert'
+ | )
+ """.stripMargin)
+ val table =
spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName))
+ assertResult(table.properties("type"))("cow")
+ assertResult(table.properties("primaryKey"))("id")
+ assertResult(table.properties("orderingFields"))("ts")
+ assertResult(tableName)(table.identifier.table)
+ assertResult("hudi")(table.provider.get)
+ assertResult(CatalogTableType.MANAGED)(table.tableType)
+ assertResult(
+ HoodieRecord.HOODIE_META_COLUMNS.asScala.map(StructField(_,
StringType))
+ ++ Seq(
+ StructField("id", IntegerType),
+ StructField("name", StringType),
+ StructField("price", DoubleType),
+ StructField("ts", LongType),
+ StructField("dt", StringType))
+ )(table.schema.fields)
+
assertFalse(table.properties.contains(HoodieTableConfig.DATABASE_NAME.key()))
+ assertFalse(table.properties.contains(HoodieTableConfig.NAME.key()))
+ assertFalse(table.properties.contains(OPERATION.key()))
- val tablePath = table.storage.properties("path")
- val metaClient = createMetaClient(spark, tablePath)
- val tableConfig = metaClient.getTableConfig.getProps.asScala.toMap
-
assertResult(true)(tableConfig.contains(HoodieTableConfig.CREATE_SCHEMA.key))
- assertResult("dt")(tableConfig(HoodieTableConfig.PARTITION_FIELDS.key))
- assertResult("id")(tableConfig(HoodieTableConfig.RECORDKEY_FIELDS.key))
- assertResult("ts")(tableConfig(HoodieTableConfig.ORDERING_FIELDS.key))
-
assertResult(KeyGeneratorType.SIMPLE.name())(tableConfig(HoodieTableConfig.KEY_GENERATOR_TYPE.key))
- assertResult("default")(tableConfig(HoodieTableConfig.DATABASE_NAME.key()))
- assertResult(tableName)(tableConfig(HoodieTableConfig.NAME.key()))
- assertFalse(tableConfig.contains(OPERATION.key()))
+ val tablePath = table.storage.properties("path")
+ val metaClient = createMetaClient(spark, tablePath)
+
assertTrue(metaClient.getTableConfig.contains(HoodieTableConfig.CREATE_SCHEMA))
+ val expectedTableVersion = tableVersion.getOrElse(
+
HoodieTableVersion.fromVersionCode(HoodieWriteConfig.WRITE_TABLE_VERSION.defaultValue()))
+ assertTableConfigs(metaClient.getStorage, tablePath,
expectedTableConfigs(tableName, expectedTableVersion).asJava)
+ }
+ }
+
+ private def expectedTableConfigs(tableName: String, tableVersion:
HoodieTableVersion): Map[String, String] = {
+ val commonConfigs = Map(
+ "hoodie.database.name" -> "default",
+ "hoodie.table.name" -> tableName,
+ "hoodie.table.type" -> "COPY_ON_WRITE",
+ "hoodie.table.version" -> tableVersion.versionCode().toString,
+ "hoodie.table.recordkey.fields" -> "id",
+ "hoodie.table.partition.fields" -> "dt",
+ "hoodie.table.ordering.fields" -> "ts",
+ "hoodie.table.keygenerator.type" -> "SIMPLE",
+ "hoodie.table.format" -> "native",
+ "hoodie.meta.fields.mode" -> "ALL",
+ "hoodie.datasource.write.drop.partition.columns" -> "false",
+ "hoodie.datasource.write.hive_style_partitioning" -> "true",
+ "hoodie.datasource.write.partitionpath.urlencode" -> "false",
+ "hoodie.archivelog.folder" -> "archived",
+ "hoodie.timeline.path" -> "timeline",
+ "hoodie.timeline.history.path" -> "history")
+ val layoutConfigs = if (tableVersion.lesserThan(HoodieTableVersion.EIGHT))
{
+ Map(
+ "hoodie.timeline.layout.version" -> "1",
+ "hoodie.table.keygenerator.class" ->
"org.apache.hudi.keygen.SimpleKeyGenerator")
+ } else {
+ Map(
+ "hoodie.timeline.layout.version" -> "2",
+ "hoodie.table.initial.version" -> tableVersion.versionCode().toString,
+ "hoodie.record.merge.mode" -> "EVENT_TIME_ORDERING")
+ }
+ val mergeConfigs = if (tableVersion.lesserThan(HoodieTableVersion.NINE)) {
+ Map(
+ "hoodie.compaction.payload.class" ->
"org.apache.hudi.common.model.DefaultHoodieRecordPayload",
+ "hoodie.record.merge.strategy.id" ->
"eeb8d96f-b1e4-49fd-bbf8-28ac514178e5")
+ } else {
+ Map.empty[String, String]
+ }
+ val metaFieldsConfigs = if
(tableVersion.lesserThan(HoodieTableVersion.TEN)) {
+ Map("hoodie.populate.meta.fields" -> "true")
+ } else {
+ Map.empty[String, String]
+ }
+ commonConfigs ++ layoutConfigs ++ mergeConfigs ++ metaFieldsConfigs
}
test("Test Create External Hoodie Table") {
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
index 8d24aa3e3156..62f9d674e051 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
@@ -171,6 +171,7 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.NullSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
@@ -205,6 +206,7 @@ import static
org.apache.hudi.common.table.checkpoint.StreamerCheckpointV1.STREA
import static
org.apache.hudi.common.table.checkpoint.StreamerCheckpointV2.STREAMER_CHECKPOINT_KEY_V2;
import static
org.apache.hudi.common.table.timeline.InstantComparison.GREATER_THAN;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME_GENERATOR;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.assertTableConfigs;
import static org.apache.hudi.common.util.StringUtils.EMPTY_STRING;
import static
org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TABLE_PERSIST_SOURCE_RDD;
import static org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient;
@@ -617,22 +619,80 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
deltaStreamer.shutdownGracefully();
}
+ /**
+ * A null version leaves the write table version unset, so the default
version is covered as well.
+ */
@ParameterizedTest
- @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "EIGHT"})
- public void testPartitionKeyFieldsBasedOnVersion(HoodieTableVersion version)
throws IOException {
- String tablePath = basePath + "/partition_key_fields_meta_client" +
version.versionCode();
+ @NullSource
+ @EnumSource(value = HoodieTableVersion.class, names = {"SIX", "EIGHT",
"NINE", "TEN"})
+ void testTableConfigsBasedOnVersion(HoodieTableVersion version) throws
IOException {
+ HoodieTableVersion expectedVersion = version == null
+ ?
HoodieTableVersion.fromVersionCode(HoodieWriteConfig.WRITE_TABLE_VERSION.defaultValue())
: version;
+ String tablePath = basePath + "/table_configs_v" + (version == null ?
"default" : String.valueOf(version.versionCode()));
HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
- cfg.configs.add(HoodieWriteConfig.WRITE_TABLE_VERSION.key() + "=" +
version.versionCode());
+ if (version != null) {
+ cfg.configs.add(HoodieWriteConfig.WRITE_TABLE_VERSION.key() + "=" +
version.versionCode());
+ }
cfg.configs.add(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key() + "=" +
CustomKeyGenerator.class.getName());
cfg.configs.add("hoodie.datasource.write.partitionpath.field=partition_path:simple");
HoodieDeltaStreamer deltaStreamer = new HoodieDeltaStreamer(cfg, jsc);
deltaStreamer.getIngestionService().ingestOnce();
- HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(context, tablePath);
- String expectedPartitionFields = version.equals(HoodieTableVersion.SIX) ?
"partition_path" : "partition_path:simple";
- assertEquals(expectedPartitionFields,
metaClient.getTableConfig().getString(HoodieTableConfig.PARTITION_FIELDS));
+ assertTableConfigs(storage, tablePath,
expectedTableConfigs(expectedVersion));
+ deltaStreamer.shutdownGracefully();
+
+ // the next write backfills the key generator class of a table created
without it
+ HoodieTableConfig.delete(storage, new StoragePath(tablePath,
HoodieTableMetaClient.METAFOLDER_NAME),
+
Collections.singleton(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key()));
+ deltaStreamer = new HoodieDeltaStreamer(cfg, jsc);
+ deltaStreamer.getIngestionService().ingestOnce();
+ assertTableConfigs(storage, tablePath,
expectedTableConfigs(expectedVersion));
deltaStreamer.shutdownGracefully();
}
+ private static Map<String, String> expectedTableConfigs(HoodieTableVersion
version) {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("hoodie.table.name", "hoodie_trips");
+ configs.put("hoodie.table.type", "COPY_ON_WRITE");
+ configs.put("hoodie.table.version", String.valueOf(version.versionCode()));
+ configs.put("hoodie.table.recordkey.fields", "_row_key");
+ configs.put("hoodie.table.ordering.fields", "timestamp");
+ configs.put("hoodie.table.keygenerator.type", "CUSTOM");
+ configs.put("hoodie.table.base.file.format", "PARQUET");
+ configs.put("hoodie.table.format", "native");
+ configs.put("hoodie.table.cdc.enabled", "false");
+ configs.put("hoodie.table.metadata.partitions.inflight", "");
+ configs.put("hoodie.table.index.defs.path",
".hoodie/.index_defs/index.json");
+ configs.put("hoodie.table.partition_extractor_class",
"org.apache.hudi.hive.MultiPartKeysValueExtractor");
+ configs.put("hoodie.meta.fields.mode", "ALL");
+ configs.put("hoodie.partition.metafile.use.base.format", "false");
+ configs.put("hoodie.datasource.write.drop.partition.columns", "false");
+ configs.put("hoodie.datasource.write.hive_style_partitioning", "false");
+ configs.put("hoodie.datasource.write.partitionpath.urlencode", "false");
+ configs.put("hoodie.archivelog.folder", "history");
+ configs.put("hoodie.timeline.path", "timeline");
+ configs.put("hoodie.timeline.history.path", "history");
+ if (version.lesserThan(HoodieTableVersion.EIGHT)) {
+ configs.put("hoodie.table.partition.fields", "partition_path");
+ configs.put("hoodie.table.metadata.partitions", "column_stats,files");
+ configs.put("hoodie.timeline.layout.version", "1");
+ configs.put("hoodie.table.keygenerator.class",
"org.apache.hudi.keygen.CustomKeyGenerator");
+ } else {
+ configs.put("hoodie.table.partition.fields", "partition_path:simple");
+ configs.put("hoodie.table.metadata.partitions",
"column_stats,files,partition_stats");
+ configs.put("hoodie.timeline.layout.version", "2");
+ configs.put("hoodie.table.initial.version",
String.valueOf(version.versionCode()));
+ configs.put("hoodie.record.merge.mode", "COMMIT_TIME_ORDERING");
+ }
+ if (version.lesserThan(HoodieTableVersion.NINE)) {
+ configs.put("hoodie.compaction.payload.class",
"org.apache.hudi.common.model.OverwriteWithLatestAvroPayload");
+ configs.put("hoodie.record.merge.strategy.id",
"ce9acb64-bde0-424c-9b91-f6ebba25356d");
+ }
+ if (version.lesserThan(HoodieTableVersion.TEN)) {
+ configs.put("hoodie.populate.meta.fields", "true");
+ }
+ return configs;
+ }
+
@ParameterizedTest
@EnumSource(value = HoodieRecordType.class, names = {"AVRO", "SPARK"})
public void testBulkInsertsAndUpsertsWithBootstrap(HoodieRecordType
recordType) throws Exception {
