You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "linliu-code (via GitHub)" <gi...@apache.org> on 2024/11/30 18:17:47 UTC
[PR] [HUDI-8522] Compaction partial update new fg [hudi]
linliu-code opened a new pull request, #12384:
URL: https://github.com/apache/hudi/pull/12384
### Change Logs
_Describe context and summary for this change. Highlight if any code was copied._
### Impact
_Describe any public API or user-facing feature change or any performance impact._
### Risk level (write none, low medium or high below)
_If medium or high, explain what verification was done to mitigate the risks._
### 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864747957
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java:
##########
@@ -83,7 +86,26 @@ public Object getValue(InternalRow row, Schema schema, String fieldName) {
@Override
public String getRecordKey(InternalRow row, Schema schema) {
- return getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD).toString();
+ Object key = getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD);
+ if (key != null) {
+ return key.toString();
+ }
+ return null;
+ }
+
+ @Override
+ public String getRecordKey(InternalRow row, Schema schema, TypedProperties props) {
+ String key = getRecordKey(row, schema);
Review Comment:
During my experiments, the record key may be found by checking different configs. Existing code tries to retrieve it in only one way, which will cause null-key issue.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864723363
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -130,8 +151,14 @@ public HoodieData<WriteStatus> compact(
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
// if this is a MDT, set up the instant range of log reader just like regular MDT snapshot reader.
Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient);
+
+ // Broadcast necessary information.
+ Option<CompactorBroadcastManager> broadcastManagerOpt = getCompactorBroadcastManager(context);
Review Comment:
Like you said, partial update check can go before it to avoid broadcast as of now.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509607390
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
}, {
"hash" : "1ab58804320d995919865e16fb29f909f898a062",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ab58804320d995919865e16fb29f909f898a062",
"triggerType" : "PUSH"
}, {
"hash" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"triggerType" : "PUSH"
}, {
"hash" : "c2ea5b14785a2d0845a76bd16ed6dbcc42744af5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2144",
"triggerID" : "c2ea5b14785a2d0845a76bd16ed6dbcc42744af5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5ffc8c5ad207e7b8f293d15753bb78d485742a8a Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143)
* 1ab58804320d995919865e16fb29f909f898a062 UNKNOWN
* d9602ea0bc5dc6a2f9b91594cd1937d39803650c UNKNOWN
* c2ea5b14785a2d0845a76bd16ed6dbcc42744af5 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2144)
<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-8522] Compaction using fg reader for partial update [hudi]
Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.
vinothchandar closed pull request #12384: [HUDI-8522] Compaction using fg reader for partial update
URL: https://github.com/apache/hudi/pull/12384
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864692809
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieUnMergedLogRecordScanner.java:
##########
@@ -19,6 +19,7 @@
package org.apache.hudi.common.table.log;
import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieKey;
Review Comment:
The changes in this file should be gone after rebase.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864751295
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java:
##########
@@ -122,19 +122,19 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>> deletePrepped(HoodieEngineCo
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> upsertPrepped(HoodieEngineContext context, String instantTime,
- HoodieData<HoodieRecord<T>> preppedRecords) {
+ HoodieData<HoodieRecord<T>> preppedRecords) {
return new SparkUpsertPreppedDeltaCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, preppedRecords).execute();
}
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> insertPrepped(HoodieEngineContext context, String instantTime,
- HoodieData<HoodieRecord<T>> preppedRecords) {
+ HoodieData<HoodieRecord<T>> preppedRecords) {
Review Comment:
yeah.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864747957
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java:
##########
@@ -83,7 +86,26 @@ public Object getValue(InternalRow row, Schema schema, String fieldName) {
@Override
public String getRecordKey(InternalRow row, Schema schema) {
- return getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD).toString();
+ Object key = getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD);
+ if (key != null) {
+ return key.toString();
+ }
+ return null;
+ }
+
+ @Override
+ public String getRecordKey(InternalRow row, Schema schema, TypedProperties props) {
+ String key = getRecordKey(row, schema);
Review Comment:
During my experiments, the record key may be found by checking different configs. Existing code tries to retrieve it in only one way, which will cause null-key issue.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509594366
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
}, {
"hash" : "1ab58804320d995919865e16fb29f909f898a062",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ab58804320d995919865e16fb29f909f898a062",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5ffc8c5ad207e7b8f293d15753bb78d485742a8a Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143)
* 1ab58804320d995919865e16fb29f909f898a062 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864740750
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java:
##########
@@ -112,12 +112,12 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>> upsert(HoodieEngineContext c
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> insert(HoodieEngineContext context, String instantTime, HoodieData<HoodieRecord<T>> records) {
- return new SparkInsertCommitActionExecutor<>((HoodieSparkEngineContext)context, config, this, instantTime, records).execute();
+ return new SparkInsertCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, records).execute();
}
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> bulkInsert(HoodieEngineContext context, String instantTime, HoodieData<HoodieRecord<T>> records,
- Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner) {
+ Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner) {
Review Comment:
yes, I thought they should be removed.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.
vinothchandar commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2510998346
Closing this in favor of #[12390](https://github.com/apache/hudi/pull/12390)
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864729324
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -130,8 +151,14 @@ public HoodieData<WriteStatus> compact(
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
// if this is a MDT, set up the instant range of log reader just like regular MDT snapshot reader.
Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient);
+
+ // Broadcast necessary information.
+ Option<CompactorBroadcastManager> broadcastManagerOpt = getCompactorBroadcastManager(context);
Review Comment:
> and partial update is just a part of it. So emphasizing partial update is too specific.
+1
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -130,8 +151,14 @@ public HoodieData<WriteStatus> compact(
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
// if this is a MDT, set up the instant range of log reader just like regular MDT snapshot reader.
Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient);
+
+ // Broadcast necessary information.
+ Option<CompactorBroadcastManager> broadcastManagerOpt = getCompactorBroadcastManager(context);
Review Comment:
> and partial update is just a part of it. So emphasizing partial update is too specific.
+1
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864723411
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -130,8 +151,14 @@ public HoodieData<WriteStatus> compact(
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
// if this is a MDT, set up the instant range of log reader just like regular MDT snapshot reader.
Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient);
+
+ // Broadcast necessary information.
+ Option<CompactorBroadcastManager> broadcastManagerOpt = getCompactorBroadcastManager(context);
+ if (broadcastManagerOpt.isPresent()) {
+ broadcastManagerOpt.get().prepareAndBroadcast();
Review Comment:
yes
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864692974
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestCompactionWithFileGroupReader.scala:
##########
@@ -0,0 +1,148 @@
+/*
Review Comment:
This test has been updated to have both cases: enabled or disabled partial updates.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509595776
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
}, {
"hash" : "1ab58804320d995919865e16fb29f909f898a062",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ab58804320d995919865e16fb29f909f898a062",
"triggerType" : "PUSH"
}, {
"hash" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"triggerType" : "PUSH"
}, {
"hash" : "c2ea5b14785a2d0845a76bd16ed6dbcc42744af5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c2ea5b14785a2d0845a76bd16ed6dbcc42744af5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5ffc8c5ad207e7b8f293d15753bb78d485742a8a Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143)
* 1ab58804320d995919865e16fb29f909f898a062 UNKNOWN
* d9602ea0bc5dc6a2f9b91594cd1937d39803650c UNKNOWN
* c2ea5b14785a2d0845a76bd16ed6dbcc42744af5 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864751904
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieUnMergedLogRecordScanner.java:
##########
@@ -19,6 +19,7 @@
package org.apache.hudi.common.table.log;
import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieKey;
Review Comment:
I rebased, but it did not go. I think it is from your branch. I will try to override 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864740303
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -189,26 +218,58 @@ public List<WriteStatus> compact(HoodieCompactionHandler compactionHandler,
new StoragePath(FSUtils.constructAbsolutePath(
metaClient.getBasePath(), operation.getPartitionPath()), p).toString())
.collect(toList());
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(storage)
- .withBasePath(metaClient.getBasePath())
- .withLogFilePaths(logFiles)
- .withReaderSchema(readerSchema)
- .withLatestInstantTime(executionHelper.instantTimeToUseForScanning(instantTime, maxInstantTime))
- .withInstantRange(instantRange)
- .withInternalSchema(internalSchemaOption.orElse(InternalSchema.getEmptyInternalSchema()))
- .withMaxMemorySizeInBytes(maxMemoryPerCompaction)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withOperationField(config.allowOperationMetadataField())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(executionHelper.enableOptimizedLogBlockScan(config))
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(metaClient)
+
+ boolean isPartialUpdateEnabled = config.getBooleanOrDefault("hoodie.spark.sql.merge.into.partial.updates", false);
+ if (isPartialUpdateEnabled && broadcastManagerOpt.isPresent() && !isMetadataTable(metaClient)) {
+ return compactUsingFileGroupReader(
+ logFiles, readerSchema, instantTime, internalSchemaOption, config, operation, metaClient, broadcastManagerOpt, taskContextSupplier);
+ } else {
+ return compactUsingLegacyMethod(
+ storage, logFiles, readerSchema, executionHelper, instantTime, maxInstantTime, instantRange, internalSchemaOption, config,
+ maxMemoryPerCompaction, operation, metaClient, compactionHandler);
+ }
+ }
+
+ private List<WriteStatus> compactUsingFileGroupReader(List<String> logFiles, Schema readerSchema, String instantTime,
+ Option<InternalSchema> internalSchemaOption, HoodieWriteConfig config,
+ CompactionOperation operation, HoodieTableMetaClient metaClient,
+ Option<CompactorBroadcastManager> partialUpdateReaderContextOpt,
+ TaskContextSupplier taskContextSupplier) throws IOException {
+ CompactorBroadcastManager compactorBroadcastManager = partialUpdateReaderContextOpt.get();
+ // PATH 1: When the engine decides to return a valid reader context object.
Review Comment:
Done.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864747430
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -278,4 +368,123 @@ public CompactionExecutionHelper getCompactionExecutionStrategy(HoodieCompaction
}
}
+ List<WriteStatus> compactWithFileGroupReader(HoodieReaderContext readerContext,
+ String instantTime,
+ HoodieTableMetaClient metaClient,
+ CompactionOperation operation,
+ List<HoodieLogFile> logFiles,
+ Schema readerSchema,
+ Option<InternalSchema> internalSchemaOpt,
+ TypedProperties props,
+ HoodieWriteConfig config,
+ TaskContextSupplier taskContextSupplier) throws IOException {
+ HoodieTimer timer = HoodieTimer.start();
+ timer.startTimer();
+ Option<HoodieBaseFile> baseFileOpt =
+ operation.getBaseFile(metaClient.getBasePath().toString(), operation.getPartitionPath());
+ FileSlice fileSlice = new FileSlice(
+ operation.getFileGroupId(),
+ operation.getBaseInstantTime(),
+ baseFileOpt.isPresent() ? baseFileOpt.get() : null,
+ logFiles);
+
+ // 1. Generate the input for fg reader.
+ boolean usePosition = config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
+ readerContext,
+ metaClient.getStorage(),
+ metaClient.getBasePath().toString(),
+ instantTime,
+ fileSlice,
+ readerSchema,
+ readerSchema,
+ internalSchemaOpt,
+ metaClient,
+ props,
+ 0,
+ Long.MAX_VALUE,
+ usePosition);
+
+ // 2. Get the `HoodieFileGroupReaderIterator` from the fg reader.
+ fileGroupReader.initRecordIterators();
+ HoodieFileGroupReader.HoodieFileGroupReaderIterator<T> recordIterator
+ = fileGroupReader.getClosableIterator();
+
+ // 3. Write the record using parquet writer.
+ String writeToken = FSUtils.makeWriteToken(
Review Comment:
CC: @yihua
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509594736
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
}, {
"hash" : "1ab58804320d995919865e16fb29f909f898a062",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ab58804320d995919865e16fb29f909f898a062",
"triggerType" : "PUSH"
}, {
"hash" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5ffc8c5ad207e7b8f293d15753bb78d485742a8a Azure: [CANCELED](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143)
* 1ab58804320d995919865e16fb29f909f898a062 UNKNOWN
* d9602ea0bc5dc6a2f9b91594cd1937d39803650c 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509590853
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138)
* 5ffc8c5ad207e7b8f293d15753bb78d485742a8a 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509504987
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864716954
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -189,26 +218,58 @@ public List<WriteStatus> compact(HoodieCompactionHandler compactionHandler,
new StoragePath(FSUtils.constructAbsolutePath(
metaClient.getBasePath(), operation.getPartitionPath()), p).toString())
.collect(toList());
- HoodieMergedLogRecordScanner scanner = HoodieMergedLogRecordScanner.newBuilder()
- .withStorage(storage)
- .withBasePath(metaClient.getBasePath())
- .withLogFilePaths(logFiles)
- .withReaderSchema(readerSchema)
- .withLatestInstantTime(executionHelper.instantTimeToUseForScanning(instantTime, maxInstantTime))
- .withInstantRange(instantRange)
- .withInternalSchema(internalSchemaOption.orElse(InternalSchema.getEmptyInternalSchema()))
- .withMaxMemorySizeInBytes(maxMemoryPerCompaction)
- .withReverseReader(config.getCompactionReverseLogReadEnabled())
- .withBufferSize(config.getMaxDFSStreamBufferSize())
- .withSpillableMapBasePath(config.getSpillableMapBasePath())
- .withDiskMapType(config.getCommonConfig().getSpillableDiskMapType())
- .withBitCaskDiskMapCompressionEnabled(config.getCommonConfig().isBitCaskDiskMapCompressionEnabled())
- .withOperationField(config.allowOperationMetadataField())
- .withPartition(operation.getPartitionPath())
- .withOptimizedLogBlocksScan(executionHelper.enableOptimizedLogBlockScan(config))
- .withRecordMerger(config.getRecordMerger())
- .withTableMetaClient(metaClient)
+
+ boolean isPartialUpdateEnabled = config.getBooleanOrDefault("hoodie.spark.sql.merge.into.partial.updates", false);
+ if (isPartialUpdateEnabled && broadcastManagerOpt.isPresent() && !isMetadataTable(metaClient)) {
+ return compactUsingFileGroupReader(
+ logFiles, readerSchema, instantTime, internalSchemaOption, config, operation, metaClient, broadcastManagerOpt, taskContextSupplier);
+ } else {
+ return compactUsingLegacyMethod(
+ storage, logFiles, readerSchema, executionHelper, instantTime, maxInstantTime, instantRange, internalSchemaOption, config,
+ maxMemoryPerCompaction, operation, metaClient, compactionHandler);
+ }
+ }
+
+ private List<WriteStatus> compactUsingFileGroupReader(List<String> logFiles, Schema readerSchema, String instantTime,
+ Option<InternalSchema> internalSchemaOption, HoodieWriteConfig config,
+ CompactionOperation operation, HoodieTableMetaClient metaClient,
+ Option<CompactorBroadcastManager> partialUpdateReaderContextOpt,
+ TaskContextSupplier taskContextSupplier) throws IOException {
+ CompactorBroadcastManager compactorBroadcastManager = partialUpdateReaderContextOpt.get();
+ // PATH 1: When the engine decides to return a valid reader context object.
Review Comment:
can you remove this comment. or fix it as needed
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java:
##########
@@ -112,12 +112,12 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>> upsert(HoodieEngineContext c
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> insert(HoodieEngineContext context, String instantTime, HoodieData<HoodieRecord<T>> records) {
- return new SparkInsertCommitActionExecutor<>((HoodieSparkEngineContext)context, config, this, instantTime, records).execute();
+ return new SparkInsertCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, records).execute();
}
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> bulkInsert(HoodieEngineContext context, String instantTime, HoodieData<HoodieRecord<T>> records,
- Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner) {
+ Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner) {
Review Comment:
can you revert changes in these files if these are just unintentional formatting changes.
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java:
##########
@@ -122,19 +122,19 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>> deletePrepped(HoodieEngineCo
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> upsertPrepped(HoodieEngineContext context, String instantTime,
- HoodieData<HoodieRecord<T>> preppedRecords) {
+ HoodieData<HoodieRecord<T>> preppedRecords) {
return new SparkUpsertPreppedDeltaCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, preppedRecords).execute();
}
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>> insertPrepped(HoodieEngineContext context, String instantTime,
- HoodieData<HoodieRecord<T>> preppedRecords) {
+ HoodieData<HoodieRecord<T>> preppedRecords) {
Review Comment:
can you revert changes in these files if these are just unintentional formatting changes.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -278,4 +368,123 @@ public CompactionExecutionHelper getCompactionExecutionStrategy(HoodieCompaction
}
}
+ List<WriteStatus> compactWithFileGroupReader(HoodieReaderContext readerContext,
+ String instantTime,
+ HoodieTableMetaClient metaClient,
+ CompactionOperation operation,
+ List<HoodieLogFile> logFiles,
+ Schema readerSchema,
+ Option<InternalSchema> internalSchemaOpt,
+ TypedProperties props,
+ HoodieWriteConfig config,
+ TaskContextSupplier taskContextSupplier) throws IOException {
+ HoodieTimer timer = HoodieTimer.start();
+ timer.startTimer();
+ Option<HoodieBaseFile> baseFileOpt =
+ operation.getBaseFile(metaClient.getBasePath().toString(), operation.getPartitionPath());
+ FileSlice fileSlice = new FileSlice(
+ operation.getFileGroupId(),
+ operation.getBaseInstantTime(),
+ baseFileOpt.isPresent() ? baseFileOpt.get() : null,
+ logFiles);
+
+ // 1. Generate the input for fg reader.
+ boolean usePosition = config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
+ readerContext,
+ metaClient.getStorage(),
+ metaClient.getBasePath().toString(),
+ instantTime,
+ fileSlice,
+ readerSchema,
+ readerSchema,
+ internalSchemaOpt,
+ metaClient,
+ props,
+ 0,
+ Long.MAX_VALUE,
+ usePosition);
+
+ // 2. Get the `HoodieFileGroupReaderIterator` from the fg reader.
+ fileGroupReader.initRecordIterators();
+ HoodieFileGroupReader.HoodieFileGroupReaderIterator<T> recordIterator
+ = fileGroupReader.getClosableIterator();
+
+ // 3. Write the record using parquet writer.
+ String writeToken = FSUtils.makeWriteToken(
Review Comment:
hey @yihua : Should we use HoodieCreateHandle here.
I am worried about adhoc code written here. what incase we miss out something which regular MergeHandle should do?
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java:
##########
@@ -101,6 +123,25 @@ public HoodieRecord<InternalRow> constructHoodieRecord(Option<InternalRow> rowOp
return new HoodieSparkRecord(row, HoodieInternalRowUtils.getCachedSchema(schema));
}
+ @Override
+ public HoodieRecord<InternalRow> constructHoodieRecord(Option<InternalRow> rowOption,
+ Map<String, Object> metadataMap,
+ TypedProperties props) {
+ if (!rowOption.isPresent()) {
Review Comment:
again, can you help me understand why do we need these adhoc changes ?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -130,8 +151,14 @@ public HoodieData<WriteStatus> compact(
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
// if this is a MDT, set up the instant range of log reader just like regular MDT snapshot reader.
Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient);
+
+ // Broadcast necessary information.
+ Option<CompactorBroadcastManager> broadcastManagerOpt = getCompactorBroadcastManager(context);
+ if (broadcastManagerOpt.isPresent()) {
+ broadcastManagerOpt.get().prepareAndBroadcast();
Review Comment:
see. why we are broadcasting if there is no partial updates?
can we avoid that
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -130,8 +151,14 @@ public HoodieData<WriteStatus> compact(
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
// if this is a MDT, set up the instant range of log reader just like regular MDT snapshot reader.
Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient);
+
+ // Broadcast necessary information.
+ Option<CompactorBroadcastManager> broadcastManagerOpt = getCompactorBroadcastManager(context);
Review Comment:
I would say, we should not even instantiate CompactorBroadcastManager if partial update is not enabled.
also, broadcastManager naming does not sit well. for eg, why would this be a required entity for non partial update flows.
So, can we rename it to be little generic.
something like PartialUpdateReaderContext or PartialUpdateContext.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -278,4 +368,123 @@ public CompactionExecutionHelper getCompactionExecutionStrategy(HoodieCompaction
}
}
+ List<WriteStatus> compactWithFileGroupReader(HoodieReaderContext readerContext,
+ String instantTime,
+ HoodieTableMetaClient metaClient,
+ CompactionOperation operation,
+ List<HoodieLogFile> logFiles,
+ Schema readerSchema,
+ Option<InternalSchema> internalSchemaOpt,
+ TypedProperties props,
+ HoodieWriteConfig config,
+ TaskContextSupplier taskContextSupplier) throws IOException {
+ HoodieTimer timer = HoodieTimer.start();
+ timer.startTimer();
+ Option<HoodieBaseFile> baseFileOpt =
+ operation.getBaseFile(metaClient.getBasePath().toString(), operation.getPartitionPath());
+ FileSlice fileSlice = new FileSlice(
+ operation.getFileGroupId(),
+ operation.getBaseInstantTime(),
+ baseFileOpt.isPresent() ? baseFileOpt.get() : null,
+ logFiles);
+
+ // 1. Generate the input for fg reader.
+ boolean usePosition = config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
+ readerContext,
+ metaClient.getStorage(),
+ metaClient.getBasePath().toString(),
+ instantTime,
+ fileSlice,
+ readerSchema,
+ readerSchema,
+ internalSchemaOpt,
+ metaClient,
+ props,
+ 0,
+ Long.MAX_VALUE,
+ usePosition);
+
+ // 2. Get the `HoodieFileGroupReaderIterator` from the fg reader.
+ fileGroupReader.initRecordIterators();
+ HoodieFileGroupReader.HoodieFileGroupReaderIterator<T> recordIterator
+ = fileGroupReader.getClosableIterator();
+
+ // 3. Write the record using parquet writer.
+ String writeToken = FSUtils.makeWriteToken(
Review Comment:
see. we are not creating markers here. thats the reason I told I was not comfortable making this default for whole of spark.
unless we go through one of the existing write handles, the new custom code we write might have some gaps.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java:
##########
@@ -83,7 +86,26 @@ public Object getValue(InternalRow row, Schema schema, String fieldName) {
@Override
public String getRecordKey(InternalRow row, Schema schema) {
- return getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD).toString();
+ Object key = getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD);
+ if (key != null) {
+ return key.toString();
+ }
+ return null;
+ }
+
+ @Override
+ public String getRecordKey(InternalRow row, Schema schema, TypedProperties props) {
+ String key = getRecordKey(row, schema);
Review Comment:
can you help me understand why do we need these changes?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieUnMergedLogRecordScanner.java:
##########
@@ -19,6 +19,7 @@
package org.apache.hudi.common.table.log;
import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieKey;
Review Comment:
then, can we do the rebase?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -278,4 +368,123 @@ public CompactionExecutionHelper getCompactionExecutionStrategy(HoodieCompaction
}
}
+ List<WriteStatus> compactWithFileGroupReader(HoodieReaderContext readerContext,
+ String instantTime,
+ HoodieTableMetaClient metaClient,
+ CompactionOperation operation,
+ List<HoodieLogFile> logFiles,
+ Schema readerSchema,
+ Option<InternalSchema> internalSchemaOpt,
+ TypedProperties props,
+ HoodieWriteConfig config,
+ TaskContextSupplier taskContextSupplier) throws IOException {
+ HoodieTimer timer = HoodieTimer.start();
+ timer.startTimer();
+ Option<HoodieBaseFile> baseFileOpt =
+ operation.getBaseFile(metaClient.getBasePath().toString(), operation.getPartitionPath());
+ FileSlice fileSlice = new FileSlice(
+ operation.getFileGroupId(),
+ operation.getBaseInstantTime(),
+ baseFileOpt.isPresent() ? baseFileOpt.get() : null,
+ logFiles);
+
+ // 1. Generate the input for fg reader.
+ boolean usePosition = config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
+ readerContext,
+ metaClient.getStorage(),
+ metaClient.getBasePath().toString(),
+ instantTime,
+ fileSlice,
+ readerSchema,
+ readerSchema,
+ internalSchemaOpt,
+ metaClient,
+ props,
+ 0,
+ Long.MAX_VALUE,
+ usePosition);
+
+ // 2. Get the `HoodieFileGroupReaderIterator` from the fg reader.
+ fileGroupReader.initRecordIterators();
+ HoodieFileGroupReader.HoodieFileGroupReaderIterator<T> recordIterator
+ = fileGroupReader.getClosableIterator();
+
+ // 3. Write the record using parquet writer.
+ String writeToken = FSUtils.makeWriteToken(
Review Comment:
For eg, does this handle virtual keys?
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509594019
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
}, {
"hash" : "1ab58804320d995919865e16fb29f909f898a062",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ab58804320d995919865e16fb29f909f898a062",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138)
* 5ffc8c5ad207e7b8f293d15753bb78d485742a8a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143)
* 1ab58804320d995919865e16fb29f909f898a062 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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509591250
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138)
* 5ffc8c5ad207e7b8f293d15753bb78d485742a8a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143)
<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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864748380
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala:
##########
@@ -582,7 +582,7 @@ object DataSourceWriteOptions {
val ENABLE_MERGE_INTO_PARTIAL_UPDATES: ConfigProperty[Boolean] = ConfigProperty
.key("hoodie.spark.sql.merge.into.partial.updates")
- .defaultValue(true)
+ .defaultValue(false)
Review Comment:
I think @yihua and @nsivabalan want to use the config to control the compaction code path based on partial updates. Therefore, logically the default value should false; otherwise, the flag is meaningless.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864748608
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java:
##########
@@ -101,6 +123,25 @@ public HoodieRecord<InternalRow> constructHoodieRecord(Option<InternalRow> rowOp
return new HoodieSparkRecord(row, HoodieInternalRowUtils.getCachedSchema(schema));
}
+ @Override
+ public HoodieRecord<InternalRow> constructHoodieRecord(Option<InternalRow> rowOption,
+ Map<String, Object> metadataMap,
+ TypedProperties props) {
+ if (!rowOption.isPresent()) {
Review Comment:
This is also related to the record key issue.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509523878
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138)
<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-8522] Compaction using fg reader for partial update [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864701463
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala:
##########
@@ -582,7 +582,7 @@ object DataSourceWriteOptions {
val ENABLE_MERGE_INTO_PARTIAL_UPDATES: ConfigProperty[Boolean] = ConfigProperty
.key("hoodie.spark.sql.merge.into.partial.updates")
- .defaultValue(true)
+ .defaultValue(false)
Review Comment:
why disable the partial update, is there any known bug?
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864723229
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java:
##########
@@ -130,8 +151,14 @@ public HoodieData<WriteStatus> compact(
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
// if this is a MDT, set up the instant range of log reader just like regular MDT snapshot reader.
Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient);
+
+ // Broadcast necessary information.
+ Option<CompactorBroadcastManager> broadcastManagerOpt = getCompactorBroadcastManager(context);
Review Comment:
All it does is to enable fg reader, and partial update is just a part of it. So emphasizing partial update is too specific.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864750837
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java:
##########
@@ -83,7 +86,26 @@ public Object getValue(InternalRow row, Schema schema, String fieldName) {
@Override
public String getRecordKey(InternalRow row, Schema schema) {
- return getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD).toString();
+ Object key = getFieldValueFromInternalRow(row, schema, RECORD_KEY_METADATA_FIELD);
+ if (key != null) {
+ return key.toString();
+ }
+ return null;
+ }
+
+ @Override
+ public String getRecordKey(InternalRow row, Schema schema, TypedProperties props) {
+ String key = getRecordKey(row, schema);
Review Comment:
In my experiments, the schema may contain metadata fields, but there is no such columns in the data. Therefore, these metadata fields are NULL after retrieved from the disk. When we tried to get the record key, only the `_record_key` is used, which gives us NULL, and cause the merge failure. To avoid such problem, we should try to find the key field; there are different ways to find out the key field, either from the table config or from the write config.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #12384:
URL: https://github.com/apache/hudi/pull/12384#discussion_r1864750881
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieUnMergedLogRecordScanner.java:
##########
@@ -19,6 +19,7 @@
package org.apache.hudi.common.table.log;
import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieKey;
Review Comment:
yeah. Will do.
--
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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509635091
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
}, {
"hash" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2143",
"triggerID" : "5ffc8c5ad207e7b8f293d15753bb78d485742a8a",
"triggerType" : "PUSH"
}, {
"hash" : "1ab58804320d995919865e16fb29f909f898a062",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ab58804320d995919865e16fb29f909f898a062",
"triggerType" : "PUSH"
}, {
"hash" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d9602ea0bc5dc6a2f9b91594cd1937d39803650c",
"triggerType" : "PUSH"
}, {
"hash" : "c2ea5b14785a2d0845a76bd16ed6dbcc42744af5",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2144",
"triggerID" : "c2ea5b14785a2d0845a76bd16ed6dbcc42744af5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1ab58804320d995919865e16fb29f909f898a062 UNKNOWN
* d9602ea0bc5dc6a2f9b91594cd1937d39803650c UNKNOWN
* c2ea5b14785a2d0845a76bd16ed6dbcc42744af5 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2144)
<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-8522] Compaction using fg reader for partial update [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #12384:
URL: https://github.com/apache/hudi/pull/12384#issuecomment-2509505360
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138",
"triggerID" : "b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7bff1595b3eb3a426cff7013d6dc2bd3b1ade98 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=2138)
<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