You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "nsivabalan (via GitHub)" <gi...@apache.org> on 2025/07/25 06:48:51 UTC
[PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
nsivabalan opened a new pull request, #13623:
URL: https://github.com/apache/hudi/pull/13623
### Change Logs
- Unifying Partial Update Modes from TableProperty and Merge Into Partial Update Encoding.
- Fixed all of partial update handling within BufferedRecordMergerFactory.
- Fixed Partial update encoding support across all record formats (prior to this patch only spark record was supported).
Detailed changes:
Possible values for table property PartialUpdateMode:
IGNORE_DEFAULTS, IGNORE_DEFAULTS_NULLS, FILL_UNAVAILABLE.
And for table's not having such partial update requirement, we may not find any value for `PartialUpdateMode` in table property.
In addition to this, we also have another enum value named "KEEP_VALUES" which will take effect when Partial columns are encoded from Merge Into.
Hence within BufferedRecordMergerFacotry.create(), we take in two arguments named `enablePartialEncoding` and `Option<PartialUpdateMode>`. `enablePartialEncoding` boolean flag will denote whether is full record merging or not.
If enabled, the value set for PartialUpdateMode should be honored.
### Impact
- Cleaning up PartialUpdateMode feature and unifying MergeInto partial update encoding along with other PartialUpdateModes from table config.
### Risk level (write none, low medium or high below)
medium
### Documentation Update
_Describe any necessary documentation update if there is any new feature, config, or user-facing change. If not, put "none"._
- _The config description must be updated if new configs are added or the default value of the configs are changed_
- _Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
ticket number here and follow the [instruction](https://hudi.apache.org/contribute/developer-setup#website) to make
changes to the website._
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230991975
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateStrategy.java:
##########
@@ -63,44 +68,82 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ Schema readerSchema,
+ boolean keepOldMetadataColumns,
+ TypedProperties props) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema, props);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
newRecord, oldRecord, newSchema, oldSchema);
default:
- return newRecord;
+ throw new HoodieIOException("Unsupported PartialUpdateMode " + partialUpdateMode + " detected");
}
}
+ BufferedRecord<T> reconcileBasedOnKeepValues(BufferedRecord<T> newRecord,
+ BufferedRecord<T> oldRecord,
+ Schema newSchema,
+ Schema oldSchema,
+ Schema readerSchema,
+ TypedProperties props) {
+
+ // Merge and store the combined record
+ Option<Pair<BufferedRecord, Schema>> deleteHandlingResult = handleDeletes(oldRecord, oldSchema, newRecord, newSchema, props);
+ if (deleteHandlingResult != null) {
+ return newRecord; // if deleted, return newRecord. newRecord.isDelete() will return anyways.
+ }
+
+ return (BufferedRecord<T>) keepValuesPartialMergingUtils.mergePartialRecords(oldRecord, oldSchema, newRecord, newSchema, readerSchema, readerContext).getLeft();
+ }
+
+ /**
+ * Basic handling of deletes that is used by many of the spark mergers
+ * returns null if merger specific logic should be used
+ */
+ protected Option<Pair<BufferedRecord, Schema>> handleDeletes(BufferedRecord older, Schema oldSchema, BufferedRecord newer, Schema newSchema, TypedProperties props) {
+
+ if (newer.isDelete()) {
+ // Delete record
+ return Option.empty();
+ }
+
+ // old record
+ if (older.isDelete()) {
+ return Option.of(Pair.of(newer, newSchema));
+ }
+ return null;
+ }
+
/**
* @param newRecord The newer record determined by the merge mode.
* @param oldRecord The older record determined by the merge mode.
* @param newSchema The schema of the newer record.
* @param oldSchema The schema of the older record.
* @param keepOldMetadataColumns Keep the metadata columns from the older record.
+ * @param includeNulls true if nulls are expected to be ignored as well. false otherwise.
* @return
*/
BufferedRecord<T> reconcileDefaultValues(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ boolean keepOldMetadataColumns,
+ boolean includeNulls) {
Review Comment:
Again I did not see any payload exclude NULL when comparing with default values.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230988732
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateStrategy.java:
##########
@@ -63,44 +68,82 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ Schema readerSchema,
+ boolean keepOldMetadataColumns,
+ TypedProperties props) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema, props);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
newRecord, oldRecord, newSchema, oldSchema);
default:
- return newRecord;
+ throw new HoodieIOException("Unsupported PartialUpdateMode " + partialUpdateMode + " detected");
}
}
+ BufferedRecord<T> reconcileBasedOnKeepValues(BufferedRecord<T> newRecord,
+ BufferedRecord<T> oldRecord,
+ Schema newSchema,
+ Schema oldSchema,
+ Schema readerSchema,
+ TypedProperties props) {
+
+ // Merge and store the combined record
+ Option<Pair<BufferedRecord, Schema>> deleteHandlingResult = handleDeletes(oldRecord, oldSchema, newRecord, newSchema, props);
+ if (deleteHandlingResult != null) {
+ return newRecord; // if deleted, return newRecord. newRecord.isDelete() will return anyways.
+ }
+
+ return (BufferedRecord<T>) keepValuesPartialMergingUtils.mergePartialRecords(oldRecord, oldSchema, newRecord, newSchema, readerSchema, readerContext).getLeft();
+ }
+
+ /**
+ * Basic handling of deletes that is used by many of the spark mergers
Review Comment:
Comment should be improved, this is not just for Spark.
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231423817
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieReaderContext.java:
##########
@@ -413,9 +413,18 @@ public Comparable getOrderingValue(T record,
* @param baseRecord The record based on which the engine record is built.
* @return A new instance of engine record type {@link T}.
*/
- public abstract T constructEngineRecord(Schema schema,
- Map<Integer, Object> updateValues,
- BufferedRecord<T> baseRecord);
+ public abstract T mergeEngineRecord(Schema schema,
Review Comment:
Can this be `update` instead of `merge`? It is confusing since we use merge elsewhere to mean taking two records, comparing them, and then outputting a record.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230994550
##########
hudi-flink-datasource/hudi-flink1.16.x/pom.xml:
##########
@@ -0,0 +1,162 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
Review Comment:
Why include these pom?
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3190745616
For 1.1, we may not go w/ entire patch.
Hence, raised https://github.com/apache/hudi/pull/13721 to address feedbacks from vinoth on original PR on upgrade downgrade fix.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230975789
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateStrategy.java:
##########
@@ -63,44 +68,82 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ Schema readerSchema,
+ boolean keepOldMetadataColumns,
+ TypedProperties props) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema, props);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
newRecord, oldRecord, newSchema, oldSchema);
default:
- return newRecord;
+ throw new HoodieIOException("Unsupported PartialUpdateMode " + partialUpdateMode + " detected");
}
}
+ BufferedRecord<T> reconcileBasedOnKeepValues(BufferedRecord<T> newRecord,
+ BufferedRecord<T> oldRecord,
+ Schema newSchema,
+ Schema oldSchema,
+ Schema readerSchema,
+ TypedProperties props) {
+
+ // Merge and store the combined record
+ Option<Pair<BufferedRecord, Schema>> deleteHandlingResult = handleDeletes(oldRecord, oldSchema, newRecord, newSchema, props);
+ if (deleteHandlingResult != null) {
+ return newRecord; // if deleted, return newRecord. newRecord.isDelete() will return anyways.
+ }
+
+ return (BufferedRecord<T>) keepValuesPartialMergingUtils.mergePartialRecords(oldRecord, oldSchema, newRecord, newSchema, readerSchema, readerContext).getLeft();
+ }
+
+ /**
+ * Basic handling of deletes that is used by many of the spark mergers
+ * returns null if merger specific logic should be used
+ */
+ protected Option<Pair<BufferedRecord, Schema>> handleDeletes(BufferedRecord older, Schema oldSchema, BufferedRecord newer, Schema newSchema, TypedProperties props) {
+
Review Comment:
extra new line.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231188992
##########
hudi-common/src/main/java/org/apache/hudi/common/table/PartialUpdateMode.java:
##########
@@ -22,28 +22,24 @@
import org.apache.hudi.common.config.EnumFieldDescription;
public enum PartialUpdateMode {
- @EnumFieldDescription(
- "No partial update logic should be employed.")
- NONE,
-
@EnumFieldDescription(
"For any column values missing in current record, pick value from previous version of the record.")
KEEP_VALUES,
- @EnumFieldDescription(
- "For column values missing in current record, pick the default value from the schema.")
- FILL_DEFAULTS,
-
@EnumFieldDescription(
"For columns having default values set in current record, pick the value from previous version of the record."
- + "Only top level data type default is checked, which means this mode does not check leaf level data type default"
- + "value for nested data types.")
+ + "Only top level data type default is checked, which means this mode does not check leaf level data type default"
+ + "value for nested data types.")
IGNORE_DEFAULTS,
@EnumFieldDescription(
- "For columns having marker in the current record, pick value from previous version of the record during write."
- + "Marker value can be defined using `hoodie.write.partial.update.custom.marker`, which should be added to"
- + "the value of table config `hoodie.write.partial.update.properties`. During upgrade, it could be"
- + "added automatically for some payload classes, e.g., `PostgresDebeziumAvroPayload`.")
- IGNORE_MARKERS
+ "For columns having default values or nulls set in current record, pick the value from previous version of the record."
+ + "Only top level data type default is checked, which means this mode does not check nested fields.")
+ IGNORE_DEFAULTS_NULLS,
Review Comment:
oh, we still have the other PR open right https://github.com/apache/hudi/pull/13519/
So, I did not integrate it fully yet. but the intention is that, IGNORE_DEFAULTS is for OverwriteNonDefaultsWithLatestAvroPayload and IGNORE_DEFAULTS_NULLS is for PartialUpdateAvroPayload
13519 is where we are tieing payload to the PartialUpdateMode. and thats why.
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3120996976
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
}, {
"hash" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129",
"triggerID" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"triggerType" : "PUSH"
}, {
"hash" : "2ac8fe4adf15f5f9903e175fe54a3d6693ad939e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7156",
"triggerID" : "2ac8fe4adf15f5f9903e175fe54a3d6693ad939e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ac8fe4adf15f5f9903e175fe54a3d6693ad939e Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7156)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231192502
##########
hudi-flink-datasource/hudi-flink1.16.x/pom.xml:
##########
@@ -0,0 +1,162 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
Review Comment:
oh, did not realize. will fix it
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231227284
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateStrategy.java:
##########
@@ -63,44 +68,82 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ Schema readerSchema,
+ boolean keepOldMetadataColumns,
+ TypedProperties props) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema, props);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
newRecord, oldRecord, newSchema, oldSchema);
default:
- return newRecord;
+ throw new HoodieIOException("Unsupported PartialUpdateMode " + partialUpdateMode + " detected");
}
}
+ BufferedRecord<T> reconcileBasedOnKeepValues(BufferedRecord<T> newRecord,
+ BufferedRecord<T> oldRecord,
+ Schema newSchema,
+ Schema oldSchema,
+ Schema readerSchema,
+ TypedProperties props) {
+
+ // Merge and store the combined record
+ Option<Pair<BufferedRecord, Schema>> deleteHandlingResult = handleDeletes(oldRecord, oldSchema, newRecord, newSchema, props);
+ if (deleteHandlingResult != null) {
+ return newRecord; // if deleted, return newRecord. newRecord.isDelete() will return anyways.
+ }
+
+ return (BufferedRecord<T>) keepValuesPartialMergingUtils.mergePartialRecords(oldRecord, oldSchema, newRecord, newSchema, readerSchema, readerContext).getLeft();
+ }
+
+ /**
+ * Basic handling of deletes that is used by many of the spark mergers
+ * returns null if merger specific logic should be used
+ */
+ protected Option<Pair<BufferedRecord, Schema>> handleDeletes(BufferedRecord older, Schema oldSchema, BufferedRecord newer, Schema newSchema, TypedProperties props) {
+
+ if (newer.isDelete()) {
+ // Delete record
+ return Option.empty();
+ }
+
+ // old record
+ if (older.isDelete()) {
+ return Option.of(Pair.of(newer, newSchema));
+ }
+ return null;
Review Comment:
actually we can remove this. at the beginning of PartialUpdateHandler.partialMerge(), we already account for all delete records
```
/**
* Merge records based on partial update mode.
* Note that {@param newRecord} refers to the record with higher commit time if COMMIT_TIME_ORDERING mode is used,
* or higher event time if EVENT_TIME_ORDERING mode us used.
*/
BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
Schema readerSchema,
boolean keepOldMetadataColumns) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
.
.
```
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3118401177
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
}, {
"hash" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129",
"triggerID" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3edf908d89bb9489147fef5ff74ef527b208fdf0 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117)
* 858c8ecd781e8731de288fb0ccdf54068fc5d83a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3116796800
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3edf908d89bb9489147fef5ff74ef527b208fdf0 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2232019216
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Class to assist with merging two versions of the record that may contain partial updates using
+ * {@link org.apache.hudi.common.table.PartialUpdateMode#KEEP_VALUES} mode.
+ */
+public class KeepValuesPartialMergingUtils<T> {
+ static KeepValuesPartialMergingUtils INSTANCE = new KeepValuesPartialMergingUtils();
+ private static final Map<Schema, Map<String, Integer>>
+ FIELD_NAME_TO_ID_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Pair<Pair<Schema, Schema>, Schema>, Schema>
+ MERGED_SCHEMA_CACHE = new ConcurrentHashMap<>();
+
+ /**
+ * Merges records which can contain partial updates.
+ *
+ * @param older Older record of type {@BufferedRecord<T>}.
+ * @param oldSchema Schema of the older record.
+ * @param newer Newer record of type {@BufferedRecord<T>}.
+ * @param newSchema Schema of the newer record.
+ * @param readerSchema Reader schema containing all the fields to read. This is used to maintain
+ * the ordering of the fields of the merged record.
+ * @param readerContext ReaderContext instance.
+ * @return The merged record and schema.
+ */
+ Pair<BufferedRecord<T>, Schema> mergePartialRecords(BufferedRecord<T> older,
+ Schema oldSchema,
+ BufferedRecord<T> newer,
+ Schema newSchema,
+ Schema readerSchema,
+ HoodieReaderContext<T> readerContext) {
+ // The merged schema contains fields that only appear in either older and/or newer record.
+ Schema mergedSchema =
+ getCachedMergedSchema(oldSchema, newSchema, readerSchema);
+ boolean isNewerPartial = isPartial(newSchema, mergedSchema);
+ if (!isNewerPartial) {
+ return Pair.of(newer, newSchema);
+ }
+ Set<String> fieldNamesInNewRecord =
+ getCachedFieldNameToIdMapping(newSchema).keySet();
+ // Collect field values.
+ List<Object> values = new ArrayList<>();
+ List<Schema.Field> mergedSchemaFields = mergedSchema.getFields();
+ for (Schema.Field mergedSchemaField : mergedSchemaFields) {
+ String fieldName = mergedSchemaField.name();
+ if (fieldNamesInNewRecord.contains(fieldName)) { // field present in newer record.
Review Comment:
At this point in the code if the merge mode is event time, are the older and newer records already in the correct order or is the "older" here just meaning it is in the previous set of data.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateHandler.java:
##########
@@ -41,17 +42,19 @@
* {@link BufferedRecordMergerFactory.CommitTimeBufferedRecordPartialUpdateMerger} and
* {@link BufferedRecordMergerFactory.EventTimeBufferedRecordPartialUpdateMerger}.
*/
-public class PartialUpdateStrategy<T> {
+public class PartialUpdateHandler<T> {
Review Comment:
I'm not seeing unit tests for this, let's make sure there are some unit test cases here.
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java:
##########
@@ -346,4 +365,80 @@ public float getProgress() throws IOException {
}
throw new IllegalStateException("getProgress() should not be called before a record reader has been initialized");
}
+
+ Schema resolveUnion(Schema schema) {
Review Comment:
`AvroSchemaUtils.resolveNullableSchema` provides similar functionality
##########
hudi-common/src/test/java/org/apache/hudi/common/table/read/TestKeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.table.read;
+
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.collection.Pair;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TestKeepValuesPartialMergingUtils {
+ private KeepValuesPartialMergingUtils<IndexedRecord> keepValuesPartialMergingUtils;
+ private HoodieReaderContext<IndexedRecord> mockReaderContext;
+ private Schema fullSchema;
+ private Schema partialSchema;
+ private Schema readerSchema;
+
+ @BeforeEach
+ void setUp() {
+ keepValuesPartialMergingUtils = new KeepValuesPartialMergingUtils<>();
+ mockReaderContext = mock(HoodieReaderContext.class);
+
+ // Create test schemas
+ fullSchema = Schema.createRecord("TestRecord", "Test record", "test", false);
+ fullSchema.setFields(Arrays.asList(
+ new Schema.Field("id", Schema.create(Schema.Type.STRING), "ID field", null),
+ new Schema.Field("name", Schema.create(Schema.Type.STRING), "Name field", null),
+ new Schema.Field("age", Schema.create(Schema.Type.INT), "Age field", null),
+ new Schema.Field("city", Schema.create(Schema.Type.STRING), "City field", null)
+ ));
+
+ partialSchema = Schema.createRecord("TestRecord", "Test record", "test", false);
+ partialSchema.setFields(Arrays.asList(
+ new Schema.Field("id", Schema.create(Schema.Type.STRING), "ID field", null),
+ new Schema.Field("name", Schema.create(Schema.Type.STRING), "Name field", null)
+ ));
+
+ readerSchema = Schema.createRecord("TestRecord", "Test record", "test", false);
Review Comment:
`readerSchema` is the same as `fullSchema`. I think you should have another schema that is different so the case where new, old, and reader schemas are all different.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateHandler.java:
##########
@@ -63,44 +66,65 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
+ Schema readerSchema,
boolean keepOldMetadataColumns) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
Review Comment:
This method fetches a config per record for the marker value, let's pull out the relevant configs in the constructor so it decreases the merging overhead.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Class to assist with merging two versions of the record that may contain partial updates using
+ * {@link org.apache.hudi.common.table.PartialUpdateMode#KEEP_VALUES} mode.
+ */
+public class KeepValuesPartialMergingUtils<T> {
+ static KeepValuesPartialMergingUtils INSTANCE = new KeepValuesPartialMergingUtils();
+ private static final Map<Schema, Map<String, Integer>>
+ FIELD_NAME_TO_ID_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Pair<Pair<Schema, Schema>, Schema>, Schema>
+ MERGED_SCHEMA_CACHE = new ConcurrentHashMap<>();
+
+ /**
+ * Merges records which can contain partial updates.
+ *
+ * @param older Older record of type {@BufferedRecord<T>}.
+ * @param oldSchema Schema of the older record.
+ * @param newer Newer record of type {@BufferedRecord<T>}.
+ * @param newSchema Schema of the newer record.
+ * @param readerSchema Reader schema containing all the fields to read. This is used to maintain
+ * the ordering of the fields of the merged record.
+ * @param readerContext ReaderContext instance.
+ * @return The merged record and schema.
+ */
+ Pair<BufferedRecord<T>, Schema> mergePartialRecords(BufferedRecord<T> older,
+ Schema oldSchema,
+ BufferedRecord<T> newer,
+ Schema newSchema,
+ Schema readerSchema,
+ HoodieReaderContext<T> readerContext) {
+ // The merged schema contains fields that only appear in either older and/or newer record.
+ Schema mergedSchema =
+ getCachedMergedSchema(oldSchema, newSchema, readerSchema);
+ boolean isNewerPartial = isPartial(newSchema, mergedSchema);
+ if (!isNewerPartial) {
+ return Pair.of(newer, newSchema);
+ }
+ Set<String> fieldNamesInNewRecord =
+ getCachedFieldNameToIdMapping(newSchema).keySet();
+ // Collect field values.
+ List<Object> values = new ArrayList<>();
+ List<Schema.Field> mergedSchemaFields = mergedSchema.getFields();
+ for (Schema.Field mergedSchemaField : mergedSchemaFields) {
+ String fieldName = mergedSchemaField.name();
+ if (fieldNamesInNewRecord.contains(fieldName)) { // field present in newer record.
+ values.add(readerContext.getValue(newer.getRecord(), newSchema, fieldName));
+ } else { // if not present in newer record pick from old record
+ values.add(readerContext.getValue(older.getRecord(), oldSchema, fieldName));
+ }
+ }
+ // Build merged record.
+ T engineRecord = readerContext.createEngineRecord(mergedSchema, values);
+ BufferedRecord<T> mergedRecord = new BufferedRecord<>(
+ newer.getRecordKey(),
+ newer.getOrderingValue(),
+ engineRecord,
+ readerContext.encodeAvroSchema(mergedSchema),
+ newer.isDelete());
+ return Pair.of(mergedRecord, mergedSchema);
+ }
+
+ /**
+ * @param avroSchema Avro schema.
+ * @return The field name to ID mapping.
+ */
+ static Map<String, Integer> getCachedFieldNameToIdMapping(Schema avroSchema) {
+ return FIELD_NAME_TO_ID_MAPPING_CACHE.computeIfAbsent(avroSchema, schema -> {
+ Map<String, Integer> schemaFieldIdMapping = new HashMap<>();
+ int fieldId = 0;
+ for (Schema.Field field : schema.getFields()) {
+ schemaFieldIdMapping.put(field.name(), fieldId);
+ fieldId++;
+ }
+ return schemaFieldIdMapping;
+ });
+ }
+
+ /**
+ * Merges the two schemas so the merged schema contains all the fields from the two schemas,
+ * with the same ordering of fields based on the provided reader schema.
+ *
+ * @param oldSchema Old schema.
+ * @param newSchema New schema.
+ * @param readerSchema Reader schema containing all the fields to read.
+ * @return The merged Avro schema.
+ */
+ static Schema getCachedMergedSchema(Schema oldSchema,
+ Schema newSchema,
+ Schema readerSchema) {
+ return MERGED_SCHEMA_CACHE.computeIfAbsent(
+ Pair.of(Pair.of(oldSchema, newSchema), readerSchema), schemaPair -> {
+ Schema schema1 = schemaPair.getLeft().getLeft();
+ Schema schema2 = schemaPair.getLeft().getRight();
+ Schema refSchema = schemaPair.getRight();
+ Set<String> schema1Keys =
+ getCachedFieldNameToIdMapping(schema1).keySet();
+ Set<String> schema2Keys =
+ getCachedFieldNameToIdMapping(schema2).keySet();
+ List<Schema.Field> mergedFieldList = new ArrayList<>();
+ for (int i = 0; i < refSchema.getFields().size(); i++) {
Review Comment:
The `refSchema` is the `newSchema` - this means we are only looking for fields in the new schema?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Class to assist with merging two versions of the record that may contain partial updates using
+ * {@link org.apache.hudi.common.table.PartialUpdateMode#KEEP_VALUES} mode.
+ */
+public class KeepValuesPartialMergingUtils<T> {
+ static KeepValuesPartialMergingUtils INSTANCE = new KeepValuesPartialMergingUtils();
+ private static final Map<Schema, Map<String, Integer>>
+ FIELD_NAME_TO_ID_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Pair<Pair<Schema, Schema>, Schema>, Schema>
+ MERGED_SCHEMA_CACHE = new ConcurrentHashMap<>();
+
+ /**
+ * Merges records which can contain partial updates.
+ *
+ * @param older Older record of type {@BufferedRecord<T>}.
+ * @param oldSchema Schema of the older record.
+ * @param newer Newer record of type {@BufferedRecord<T>}.
+ * @param newSchema Schema of the newer record.
+ * @param readerSchema Reader schema containing all the fields to read. This is used to maintain
+ * the ordering of the fields of the merged record.
+ * @param readerContext ReaderContext instance.
+ * @return The merged record and schema.
+ */
+ Pair<BufferedRecord<T>, Schema> mergePartialRecords(BufferedRecord<T> older,
+ Schema oldSchema,
+ BufferedRecord<T> newer,
+ Schema newSchema,
+ Schema readerSchema,
+ HoodieReaderContext<T> readerContext) {
+ // The merged schema contains fields that only appear in either older and/or newer record.
Review Comment:
nit: You can also describe this as the union of the fields
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Class to assist with merging two versions of the record that may contain partial updates using
+ * {@link org.apache.hudi.common.table.PartialUpdateMode#KEEP_VALUES} mode.
+ */
+public class KeepValuesPartialMergingUtils<T> {
+ static KeepValuesPartialMergingUtils INSTANCE = new KeepValuesPartialMergingUtils();
+ private static final Map<Schema, Map<String, Integer>>
+ FIELD_NAME_TO_ID_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Pair<Pair<Schema, Schema>, Schema>, Schema>
+ MERGED_SCHEMA_CACHE = new ConcurrentHashMap<>();
+
+ /**
+ * Merges records which can contain partial updates.
+ *
+ * @param older Older record of type {@BufferedRecord<T>}.
+ * @param oldSchema Schema of the older record.
Review Comment:
Can you note that this schema may not contain all the values due to operations writing only the values that change?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java:
##########
@@ -72,6 +72,9 @@ public void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec> keySpecO
// When a data block contains partial updates, subsequent record merging must always use
// partial merging.
enablePartialMerging = true;
+ if (partialUpdateMode.isEmpty()) {
+ this.partialUpdateMode = Option.of(PartialUpdateMode.KEEP_VALUES);
Review Comment:
Synced with Siva, this is because of the flag above. This block of code is hit with a merge into command that only writes out partial updates to the log files.
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3120845640
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
}, {
"hash" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129",
"triggerID" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"triggerType" : "PUSH"
}, {
"hash" : "2ac8fe4adf15f5f9903e175fe54a3d6693ad939e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2ac8fe4adf15f5f9903e175fe54a3d6693ad939e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 858c8ecd781e8731de288fb0ccdf54068fc5d83a Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129)
* 2ac8fe4adf15f5f9903e175fe54a3d6693ad939e UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230976224
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateStrategy.java:
##########
@@ -63,44 +68,82 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ Schema readerSchema,
+ boolean keepOldMetadataColumns,
+ TypedProperties props) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema, props);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
newRecord, oldRecord, newSchema, oldSchema);
default:
- return newRecord;
+ throw new HoodieIOException("Unsupported PartialUpdateMode " + partialUpdateMode + " detected");
}
}
+ BufferedRecord<T> reconcileBasedOnKeepValues(BufferedRecord<T> newRecord,
+ BufferedRecord<T> oldRecord,
+ Schema newSchema,
+ Schema oldSchema,
+ Schema readerSchema,
+ TypedProperties props) {
+
Review Comment:
extra new line.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3116616594
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3edf908d89bb9489147fef5ff74ef527b208fdf0 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230963336
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Generic util class to merge records of type T that may contain partial updates.
+ */
+public class KeepValuesPartialMergingUtils<T> {
+ private static final Map<Schema, Map<Integer, String>>
+ FIELD_ID_TO_NAME_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Schema, Map<String, Integer>>
+ FIELD_NAME_TO_ID_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Pair<Pair<Schema, Schema>, Schema>, Schema>
+ MERGED_SCHEMA_CACHE = new ConcurrentHashMap<>();
+
+ /**
+ * Merges records which can contain partial updates.
+ *
+ * @param older Older record of type {@BufferedRecord<T>}.
+ * @param oldSchema Schema of the older record.
+ * @param newer Newer record of type {@BufferedRecord<T>}.
+ * @param newSchema Schema of the newer record.
+ * @param readerSchema Reader schema containing all the fields to read. This is used to maintain
+ * the ordering of the fields of the merged record.
+ * @param readerContext ReaderContext instance.
+ * @return The merged record and schema.
+ */
+ public Pair<BufferedRecord<T>, Schema> mergePartialRecords(BufferedRecord<T> older,
+ Schema oldSchema,
+ BufferedRecord<T> newer,
+ Schema newSchema,
+ Schema readerSchema,
+ HoodieReaderContext<T> readerContext) {
+ // The merged schema contains fields that only appear in either older and/or newer record.
+ Schema mergedSchema =
+ getCachedMergedSchema(oldSchema, newSchema, readerSchema);
+ boolean isNewerPartial = isPartial(newSchema, mergedSchema);
+ if (!isNewerPartial) {
+ return Pair.of(newer, newSchema);
+ }
+ Set<String> fieldNamesInNewRecord =
+ getCachedFieldNameToIdMapping(newSchema).keySet();
+ // Collect field values.
+ List<Object> values = new ArrayList<>();
+ List<Schema.Field> mergedSchemaFields = mergedSchema.getFields();
+ for (int index = 0; index < mergedSchemaFields.size(); index++) {
+ String fieldName = mergedSchemaFields.get(index).name();
+ if (fieldNamesInNewRecord.contains(fieldName)) { // field present in newer record.
+ values.add(readerContext.getValue(newer.getRecord(), newSchema, fieldName));
+ } else { // if not present in newer record pick from old record
+ values.add(readerContext.getValue(older.getRecord(), oldSchema, fieldName));
+ }
+ }
+ // Build merged record.
+ T engineRecord = readerContext.createEngineRecord(mergedSchema, values);
+ BufferedRecord<T> mergedRecord = new BufferedRecord<>(
+ newer.getRecordKey(),
+ newer.getOrderingValue(),
+ engineRecord,
+ readerContext.encodeAvroSchema(mergedSchema),
+ newer.isDelete());
+ return Pair.of(mergedRecord, mergedSchema);
+ }
+
+ /**
+ * @param avroSchema Avro schema.
+ * @return The field ID to field name mapping.
+ */
+ public static Map<Integer, String> getCachedFieldIdToNameMapping(Schema avroSchema) {
Review Comment:
This function is not used anymore, can be removed. I remember I have removed it. Maybe I haven't pushed the latest one.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231188992
##########
hudi-common/src/main/java/org/apache/hudi/common/table/PartialUpdateMode.java:
##########
@@ -22,28 +22,24 @@
import org.apache.hudi.common.config.EnumFieldDescription;
public enum PartialUpdateMode {
- @EnumFieldDescription(
- "No partial update logic should be employed.")
- NONE,
-
@EnumFieldDescription(
"For any column values missing in current record, pick value from previous version of the record.")
KEEP_VALUES,
- @EnumFieldDescription(
- "For column values missing in current record, pick the default value from the schema.")
- FILL_DEFAULTS,
-
@EnumFieldDescription(
"For columns having default values set in current record, pick the value from previous version of the record."
- + "Only top level data type default is checked, which means this mode does not check leaf level data type default"
- + "value for nested data types.")
+ + "Only top level data type default is checked, which means this mode does not check leaf level data type default"
+ + "value for nested data types.")
IGNORE_DEFAULTS,
@EnumFieldDescription(
- "For columns having marker in the current record, pick value from previous version of the record during write."
- + "Marker value can be defined using `hoodie.write.partial.update.custom.marker`, which should be added to"
- + "the value of table config `hoodie.write.partial.update.properties`. During upgrade, it could be"
- + "added automatically for some payload classes, e.g., `PostgresDebeziumAvroPayload`.")
- IGNORE_MARKERS
+ "For columns having default values or nulls set in current record, pick the value from previous version of the record."
+ + "Only top level data type default is checked, which means this mode does not check nested fields.")
+ IGNORE_DEFAULTS_NULLS,
Review Comment:
oh, we still have the other PR open right https://github.com/apache/hudi/pull/13519/
So, I did not integrate it fully yet. but the intention is that, IGNORE_DEFAULTS is for OverwriteNonDefaultsWithLatestAvroPayload and IGNORE_DEFAULTS_NULLS is for PartialUpdateAvroPayload
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230954096
##########
hudi-common/src/main/java/org/apache/hudi/common/table/PartialUpdateMode.java:
##########
@@ -22,28 +22,24 @@
import org.apache.hudi.common.config.EnumFieldDescription;
public enum PartialUpdateMode {
- @EnumFieldDescription(
- "No partial update logic should be employed.")
- NONE,
-
@EnumFieldDescription(
"For any column values missing in current record, pick value from previous version of the record.")
KEEP_VALUES,
- @EnumFieldDescription(
- "For column values missing in current record, pick the default value from the schema.")
- FILL_DEFAULTS,
-
@EnumFieldDescription(
"For columns having default values set in current record, pick the value from previous version of the record."
- + "Only top level data type default is checked, which means this mode does not check leaf level data type default"
- + "value for nested data types.")
+ + "Only top level data type default is checked, which means this mode does not check leaf level data type default"
+ + "value for nested data types.")
IGNORE_DEFAULTS,
@EnumFieldDescription(
- "For columns having marker in the current record, pick value from previous version of the record during write."
- + "Marker value can be defined using `hoodie.write.partial.update.custom.marker`, which should be added to"
- + "the value of table config `hoodie.write.partial.update.properties`. During upgrade, it could be"
- + "added automatically for some payload classes, e.g., `PostgresDebeziumAvroPayload`.")
- IGNORE_MARKERS
+ "For columns having default values or nulls set in current record, pick the value from previous version of the record."
+ + "Only top level data type default is checked, which means this mode does not check nested fields.")
+ IGNORE_DEFAULTS_NULLS,
Review Comment:
Both `OverwriteNonDefaultsWithLatestAvroPayload` and `PartialUpdateAvroPayload` uses `IGNORE_DEFAULTS`. Not sure which payloads uses `IGNORE_DEFAULTS_NULLS` mode. Meanwhile, default value of most columns are NULL. so, not sure the real difference between this two modes.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3116612287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3edf908d89bb9489147fef5ff74ef527b208fdf0 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231485624
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java:
##########
@@ -124,12 +124,126 @@ void testConstructEngineRecordWithNullValueFromBase() {
BufferedRecord<InternalRow> record = new BufferedRecord<>(
"record_key", 1, base, 1, false);
- InternalRow result = readerContext.constructEngineRecord(SCHEMA, updates, record);
+ InternalRow result = readerContext.mergeEngineRecord(SCHEMA, updates, record);
assertTrue(result.isNullAt(0));
assertEquals("Dan", result.getUTF8String(1).toString());
assertTrue(result.getBoolean(2));
}
+ @Test
+ void testConstructEngineRecordWithListOfValues() {
+ List<Object> values = Arrays.asList(1, UTF8String.fromString("Alice"), true);
+ InternalRow result = readerContext.createEngineRecord(SCHEMA, values);
+
+ assertEquals(1, result.getInt(0));
+ assertEquals("Alice", result.getString(1));
+ assertTrue(result.getBoolean(2));
+ }
+
+ @Test
+ void testConstructEngineRecordWithNullValues() {
+ List<Object> values = Arrays.asList(null, UTF8String.fromString("Bob"), null);
+ InternalRow result = readerContext.createEngineRecord(SCHEMA, values);
+
+ assertTrue(result.isNullAt(0));
+ assertEquals("Bob", result.getString(1));
+ assertTrue(result.isNullAt(2));
+ }
+
+ @Test
+ void testConstructEngineRecordWithMixedTypes() {
+ List<Object> values = Arrays.asList(42, UTF8String.fromString("Carol"), false);
+ InternalRow result = readerContext.createEngineRecord(SCHEMA, values);
+
+ assertEquals(42, result.getInt(0));
+ assertEquals("Carol", result.getString(1));
+ assertFalse(result.getBoolean(2));
+ }
+
+ @Test
+ void testConstructEngineRecordWithEmptyValues() {
+ List<Object> values = Arrays.asList(0, UTF8String.fromString(""), false);
+ InternalRow result = readerContext.createEngineRecord(SCHEMA, values);
+
+ assertEquals(0, result.getInt(0));
+ assertEquals("", result.getString(1));
+ assertFalse(result.getBoolean(2));
+ }
+
+ @Test
+ void testConstructEngineRecordWithValueCountMismatch() {
+ List<Object> values = Arrays.asList(1, UTF8String.fromString("Alice")); // Missing boolean value
+
+ try {
+ readerContext.createEngineRecord(SCHEMA, values);
+ // Should not reach here
+ assertTrue(false, "Expected IllegalArgumentException");
+ } catch (IllegalArgumentException e) {
+ assertTrue(e.getMessage().contains("Value count (2) does not match field count (3)"));
+ }
+ }
+
+ @Test
+ void testConstructEngineRecordWithExtraValues() {
+ List<Object> values = Arrays.asList(1, UTF8String.fromString("Alice"), true, "extra");
+
+ try {
+ readerContext.createEngineRecord(SCHEMA, values);
+ // Should not reach here
+ assertTrue(false, "Expected IllegalArgumentException");
+ } catch (IllegalArgumentException e) {
+ assertTrue(e.getMessage().contains("Value count (4) does not match field count (3)"));
+ }
+ }
+
+ @Test
+ void testConstructEngineRecordWithComplexSchema() {
+ // Create a more complex schema with nested fields
+ Schema complexSchema = SchemaBuilder.record("ComplexRecord").fields()
Review Comment:
We need at least one test with nested fields for sanity.
##########
hudi-common/src/test/java/org/apache/hudi/common/table/read/TestKeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.table.read;
+
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.collection.Pair;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TestKeepValuesPartialMergingUtils {
+ private KeepValuesPartialMergingUtils<IndexedRecord> keepValuesPartialMergingUtils;
+ private HoodieReaderContext<IndexedRecord> mockReaderContext;
+ private Schema fullSchema;
+ private Schema partialSchema;
+ private Schema readerSchema;
+
+ @BeforeEach
+ void setUp() {
+ keepValuesPartialMergingUtils = new KeepValuesPartialMergingUtils<>();
+ mockReaderContext = mock(HoodieReaderContext.class);
+
+ // Create test schemas
Review Comment:
Let's make some schemas with nested fields here as well. Maybe lists and maps for robustness.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java:
##########
@@ -129,6 +129,16 @@ public InternalRow constructEngineRecord(Schema schema,
return new GenericInternalRow(values);
}
+ @Override
+ public InternalRow createEngineRecord(Schema schema, List<Object> values) {
+ List<Schema.Field> fields = schema.getFields();
+ if (fields.size() != values.size()) {
+ throw new IllegalArgumentException(
+ "Value count (" + values.size() + ") does not match field count (" + fields.size() + ")");
Review Comment:
This is only validating the first level of the schema, it does not account for nested fields. Will the toBinaryRow handle validating nested fields?
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java:
##########
@@ -230,16 +234,31 @@ public HoodieRecord<ArrayWritable> constructHoodieRecord(BufferedRecord<ArrayWri
}
@Override
- public ArrayWritable constructEngineRecord(Schema schema,
- Map<Integer, Object> updateValues,
- BufferedRecord<ArrayWritable> baseRecord) {
+ public ArrayWritable mergeEngineRecord(Schema schema,
+ Map<Integer, Object> updateValues,
+ BufferedRecord<ArrayWritable> baseRecord) {
Writable[] engineRecord = baseRecord.getRecord().get();
for (Map.Entry<Integer, Object> value : updateValues.entrySet()) {
engineRecord[value.getKey()] = (Writable) value.getValue();
}
return baseRecord.getRecord();
}
+ @Override
+ public ArrayWritable createEngineRecord(Schema schema, List<Object> values) {
+ List<Schema.Field> fields = schema.getFields();
+ if (fields.size() != values.size()) {
+ throw new IllegalArgumentException("Mismatch between schema fields and values");
+ }
+
+ Writable[] writables = new Writable[fields.size()];
+ for (int i = 0; i < fields.size(); i++) {
+ Schema fieldSchema = resolveUnion(fields.get(i).schema());
+ writables[i] = convertToWritable(fieldSchema, values.get(i));
Review Comment:
This is the only engine that is converting the output into engine specific format. Do the other engines need to do this?
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieReaderContext.java:
##########
@@ -413,9 +413,18 @@ public Comparable getOrderingValue(T record,
* @param baseRecord The record based on which the engine record is built.
* @return A new instance of engine record type {@link T}.
*/
- public abstract T constructEngineRecord(Schema schema,
- Map<Integer, Object> updateValues,
- BufferedRecord<T> baseRecord);
+ public abstract T mergeEngineRecord(Schema schema,
+ Map<Integer, Object> updateValues,
+ BufferedRecord<T> baseRecord);
+
+ /**
+ * Creates a new Engine based record based on a given schema and its field values.
+ *
+ * @param schema The schema of the new record.
+ * @param values The list of values.
+ * @return A new instance of engine record type {@link T}.
+ */
+ public abstract T createEngineRecord(Schema schema, List<Object> values);
Review Comment:
Let's consider using an array of values instead. It looks like in some cases we can simplify the code a bit with this.
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3119047833
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
}, {
"hash" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129",
"triggerID" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 858c8ecd781e8731de288fb0ccdf54068fc5d83a Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3120847469
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
}, {
"hash" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129",
"triggerID" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"triggerType" : "PUSH"
}, {
"hash" : "2ac8fe4adf15f5f9903e175fe54a3d6693ad939e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7156",
"triggerID" : "2ac8fe4adf15f5f9903e175fe54a3d6693ad939e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 858c8ecd781e8731de288fb0ccdf54068fc5d83a Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7129)
* 2ac8fe4adf15f5f9903e175fe54a3d6693ad939e Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7156)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2230985993
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateStrategy.java:
##########
@@ -63,44 +68,82 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ Schema readerSchema,
+ boolean keepOldMetadataColumns,
+ TypedProperties props) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema, props);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
newRecord, oldRecord, newSchema, oldSchema);
default:
- return newRecord;
+ throw new HoodieIOException("Unsupported PartialUpdateMode " + partialUpdateMode + " detected");
}
}
+ BufferedRecord<T> reconcileBasedOnKeepValues(BufferedRecord<T> newRecord,
+ BufferedRecord<T> oldRecord,
+ Schema newSchema,
+ Schema oldSchema,
+ Schema readerSchema,
+ TypedProperties props) {
+
+ // Merge and store the combined record
+ Option<Pair<BufferedRecord, Schema>> deleteHandlingResult = handleDeletes(oldRecord, oldSchema, newRecord, newSchema, props);
+ if (deleteHandlingResult != null) {
+ return newRecord; // if deleted, return newRecord. newRecord.isDelete() will return anyways.
+ }
+
+ return (BufferedRecord<T>) keepValuesPartialMergingUtils.mergePartialRecords(oldRecord, oldSchema, newRecord, newSchema, readerSchema, readerContext).getLeft();
+ }
+
+ /**
+ * Basic handling of deletes that is used by many of the spark mergers
+ * returns null if merger specific logic should be used
+ */
+ protected Option<Pair<BufferedRecord, Schema>> handleDeletes(BufferedRecord older, Schema oldSchema, BufferedRecord newer, Schema newSchema, TypedProperties props) {
+
+ if (newer.isDelete()) {
+ // Delete record
+ return Option.empty();
+ }
+
+ // old record
+ if (older.isDelete()) {
+ return Option.of(Pair.of(newer, newSchema));
+ }
+ return null;
Review Comment:
This logic can be simplified as:
if (newer.isDelete() || older.isDelete()) return Option.of(Pair.of(newer, newSchema);
else return Option.empty();
Then the calling site can handle Option instead of NULL.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231191870
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PartialUpdateStrategy.java:
##########
@@ -63,44 +68,82 @@ BufferedRecord<T> partialMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ Schema readerSchema,
+ boolean keepOldMetadataColumns,
+ TypedProperties props) {
// Note that: When either newRecord or oldRecord is a delete record,
// skip partial update since delete records do not have meaningful columns.
- if (partialUpdateMode == PartialUpdateMode.NONE
- || null == oldRecord
+ if (null == oldRecord
|| newRecord.isDelete()
|| oldRecord.isDelete()) {
return newRecord;
}
switch (partialUpdateMode) {
case KEEP_VALUES:
- case FILL_DEFAULTS:
- return newRecord;
+ return reconcileBasedOnKeepValues(newRecord, oldRecord, newSchema, oldSchema, readerSchema, props);
case IGNORE_DEFAULTS:
return reconcileDefaultValues(
- newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns);
- case IGNORE_MARKERS:
+ newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, false);
+ case IGNORE_DEFAULTS_NULLS:
+ return reconcileDefaultValues(newRecord, oldRecord, newSchema, oldSchema, keepOldMetadataColumns, true);
+ case FILL_UNAVAILABLE:
return reconcileMarkerValues(
newRecord, oldRecord, newSchema, oldSchema);
default:
- return newRecord;
+ throw new HoodieIOException("Unsupported PartialUpdateMode " + partialUpdateMode + " detected");
}
}
+ BufferedRecord<T> reconcileBasedOnKeepValues(BufferedRecord<T> newRecord,
+ BufferedRecord<T> oldRecord,
+ Schema newSchema,
+ Schema oldSchema,
+ Schema readerSchema,
+ TypedProperties props) {
+
+ // Merge and store the combined record
+ Option<Pair<BufferedRecord, Schema>> deleteHandlingResult = handleDeletes(oldRecord, oldSchema, newRecord, newSchema, props);
+ if (deleteHandlingResult != null) {
+ return newRecord; // if deleted, return newRecord. newRecord.isDelete() will return anyways.
+ }
+
+ return (BufferedRecord<T>) keepValuesPartialMergingUtils.mergePartialRecords(oldRecord, oldSchema, newRecord, newSchema, readerSchema, readerContext).getLeft();
+ }
+
+ /**
+ * Basic handling of deletes that is used by many of the spark mergers
+ * returns null if merger specific logic should be used
+ */
+ protected Option<Pair<BufferedRecord, Schema>> handleDeletes(BufferedRecord older, Schema oldSchema, BufferedRecord newer, Schema newSchema, TypedProperties props) {
+
+ if (newer.isDelete()) {
+ // Delete record
+ return Option.empty();
+ }
+
+ // old record
+ if (older.isDelete()) {
+ return Option.of(Pair.of(newer, newSchema));
+ }
+ return null;
+ }
+
/**
* @param newRecord The newer record determined by the merge mode.
* @param oldRecord The older record determined by the merge mode.
* @param newSchema The schema of the newer record.
* @param oldSchema The schema of the older record.
* @param keepOldMetadataColumns Keep the metadata columns from the older record.
+ * @param includeNulls true if nulls are expected to be ignored as well. false otherwise.
* @return
*/
BufferedRecord<T> reconcileDefaultValues(BufferedRecord<T> newRecord,
BufferedRecord<T> oldRecord,
Schema newSchema,
Schema oldSchema,
- boolean keepOldMetadataColumns) {
+ boolean keepOldMetadataColumns,
+ boolean includeNulls) {
Review Comment:
as commented above, its not yet fully integrated as the other PR is still open.
--
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
Re: [PR] [HUDI-8401] Unifying. Partial Update modes from Table Property and Merge Into Partial Update Encoding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13623:
URL: https://github.com/apache/hudi/pull/13623#issuecomment-3118387193
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117",
"triggerID" : "3edf908d89bb9489147fef5ff74ef527b208fdf0",
"triggerType" : "PUSH"
}, {
"hash" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "858c8ecd781e8731de288fb0ccdf54068fc5d83a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3edf908d89bb9489147fef5ff74ef527b208fdf0 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=7117)
* 858c8ecd781e8731de288fb0ccdf54068fc5d83a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [HUDI-8401] Unifying Partial Update modes and Merge Into Partial Update Encoding [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #13623:
URL: https://github.com/apache/hudi/pull/13623#discussion_r2231460722
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java:
##########
@@ -72,6 +72,9 @@ public void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec> keySpecO
// When a data block contains partial updates, subsequent record merging must always use
// partial merging.
enablePartialMerging = true;
+ if (partialUpdateMode.isEmpty()) {
+ this.partialUpdateMode = Option.of(PartialUpdateMode.KEEP_VALUES);
Review Comment:
Why override the input?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -1097,12 +1096,16 @@ public Set<String> getMetadataPartitions() {
CONFIG_VALUES_DELIMITER));
}
- public PartialUpdateMode getPartialUpdateMode() {
+ public Option<PartialUpdateMode> getPartialUpdateMode() {
Review Comment:
Why make this empty instead of None or some Unsupported enum type?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java:
##########
@@ -60,29 +60,18 @@ public static <T> BufferedRecordMerger<T> create(HoodieReaderContext<T> readerCo
Option<String> payloadClass,
Schema readerSchema,
TypedProperties props,
- PartialUpdateMode partialUpdateMode) {
- /**
- * This part implements KEEP_VALUES partial update mode, which merges two records that do not have all columns.
- * Other Partial update modes, like IGNORE_DEFAULTS assume all columns exists in the record,
- * but some columns contain specific values that should be replaced by that from older version of the record.
- */
- if (enablePartialMerging) {
- BufferedRecordMerger<T> deleteRecordMerger = create(
- readerContext, recordMergeMode, false, recordMerger, orderingFieldName, payloadClass, readerSchema, props, partialUpdateMode);
- return new PartialUpdateBufferedRecordMerger<>(readerContext, recordMerger, deleteRecordMerger, readerSchema, props);
- }
-
+ Option<PartialUpdateMode> partialUpdateMode) {
Review Comment:
`enablePartialMerging` is the same as `partialUpdateMode.isPresent` at this point. Is there a case where we see this deviating?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Class to assist with merging two versions of the record that may contain partial updates using
+ * {@link org.apache.hudi.common.table.PartialUpdateMode#KEEP_VALUES} mode.
+ */
+public class KeepValuesPartialMergingUtils<T> {
+ static KeepValuesPartialMergingUtils INSTANCE = new KeepValuesPartialMergingUtils();
+ private static final Map<Schema, Map<String, Integer>>
+ FIELD_NAME_TO_ID_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Pair<Pair<Schema, Schema>, Schema>, Schema>
+ MERGED_SCHEMA_CACHE = new ConcurrentHashMap<>();
Review Comment:
This will grow indefinitely for long running jobs. Can these become instance variables that so their lifecycle is limited to the life of that merger?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Class to assist with merging two versions of the record that may contain partial updates using
+ * {@link org.apache.hudi.common.table.PartialUpdateMode#KEEP_VALUES} mode.
+ */
+public class KeepValuesPartialMergingUtils<T> {
Review Comment:
If this is only intended to be used in the record merging, let's make this package private.
Also I really dislike the use of `Utils` as a naming convention for things that are not simple utils like string manipulation. Let's give this a proper name please.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeepValuesPartialMergingUtils.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.read;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Class to assist with merging two versions of the record that may contain partial updates using
+ * {@link org.apache.hudi.common.table.PartialUpdateMode#KEEP_VALUES} mode.
+ */
+public class KeepValuesPartialMergingUtils<T> {
+ static KeepValuesPartialMergingUtils INSTANCE = new KeepValuesPartialMergingUtils();
+ private static final Map<Schema, Map<String, Integer>>
+ FIELD_NAME_TO_ID_MAPPING_CACHE = new ConcurrentHashMap<>();
+ private static final Map<Pair<Pair<Schema, Schema>, Schema>, Schema>
+ MERGED_SCHEMA_CACHE = new ConcurrentHashMap<>();
+
+ /**
+ * Merges records which can contain partial updates.
+ *
+ * @param older Older record of type {@BufferedRecord<T>}.
+ * @param oldSchema Schema of the older record.
+ * @param newer Newer record of type {@BufferedRecord<T>}.
+ * @param newSchema Schema of the newer record.
+ * @param readerSchema Reader schema containing all the fields to read. This is used to maintain
+ * the ordering of the fields of the merged record.
+ * @param readerContext ReaderContext instance.
+ * @return The merged record and schema.
+ */
+ Pair<BufferedRecord<T>, Schema> mergePartialRecords(BufferedRecord<T> older,
+ Schema oldSchema,
+ BufferedRecord<T> newer,
+ Schema newSchema,
+ Schema readerSchema,
+ HoodieReaderContext<T> readerContext) {
+ // The merged schema contains fields that only appear in either older and/or newer record.
+ Schema mergedSchema =
+ getCachedMergedSchema(oldSchema, newSchema, readerSchema);
+ boolean isNewerPartial = isPartial(newSchema, mergedSchema);
+ if (!isNewerPartial) {
+ return Pair.of(newer, newSchema);
+ }
+ Set<String> fieldNamesInNewRecord =
+ getCachedFieldNameToIdMapping(newSchema).keySet();
+ // Collect field values.
+ List<Object> values = new ArrayList<>();
+ List<Schema.Field> mergedSchemaFields = mergedSchema.getFields();
+ for (Schema.Field mergedSchemaField : mergedSchemaFields) {
+ String fieldName = mergedSchemaField.name();
+ if (fieldNamesInNewRecord.contains(fieldName)) { // field present in newer record.
+ values.add(readerContext.getValue(newer.getRecord(), newSchema, fieldName));
+ } else { // if not present in newer record pick from old record
+ values.add(readerContext.getValue(older.getRecord(), oldSchema, fieldName));
+ }
+ }
+ // Build merged record.
+ T engineRecord = readerContext.createEngineRecord(mergedSchema, values);
+ BufferedRecord<T> mergedRecord = new BufferedRecord<>(
+ newer.getRecordKey(),
+ newer.getOrderingValue(),
+ engineRecord,
+ readerContext.encodeAvroSchema(mergedSchema),
+ newer.isDelete());
+ return Pair.of(mergedRecord, mergedSchema);
+ }
+
+ /**
+ * @param avroSchema Avro schema.
+ * @return The field name to ID mapping.
+ */
+ static Map<String, Integer> getCachedFieldNameToIdMapping(Schema avroSchema) {
+ return FIELD_NAME_TO_ID_MAPPING_CACHE.computeIfAbsent(avroSchema, schema -> {
+ Map<String, Integer> schemaFieldIdMapping = new HashMap<>();
+ int fieldId = 0;
+ for (Schema.Field field : schema.getFields()) {
+ schemaFieldIdMapping.put(field.name(), fieldId);
+ fieldId++;
+ }
+ return schemaFieldIdMapping;
+ });
+ }
+
+ /**
+ * Merges the two schemas so the merged schema contains all the fields from the two schemas,
+ * with the same ordering of fields based on the provided reader schema.
+ *
+ * @param oldSchema Old schema.
+ * @param newSchema New schema.
+ * @param readerSchema Reader schema containing all the fields to read.
+ * @return The merged Avro schema.
+ */
+ static Schema getCachedMergedSchema(Schema oldSchema,
+ Schema newSchema,
+ Schema readerSchema) {
+ return MERGED_SCHEMA_CACHE.computeIfAbsent(
+ Pair.of(Pair.of(oldSchema, newSchema), readerSchema), schemaPair -> {
+ Schema schema1 = schemaPair.getLeft().getLeft();
+ Schema schema2 = schemaPair.getLeft().getRight();
+ Schema refSchema = schemaPair.getRight();
+ Set<String> schema1Keys =
+ getCachedFieldNameToIdMapping(schema1).keySet();
+ Set<String> schema2Keys =
+ getCachedFieldNameToIdMapping(schema2).keySet();
+ List<Schema.Field> mergedFieldList = new ArrayList<>();
+ for (int i = 0; i < refSchema.getFields().size(); i++) {
+ Schema.Field field = refSchema.getFields().get(i);
+ if (schema1Keys.contains(field.name()) || schema2Keys.contains(field.name())) {
+ mergedFieldList.add(new Schema.Field(
+ field.name(),
+ field.schema(),
+ field.doc(),
+ field.defaultVal()));
+ }
+ }
+ Schema mergedSchema = Schema.createRecord(
+ readerSchema.getName(),
+ readerSchema.getDoc(),
+ readerSchema.getNamespace(),
+ false);
+ mergedSchema.setFields(mergedFieldList);
+ return mergedSchema;
+ });
+ }
+
+ /**
+ * @param schema Avro schema to check.
+ * @param mergedSchema The merged schema for the merged record.
+ * @return whether the Avro schema is partial compared to the merged schema.
+ */
+ @VisibleForTesting
+ public static boolean isPartial(Schema schema, Schema mergedSchema) {
+ return !schema.equals(mergedSchema);
Review Comment:
`equals` also inspects the props on the schema, we are probably only worried about field equality 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org