hudi-agent commented on code in PR #18076:
URL: https://github.com/apache/hudi/pull/18076#discussion_r3485073381
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudDataFetcher.java:
##########
@@ -111,10 +113,27 @@ public Pair<Option<Dataset<Row>>, Checkpoint>
fetchPartitionedSource(
}
log.info("Adjusted end checkpoint :{}", checkPointAndDataset.getLeft());
+ // Limit the number of files per sync to prevent driver OOM with many
small files.
+ // Apply the limit at the Dataset level so the checkpoint can be
recalculated accurately.
+ long numFilesLimit = props.getLong(SOURCE_MAX_FILES_PER_SYNC.key(),
SOURCE_MAX_FILES_PER_SYNC.defaultValue());
+ Dataset<Row> metadataRows = checkPointAndDataset.getRight().get()
+ .limit((int) Math.min(numFilesLimit, Integer.MAX_VALUE));
+ // Persist to avoid re-triggering the DAG for both getObjectMetadata and
checkpoint recalculation
+ metadataRows.persist(StorageLevel.MEMORY_AND_DISK());
+
+ CloudObjectIncrCheckpoint checkpoint = checkPointAndDataset.getLeft();
boolean checkIfFileExists = getBooleanWithAltKeys(props,
ENABLE_EXISTS_CHECK);
- List<CloudObjectMetadata> cloudObjectMetadata =
CloudObjectsSelectorCommon.getObjectMetadata(cloudType, sparkContext,
checkPointAndDataset.getRight().get(), checkIfFileExists, props);
+ List<CloudObjectMetadata> cloudObjectMetadata =
CloudObjectsSelectorCommon.getObjectMetadata(cloudType, sparkContext,
metadataRows, checkIfFileExists, props);
log.info("Total number of files to process :{}",
cloudObjectMetadata.size());
+ // If the files limit was reached, recalculate the checkpoint from the
actual last row
Review Comment:
🤖 This guard compares the post-processing count against the limit, but
`getObjectMetadata()` can return fewer records than `metadataRows` has — it
applies `.distinct()` and, when `ENABLE_EXISTS_CHECK` is on, drops non-existent
files. So if `.limit()` truncated the batch but dedup/existence-check then
reduces the count below `numFilesLimit`, this is false and the checkpoint is
left at the byte-limit position (e.g. file5) while only the first N files were
processed. The next sync would resume past the unprocessed files and skip them.
Could we instead base the recalculation on whether the limit actually truncated
the dataset (e.g. compare the row count of `metadataRows` to the original
`collectedRows` count) rather than on `cloudObjectMetadata.size()`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java:
##########
@@ -322,6 +322,23 @@ public static Pair<CloudObjectIncrCheckpoint,
Option<Dataset<Row>>> filterAndGen
return Pair.of(new CloudObjectIncrCheckpoint(row.getString(0),
row.getString(1)), Option.of(collectedRows));
}
+ /**
+ * Extract the checkpoint (commit_time, file_key) from the last row of the
given dataset,
+ * ordering by the order column and key column descending.
+ * Used to recalculate the checkpoint after a files-per-sync limit is
applied.
+ *
+ * @param dataset Dataset whose last row (by order+key) determines the
checkpoint
+ * @param queryInfo Query info with column names for ordering and key
extraction
+ * @return Checkpoint corresponding to the last file in the dataset
+ */
Review Comment:
🤖 nit: `getCheckpointFromLastRow` is a bit confusing because the
implementation sorts descending and calls `.first()` — a future reader will
wonder why `first()` retrieves the "last" row. Could you consider a name like
`getEndCheckpoint` or `getCheckpointAtMaxKey` that makes it clear this returns
the boundary checkpoint for the limited batch, without implying a particular
position in the dataset?
<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]
