This is an automated email from the ASF dual-hosted git repository.
jonvex 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 5f35be01606 Fixing deltastreamer tests for auto record key gen (#11099)
5f35be01606 is described below
commit 5f35be016063123c653e5075b84a9436a58e4951
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Mon May 6 08:00:11 2024 -0700
Fixing deltastreamer tests for auto record key gen (#11099)
---
.../deltastreamer/HoodieDeltaStreamerTestBase.java | 19 ++++++++++++++-----
.../deltastreamer/TestHoodieDeltaStreamer.java | 7 +++++--
.../TestHoodieDeltaStreamerSchemaEvolutionBase.java | 2 +-
3 files changed, 20 insertions(+), 8 deletions(-)
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
index 3e32baf1aa2..e0093f3c92a 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
@@ -345,7 +345,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
protected void prepareParquetDFSSource(boolean useSchemaProvider, boolean
hasTransformer, String emptyBatchParam) throws IOException {
prepareParquetDFSSource(useSchemaProvider, hasTransformer, "source.avsc",
"target.avsc",
- PROPS_FILENAME_TEST_PARQUET, PARQUET_SOURCE_ROOT, false,
"partition_path", emptyBatchParam);
+ PROPS_FILENAME_TEST_PARQUET, PARQUET_SOURCE_ROOT, false,
"partition_path", emptyBatchParam, false);
}
protected void prepareParquetDFSSource(boolean useSchemaProvider, boolean
hasTransformer) throws IOException {
@@ -355,20 +355,27 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
protected void prepareParquetDFSSource(boolean useSchemaProvider, boolean
hasTransformer, String sourceSchemaFile, String targetSchemaFile,
String propsFileName, String
parquetSourceRoot, boolean addCommonProps, String partitionPath) throws
IOException {
prepareParquetDFSSource(useSchemaProvider, hasTransformer,
sourceSchemaFile, targetSchemaFile, propsFileName, parquetSourceRoot,
addCommonProps,
- partitionPath, "");
+ partitionPath, "", false);
}
protected void prepareParquetDFSSource(boolean useSchemaProvider, boolean
hasTransformer, String sourceSchemaFile, String targetSchemaFile,
String propsFileName, String
parquetSourceRoot, boolean addCommonProps,
String partitionPath, String
emptyBatchParam) throws IOException {
prepareParquetDFSSource(useSchemaProvider, hasTransformer,
sourceSchemaFile, targetSchemaFile, propsFileName, parquetSourceRoot,
addCommonProps,
- partitionPath, emptyBatchParam, null);
+ partitionPath, emptyBatchParam, false);
+ }
+ protected void prepareParquetDFSSource(boolean useSchemaProvider, boolean
hasTransformer, String sourceSchemaFile, String targetSchemaFile,
+ String propsFileName, String
parquetSourceRoot, boolean addCommonProps,
+ String partitionPath, String
emptyBatchParam, boolean skipRecordKeyField) throws IOException {
+ prepareParquetDFSSource(useSchemaProvider, hasTransformer,
sourceSchemaFile, targetSchemaFile, propsFileName, parquetSourceRoot,
addCommonProps,
+ partitionPath, emptyBatchParam, null, skipRecordKeyField);
}
protected void prepareParquetDFSSource(boolean useSchemaProvider, boolean
hasTransformer, String sourceSchemaFile, String targetSchemaFile,
String propsFileName, String
parquetSourceRoot, boolean addCommonProps,
- String partitionPath, String
emptyBatchParam, TypedProperties extraProps) throws IOException {
+ String partitionPath, String
emptyBatchParam, TypedProperties extraProps,
+ boolean skipRecordKeyField) throws
IOException {
// Properties used for testing delta-streamer with Parquet source
TypedProperties parquetProps = new TypedProperties(extraProps);
@@ -380,7 +387,9 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
parquetProps.setProperty("include", "base.properties");
parquetProps.setProperty("hoodie.embed.timeline.server", "false");
- parquetProps.setProperty("hoodie.datasource.write.recordkey.field",
"_row_key");
+ if (!skipRecordKeyField) {
+ parquetProps.setProperty("hoodie.datasource.write.recordkey.field",
"_row_key");
+ }
parquetProps.setProperty("hoodie.datasource.write.partitionpath.field",
partitionPath);
if (useSchemaProvider) {
parquetProps.setProperty("hoodie.streamer.schemaprovider.source.schema.file",
basePath + "/" + sourceSchemaFile);
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 8c9cbc94923..e8b72a9c924 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
@@ -832,7 +832,7 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
extraProps.setProperty("hoodie.datasource.write.table.type",
"MERGE_ON_READ");
extraProps.setProperty("hoodie.datasource.compaction.async.enable",
"false");
prepareParquetDFSSource(false, false, "source.avsc", "target.avsc",
PROPS_FILENAME_TEST_PARQUET,
- PARQUET_SOURCE_ROOT, false, "partition_path", "", extraProps);
+ PARQUET_SOURCE_ROOT, false, "partition_path", "", extraProps, false);
String tableBasePath = basePath + "test_parquet_table" + testNum;
HoodieDeltaStreamer.Config deltaCfg =
TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
ParquetDFSSource.class.getName(),
null, PROPS_FILENAME_TEST_PARQUET, false,
@@ -2817,7 +2817,7 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
boolean hasTransformer = transformerClassNames != null &&
!transformerClassNames.isEmpty();
prepareParquetDFSFiles(parquetRecordsCount, PARQUET_SOURCE_ROOT,
FIRST_PARQUET_FILE_NAME, false, null, null);
prepareParquetDFSSource(useSchemaProvider, hasTransformer, "source.avsc",
"target.avsc", PROPS_FILENAME_TEST_PARQUET,
- PARQUET_SOURCE_ROOT, false, "partition_path", "");
+ PARQUET_SOURCE_ROOT, false, "partition_path", "", true);
String tableBasePath = basePath + "/test_parquet_table" + testNum;
HoodieDeltaStreamer.Config config = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT, ParquetDFSSource.class.getName(),
@@ -2826,6 +2826,9 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
HoodieDeltaStreamer deltaStreamer = new HoodieDeltaStreamer(config, jsc);
deltaStreamer.sync();
assertRecordCount(parquetRecordsCount, tableBasePath, sqlContext);
+ // validate that auto record keys are enabled.
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.builder().setBasePath(tableBasePath).setConf(jsc.hadoopConfiguration()).build();
+ assertFalse(metaClient.getTableConfig().getRecordKeyFields().isPresent());
prepareParquetDFSFiles(200, PARQUET_SOURCE_ROOT, "2.parquet", false, null,
null);
deltaStreamer.sync();
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerSchemaEvolutionBase.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerSchemaEvolutionBase.java
index 0505ec3e2f3..c069280c8ca 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerSchemaEvolutionBase.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerSchemaEvolutionBase.java
@@ -206,7 +206,7 @@ public class TestHoodieDeltaStreamerSchemaEvolutionBase
extends HoodieDeltaStrea
transformerClassNames, PROPS_FILENAME_TEST_AVRO_KAFKA, false,
useSchemaProvider, 100000, false, null, tableType, "timestamp", null);
} else {
prepareParquetDFSSource(false, hasTransformer, sourceSchemaFile,
targetSchemaFile, PROPS_FILENAME_TEST_PARQUET,
- PARQUET_SOURCE_ROOT, false, "partition_path", "", extraProps);
+ PARQUET_SOURCE_ROOT, false, "partition_path", "", extraProps, false);
cfg = TestHoodieDeltaStreamer.TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT, ParquetDFSSource.class.getName(),
transformerClassNames, PROPS_FILENAME_TEST_PARQUET, false,
useSchemaProvider, 100000, false, null, tableType, "timestamp",
null);
