codope commented on code in PR #11077:
URL: https://github.com/apache/hudi/pull/11077#discussion_r1579848734
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java:
##########
@@ -243,34 +250,57 @@ public static HoodieMergedLogRecordScanner.Builder
newBuilder() {
@Override
public <T> void processNextRecord(HoodieRecord<T> newRecord) throws
IOException {
String key = newRecord.getRecordKey();
- HoodieRecord<T> prevRecord = records.get(key);
+ HoodieMergeKey mergeKey = newRecord.getMergeKey();
+ if (mergeKey == null) {
+ // If mergeKey is null, then create a simple merge key using record key
+ mergeKey = new HoodieSimpleMergeKey(newRecord.getKey());
+ }
+ HoodieRecord<T> prevRecord = records.get(mergeKey);
if (prevRecord != null) {
// Merge and store the combined record
- HoodieRecord<T> combinedRecord = (HoodieRecord<T>)
recordMerger.merge(prevRecord, readerSchema,
- newRecord, readerSchema, this.getPayloadProps()).get().getLeft();
- // If pre-combine returns existing record, no need to update it
- if (combinedRecord.getData() != prevRecord.getData()) {
- HoodieRecord latestHoodieRecord =
- combinedRecord.newInstance(new HoodieKey(key,
newRecord.getPartitionPath()), newRecord.getOperation());
-
- latestHoodieRecord.unseal();
- latestHoodieRecord.setCurrentLocation(newRecord.getCurrentLocation());
- latestHoodieRecord.seal();
-
- // NOTE: Record have to be cloned here to make sure if it holds
low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer
we get a clean copy of
- // it since these records will be put into records(Map).
- records.put(key, latestHoodieRecord.copy());
+ try {
+ HoodieMergeKey finalMergeKey = mergeKey;
+ recordMerger.fullOuterMerge(prevRecord, readerSchema, newRecord,
readerSchema, this.getPayloadProps()).forEach(
+ mergedRecord -> {
+ HoodieRecord<T> combinedRecord = mergedRecord.getLeft();
+ if (combinedRecord.getData() != prevRecord.getData()) {
+ HoodieRecord latestHoodieRecord =
getLatestHoodieRecord(newRecord, combinedRecord, key);
+ records.put(finalMergeKey, latestHoodieRecord.copy());
+ }
+ });
+ } catch (UnsupportedOperationException e) {
+ LOG.warn("Error merging records: Full outer merge not supported,
falling back to default merge for key {}", key);
+ HoodieRecord<T> combinedRecord = (HoodieRecord<T>)
recordMerger.merge(prevRecord, readerSchema,
+ newRecord, readerSchema, this.getPayloadProps()).get().getLeft();
+ // If pre-combine returns existing record, no need to update it
+ if (combinedRecord.getData() != prevRecord.getData()) {
Review Comment:
Basically, want to fallback to original method if a merger is used that does
not support MergeKey based merging. Wanted to avoid another boolean flag in the
scanner so as not to expose the implementation detail here. Let me think if
there can be a better way to do this.
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodieMergeKey.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.common.model;
+
+import java.io.Serializable;
+
+/**
+ * Defines a standard for all merge keys to ensure consistent handling
including simple keys and composite keys.
+ * It includes methods for retrieving the key and partition path.
+ */
+public interface HoodieMergeKey extends Serializable {
Review Comment:
Yes, that's right.
--
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]
