hudi-agent commented on code in PR #20224:
URL: https://github.com/apache/hudi/pull/20224#discussion_r4235355044
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableGlobalRecordLevelIndex.java:
##########
@@ -73,6 +73,7 @@ public
SparkMetadataTableGlobalRecordLevelIndex(HoodieWriteConfig config) {
@Override
public <R> HoodieData<HoodieRecord<R>>
tagLocation(HoodieData<HoodieRecord<R>> records, HoodieEngineContext context,
HoodieTable hoodieTable) throws HoodieIndexException {
+ context.setJobStatus(this.getClass().getSimpleName(), "Record index
lookup: " + hoodieTable.getConfig().getTableName());
Review Comment:
🤖 non-blocking: `tagLocation` here is fully lazy, and
`prepareWorkloadProfile` sets `Building workload profile:` before its
`countByKey` submits the first job, so I'd expect the lookup stages to still
show under workload profiling. Did an event log taken after this change show
`Record index lookup:` on them?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseWriteHelper.java:
##########
@@ -64,6 +64,9 @@ public HoodieWriteMetadata<O> write(String instantTime,
try {
HoodieTimer sourceReadAndIndexTimer = HoodieTimer.start();
// De-dupe/merge if needed
+ if (shouldCombine) {
+ context.setJobStatus(this.getClass().getSimpleName(), "Deduplicating
records: " + table.getConfig().getTableName());
Review Comment:
🤖 non-blocking: `deduplicateRecords` is a lazy `reduceByKey`, so on Spark
`Tagging:` (UPSERT/DELETE) or `Building workload profile:` (INSERT) replaces
this before any job is submitted, and I think the dedup shuffle will still be
attributed to tagging. Was that what you saw in the event log after the change?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkStreamingMetadataWriteHandler.java:
##########
@@ -39,6 +39,8 @@ public class SparkStreamingMetadataWriteHandler extends
StreamingMetadataWriteHa
@Override
public HoodieData<WriteStatus> streamWriteToMetadataTable(HoodieTable table,
HoodieData<WriteStatus> dataTableWriteStatuses, String instantTime,
int
coalesceDivisorForDataTableWrites) {
+ table.getContext().setJobStatus(this.getClass().getSimpleName(),
Review Comment:
🤖 non-blocking: I think this label misses the job it targets in both setups:
- With RLI/SI,
`SparkHoodieBackedTableMetadataWriter#streamWriteToMetadataTable` overwrites
it, and the MDT `BaseSparkCommitActionExecutor#execute` then calls
`clearJobStatus()`. So the `SlimWriteStats.from` collect in
`SparkRDDWriteClient#commit` still runs with no label.
- Without RLI/SI (the Spark default), the label survives to that collect, so
the whole data-table write that used to show as `Committing stats:` now shows
as a metadata-table write.
Could you set the description after
`metadataWriter.streamWriteToMetadataPartitions(...)` returns, with a name that
covers both writes, e.g. `"Writing data table and streaming to metadata table:
"`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java:
##########
@@ -710,6 +710,7 @@ Pair<InputBatch, Boolean>
fetchNextBatchFromSource(Option<Checkpoint> resumeChec
InputBatch inputBatchForWriter = null; // row writer
boolean reconcileSchema =
props.getBoolean(DataSourceWriteOptions.RECONCILE_SCHEMA().key());
if (transformer.isPresent()) {
+ hoodieSparkContext.setJobStatus(this.getClass().getSimpleName(),
"Transforming source batch: " + cfg.targetTableName);
Review Comment:
🤖 non-blocking: since this runs before
`formatAdapter.fetchNewDataInRowFormat(...)`, jobs the source runs eagerly
during fetch (e.g. the `isEmpty()`/`first()` calls in `IncrSourceHelper` used
by the S3/GCS events sources) move from `Fetching next batch:` to `Transforming
source batch:`. Could you move it below the fetch, just before
`transformer.get().apply(...)`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java:
##########
@@ -1099,6 +1100,7 @@ private WriteClientWriteResult writeToSink(InputBatch
inputBatch, String instant
metaClient = HoodieTableMetaClient.reload(metaClient);
TypedProperties mergeProps = ConfigUtils.getMergeProps(props,
metaClient.getTableConfig());
HoodieRecordType recordType =
createRecordMerger(mergeProps).getRecordType();
+ hoodieSparkContext.setJobStatus(this.getClass().getSimpleName(),
"Creating Hudi records: " + cfg.targetTableName);
Review Comment:
🤖 non-blocking: `HoodieStreamerUtils.createHoodieRecords` only builds lazy
RDDs, so for upsert/insert the write client's own labels (`Deduplicating
records:`/`Tagging:`/`Building workload profile:`) replace this before the
first job. I think record creation will still be billed under those; only eager
paths like the `DELETE_PARTITION` `collect()` would pick up this label.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
