nsivabalan commented on code in PR #19110:
URL: https://github.com/apache/hudi/pull/19110#discussion_r3611475609
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -923,24 +922,41 @@ private static void handlePartialUpdateModeConfigs(String
payloadClassName, Map<
}
}
- private static void handlePayloadAdhocConfigs(String payloadClassName,
Map<String, String> reconciledConfigs) {
+ private static void handlePayloadAdhocConfigs(String payloadClassName,
Map<String, String> reconciledConfigs,
+ String orderingFieldName) {
// Certain payloads are migrated to non payload way from 1.1 Hudi binary
and the reader might need certain properties for the
// merge to function as expected. Handing such special cases here.
if (payloadClassName.equals(PostgresDebeziumAvroPayload.class.getName())) {
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX +
PARTIAL_UPDATE_UNAVAILABLE_VALUE, DEBEZIUM_UNAVAILABLE_VALUE);
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY,
DebeziumConstants.FLATTENED_OP_COL_NAME);
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER,
DebeziumConstants.DELETE_OP);
- reconciledConfigs.put(ORDERING_FIELDS.key(),
DebeziumConstants.FLATTENED_LSN_COL_NAME);
+ reconciledConfigs.put(ORDERING_FIELDS.key(),
reconcileDebeziumOrderingFields(payloadClassName, orderingFieldName));
Review Comment:
I suggesting to move this to the call site.
why do we reconcile w/n this method?
or do we have too many call sites?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -923,24 +922,41 @@ private static void handlePartialUpdateModeConfigs(String
payloadClassName, Map<
}
}
- private static void handlePayloadAdhocConfigs(String payloadClassName,
Map<String, String> reconciledConfigs) {
+ private static void handlePayloadAdhocConfigs(String payloadClassName,
Map<String, String> reconciledConfigs,
+ String orderingFieldName) {
// Certain payloads are migrated to non payload way from 1.1 Hudi binary
and the reader might need certain properties for the
// merge to function as expected. Handing such special cases here.
if (payloadClassName.equals(PostgresDebeziumAvroPayload.class.getName())) {
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX +
PARTIAL_UPDATE_UNAVAILABLE_VALUE, DEBEZIUM_UNAVAILABLE_VALUE);
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY,
DebeziumConstants.FLATTENED_OP_COL_NAME);
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER,
DebeziumConstants.DELETE_OP);
- reconciledConfigs.put(ORDERING_FIELDS.key(),
DebeziumConstants.FLATTENED_LSN_COL_NAME);
+ reconciledConfigs.put(ORDERING_FIELDS.key(),
reconcileDebeziumOrderingFields(payloadClassName, orderingFieldName));
} else if
(payloadClassName.equals(MySqlDebeziumAvroPayload.class.getName())) {
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY,
DebeziumConstants.FLATTENED_OP_COL_NAME);
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER,
DebeziumConstants.DELETE_OP);
- reconciledConfigs.put(ORDERING_FIELDS.key(),
DebeziumConstants.FLATTENED_FILE_COL_NAME + "," +
DebeziumConstants.FLATTENED_POS_COL_NAME);
+ reconciledConfigs.put(ORDERING_FIELDS.key(),
reconcileDebeziumOrderingFields(payloadClassName, orderingFieldName));
} else if (payloadClassName.equals(AWSDmsAvroPayload.class.getName())) {
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY,
OP_FIELD);
reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER,
DELETE_OPERATION_VALUE);
}
}
+ /**
+ * Forces a Debezium payload's ordering field to the canonical column(s) the
payload merges on.
+ * The flat form ({@code _event_lsn}, or {@code _event_bin_file,_event_pos})
is used unless the
+ * caller already resolved the nested form — which only the Hudi Streamer
does when its transformer
+ * groups the CDC metadata under the {@link
DebeziumConstants#DEBEZIUM_METADATA_FIELD} struct — in
+ * which case the nested form is preserved so the payload orders against the
nested columns.
+ * Keeping this here (rather than trusting {@code orderingFieldName}
verbatim) preserves the
+ * long-standing auto-correction for every create path (Spark, Flink,
Streamer).
+ */
+ private static String reconcileDebeziumOrderingFields(String
payloadClassName, String orderingFieldName) {
+ String nestedOrderingFields =
DebeziumConstants.resolveOrderingFields(payloadClassName, true);
Review Comment:
not sure I understand this change.
I responded else where.
we should move this to call sites, were we have write configs too.
and we know if nested meta field is enabled or not.
and hence we should be able to detect the right ordering field name as well
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
