You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "lokeshj1703 (via GitHub)" <gi...@apache.org> on 2025/08/11 16:25:27 UTC
Re: [PR] [HUDI-9638] infer merging table configs during table creation for version 9 [hudi]
lokeshj1703 commented on code in PR #13615:
URL: https://github.com/apache/hudi/pull/13615#discussion_r2265949724
##########
hudi-common/src/main/java/org/apache/hudi/common/model/AWSDmsAvroPayload.java:
##########
@@ -45,6 +45,7 @@
public class AWSDmsAvroPayload extends OverwriteWithLatestAvroPayload {
public static final String OP_FIELD = "Op";
+ public static final String D_VALUE = "D";
Review Comment:
NIT: Rename to DELETE_OPERATION_VALUE
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestFlinkWriteClients.java:
##########
@@ -128,6 +128,7 @@ void testRecordMergeConfigForCommitTimeOrdering(boolean useLegacyConfig) throws
assertThat(tableConfig.getRecordMergeStrategyId(), is(HoodieRecordMerger.COMMIT_TIME_BASED_MERGE_STRATEGY_UUID));
assertThat(tableConfig.getPayloadClass(), is(OverwriteWithLatestAvroPayload.class.getName()));
+
Review Comment:
unintended change
##########
hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestHoodieWriterUtils.java:
##########
@@ -43,4 +48,67 @@ void validateTableConfig() throws IOException {
properties.put(HoodieTableConfig.DATABASE_NAME.key(), "databaseFromCatalog");
Assertions.assertDoesNotThrow(() -> HoodieWriterUtils.validateTableConfig(sparkSession, JavaScalaConverters.convertJavaPropertiesToScalaMap(properties), tableConfig));
}
+
+ @Test
Review Comment:
Can we club these testcases in a single test or fewer tests?
##########
hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java:
##########
@@ -510,4 +522,166 @@ void testInferMergeMode(RecordMergeMode inputMergeMode, String inputPayloadClass
}
});
}
+
+ private static Stream<Arguments> argumentsForInferMergingConfigsForVersion9() {
+ return Stream.of(
+ // Test case: Non-version 9 table should return empty configs
+ arguments(
+ "Non-version 9 table", EVENT_TIME_ORDERING, null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, "ts", HoodieTableVersion.EIGHT,
+ 0, null, null, null, null, null, null, null, null, null),
+
+ // Test case: Version 9 table with null payload class and event time ordering
+ arguments("Version 9 with event time ordering", EVENT_TIME_ORDERING, null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, "ts", HoodieTableVersion.NINE,
+ 2, EVENT_TIME_ORDERING.name(), null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, null, null, null, null, null),
+
+ // Test case: Version 9 table with null payload class and commit time ordering
+ arguments("Version 9 with commit time ordering", COMMIT_TIME_ORDERING, null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, null, HoodieTableVersion.NINE,
+ 2, COMMIT_TIME_ORDERING.name(), null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, null, null, null, null, null),
+
+ // Test case: Version 9 table with null payload class and custom merge mode
+ arguments("Version 9 with custom merge mode", CUSTOM, null, CUSTOM_MERGE_STRATEGY_UUID, null, HoodieTableVersion.NINE,
+ 2, CUSTOM.name(), null, CUSTOM_MERGE_STRATEGY_UUID, null, null, null, null, null),
+
+ // Test case: Version 9 table with custom payload class (not under deprecation)
+ arguments("Version 9 with custom payload", null, "com.example.CustomPayload", null, null, HoodieTableVersion.NINE,
+ 3, CUSTOM.name(), "com.example.CustomPayload", PAYLOAD_BASED_MERGE_STRATEGY_UUID, null, null, null, null, null),
+
+ // Test case: Version 9 table with event time based payload (DefaultHoodieRecordPayload)
+ arguments("Version 9 with DefaultHoodieRecordPayload", null, DefaultHoodieRecordPayload.class.getName(), null, "ts", HoodieTableVersion.NINE,
+ 3, EVENT_TIME_ORDERING.name(), null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, DefaultHoodieRecordPayload.class.getName(), null, null, null, null),
+
+ // Test case: Version 9 table with commit time based payload (OverwriteWithLatestAvroPayload)
+ arguments("Version 9 with OverwriteWithLatestAvroPayload", null, OverwriteWithLatestAvroPayload.class.getName(), null, null, HoodieTableVersion.NINE,
+ 3, COMMIT_TIME_ORDERING.name(), null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, OverwriteWithLatestAvroPayload.class.getName(), null, null, null, null),
+
+ // Test case: Version 9 table with PartialUpdateAvroPayload (should set partial update mode)
+ arguments("Version 9 with PartialUpdateAvroPayload", null, PartialUpdateAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE,
+ 4, EVENT_TIME_ORDERING.name(), null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, PartialUpdateAvroPayload.class.getName(), PartialUpdateMode.IGNORE_DEFAULTS.name(),
+ null, null, null),
+
+ // Test case: Version 9 table with OverwriteNonDefaultsWithLatestAvroPayload (should set partial update mode)
+ arguments("Version 9 with OverwriteNonDefaultsWithLatestAvroPayload", null, OverwriteNonDefaultsWithLatestAvroPayload.class.getName(), null, null, HoodieTableVersion.NINE,
+ 4, COMMIT_TIME_ORDERING.name(), null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, OverwriteNonDefaultsWithLatestAvroPayload.class.getName(),
+ PartialUpdateMode.IGNORE_DEFAULTS.name(), null, null, null),
+
+ // Test case: Version 9 table with PostgresDebeziumAvroPayload (should set partial update mode and custom properties)
+ arguments("Version 9 with PostgresDebeziumAvroPayload", null, PostgresDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE,
+ 5, EVENT_TIME_ORDERING.name(), null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, PostgresDebeziumAvroPayload.class.getName(),
+ PartialUpdateMode.IGNORE_MARKERS.name(), HoodieTableConfig.DEBEZIUM_UNAVAILABLE_VALUE, null, null),
+
+ // Test case: Version 9 table with AWSDmsAvroPayload (should set custom delete properties)
+ arguments("Version 9 with AWSDmsAvroPayload", null, AWSDmsAvroPayload.class.getName(), null, null, HoodieTableVersion.NINE,
+ 5, COMMIT_TIME_ORDERING.name(), null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, AWSDmsAvroPayload.class.getName(), null, null, "Op", "D"),
+
+ // Test case: Version 9 table with EventTimeAvroPayload (event time based payload)
+ arguments("Version 9 with EventTimeAvroPayload", null, EventTimeAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE,
+ 3, EVENT_TIME_ORDERING.name(), null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, EventTimeAvroPayload.class.getName(), null, null, null, null),
+
+ // Test case: Version 9 table with MySqlDebeziumAvroPayload (event time based payload)
+ arguments("Version 9 with MySqlDebeziumAvroPayload", null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE,
+ 3, EVENT_TIME_ORDERING.name(), null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, MySqlDebeziumAvroPayload.class.getName(), null, null, null, null)
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("argumentsForInferMergingConfigsForVersion9")
+ void testInferMergingConfigsForVersion9(String testName, RecordMergeMode recordMergeMode, String payloadClassName,
+ String recordMergeStrategyId, String orderingFieldName, HoodieTableVersion tableVersion,
+ int expectedConfigSize, String expectedMergeMode, String expectedPayloadClass,
+ String expectedMergeStrategyId, String expectedLegacyPayloadClass,
+ String expectedPartialUpdateMode, String expectedDebeziumMarker,
+ String expectedDeleteKey, String expectedDeleteMarker) {
+ Map<String, String> configs = HoodieTableConfig.inferMergingConfigsForVersion9(
+ recordMergeMode, payloadClassName, recordMergeStrategyId, orderingFieldName, tableVersion);
+
+ assertEquals(expectedConfigSize, configs.size(), "Config size mismatch for: " + testName);
+ if (expectedMergeMode != null) {
+ assertEquals(expectedMergeMode, configs.get(RECORD_MERGE_MODE.key()),
+ "Merge mode mismatch for: " + testName);
+ }
+ if (expectedPayloadClass != null) {
+ assertEquals(expectedPayloadClass, configs.get(PAYLOAD_CLASS_NAME.key()),
+ "Payload class mismatch for: " + testName);
+ }
+ if (expectedMergeStrategyId != null) {
+ assertEquals(expectedMergeStrategyId, configs.get(RECORD_MERGE_STRATEGY_ID.key()),
+ "Merge strategy ID mismatch for: " + testName);
+ }
+ if (expectedLegacyPayloadClass != null) {
+ assertEquals(expectedLegacyPayloadClass, configs.get(LEGACY_PAYLOAD_CLASS_NAME.key()),
+ "Legacy payload class mismatch for: " + testName);
+ }
+ if (expectedPartialUpdateMode != null) {
+ assertEquals(expectedPartialUpdateMode, configs.get(HoodieTableConfig.PARTIAL_UPDATE_MODE.key()),
+ "Partial update mode mismatch for: " + testName);
+ }
+ if (expectedDebeziumMarker != null) {
+ assertEquals(expectedDebeziumMarker, configs.get(
+ HoodieTableConfig.MERGE_CUSTOM_PROPERTY_PREFIX + HoodieTableConfig.PARTIAL_UPDATE_CUSTOM_MARKER),
+ "Debezium marker mismatch for: " + testName);
+ }
+ if (expectedDeleteKey != null) {
+ assertEquals(expectedDeleteKey, configs.get(HoodieTableConfig.MERGE_CUSTOM_PROPERTY_PREFIX + DELETE_KEY),
+ "Delete key mismatch for: " + testName);
+ }
+ if (expectedDeleteMarker != null) {
+ assertEquals(expectedDeleteMarker, configs.get(HoodieTableConfig.MERGE_CUSTOM_PROPERTY_PREFIX + DELETE_MARKER),
+ "Delete marker mismatch for: " + testName);
+ }
+ }
+
+ @Test
+ void testInferMergingConfigsForVersion9WithInconsistentConfigs() {
+ // Test case: Inconsistent merge mode and strategy should throw exception
+ assertThrows(IllegalArgumentException.class, () -> {
+ HoodieTableConfig.inferMergingConfigsForVersion9(
+ EVENT_TIME_ORDERING, null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, "ts", HoodieTableVersion.NINE);
+ });
+ assertThrows(IllegalArgumentException.class, () -> {
+ HoodieTableConfig.inferMergingConfigsForVersion9(
+ COMMIT_TIME_ORDERING, null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, null, HoodieTableVersion.NINE);
+ });
+ assertThrows(IllegalArgumentException.class, () -> {
+ HoodieTableConfig.inferMergingConfigsForVersion9(
+ CUSTOM, null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, null, HoodieTableVersion.NINE);
+ });
+ }
+
+ @Test
+ void testInferMergingConfigsForVersion9EdgeCases() {
+ // Test case: Empty string payload class should be treated as null
+ Map<String, String> configs = HoodieTableConfig.inferMergingConfigsForVersion9(
+ EVENT_TIME_ORDERING, "", EVENT_TIME_BASED_MERGE_STRATEGY_UUID, "ts", HoodieTableVersion.NINE);
+ assertEquals(2, configs.size());
+ assertEquals(EVENT_TIME_ORDERING.name(), configs.get(RECORD_MERGE_MODE.key()));
+ assertEquals(EVENT_TIME_BASED_MERGE_STRATEGY_UUID, configs.get(RECORD_MERGE_STRATEGY_ID.key()));
+
+ // Test case: Non-version 9 table with all parameters should return empty configs
+ configs = HoodieTableConfig.inferMergingConfigsForVersion9(
+ EVENT_TIME_ORDERING, DefaultHoodieRecordPayload.class.getName(),
+ EVENT_TIME_BASED_MERGE_STRATEGY_UUID, "ts", HoodieTableVersion.EIGHT);
+ assertEquals(0, configs.size());
+
+ // Test case: Version 9 table with null ordering field for event time ordering should still work
+ configs = HoodieTableConfig.inferMergingConfigsForVersion9(
+ EVENT_TIME_ORDERING, null, EVENT_TIME_BASED_MERGE_STRATEGY_UUID, null, HoodieTableVersion.NINE);
+ assertEquals(2, configs.size());
+ assertEquals(EVENT_TIME_ORDERING.name(), configs.get(RECORD_MERGE_MODE.key()));
+ assertEquals(EVENT_TIME_BASED_MERGE_STRATEGY_UUID, configs.get(RECORD_MERGE_STRATEGY_ID.key()));
+ }
+
+ @Test
+ void testInferMergingConfigsForVersion9WithAllTableVersions() {
+ // Test that only version 9 returns configs, others return empty
+ for (HoodieTableVersion version : HoodieTableVersion.values()) {
+ Map<String, String> configs = HoodieTableConfig.inferMergingConfigsForVersion9(
+ EVENT_TIME_ORDERING, DefaultHoodieRecordPayload.class.getName(),
+ EVENT_TIME_BASED_MERGE_STRATEGY_UUID, "ts", version);
+ if (version == HoodieTableVersion.NINE) {
+ assertTrue(configs.size() > 0, "Version 9 should return configs");
+ } else {
+ assertEquals(0, configs.size(), "Non-version 9 should return empty configs");
Review Comment:
Should we throw an error in non version 9 for this API?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org