nsivabalan commented on PR #18076:
URL: https://github.com/apache/hudi/pull/18076#issuecomment-4827413119
@lokeshj1703 — took the liberty (maintainer-can-modify is on) of pushing a
follow-up commit to this branch addressing the BLOCKING + IMPORTANT items from
my review. Happy to revert / split if you'd prefer a different shape.
Commit: c9052af32102 — \`fix: Apply files-per-sync limit deterministically
via window predicate\`
### What changed and why
**1. Files-limit now rides the existing ordered window (addresses BLOCKING
#1)**
Previously: \`metadataRows =
checkPointAndDataset.getRight().get().limit((int) numFilesLimit)\` followed by
\`getCheckpointFromLastRow(metadataRows, ...)\`.
The dataset returned by \`filterAndGenerateCheckpointBasedOnSourceLimit\` is
not explicitly ordered immediately before \`.limit()\`, so on a multi-partition
input Spark's \`GlobalLimit\` can pick an arbitrary subset. Recalculating the
checkpoint from the max of that subset then points past skipped files → silent
data loss on the next sync.
Fix: in \`IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit\`
add a \`row_number()\` column alongside the existing \`cumulativeSize\` window,
and filter on both predicates in a single window pass. The existing \`row =
collectedRows.orderBy(...desc).first()\` then produces the correct checkpoint
with no second pass.
\`\`\`java
WindowSpec windowSpec = Window.orderBy(col(orderColumn), col(keyColumn));
Dataset<Row> aggregatedData = orderedDf
.withColumn(CUMULATIVE_COLUMN_NAME,
sum(col(limitColumn)).over(windowSpec))
.withColumn(CUMULATIVE_COUNT_COLUMN_NAME, row_number().over(windowSpec));
Dataset<Row> collectedRows = aggregatedData
.filter(col(CUMULATIVE_COLUMN_NAME).leq(sourceLimit)
.and(col(CUMULATIVE_COUNT_COLUMN_NAME).leq(numFilesLimit)));
\`\`\`
I kept the existing 4-arg \`filterAndGenerateCheckpointBasedOnSourceLimit\`
signature as an overload that delegates with \`numFilesLimit =
Long.MAX_VALUE\`, so no other callers (\`TestIncrSourceHelper\` has 9) need to
change.
**2. Removed the duplicate persist/unpersist block in \`CloudDataFetcher\`
(addresses BLOCKING #2)**
With #1 the post-hoc \`metadataRows.persist()\` block is no longer needed —
the helper already manages persist/unpersist on \`sourceData\`. No more leak
surface on exception.
**3. Removed the \`cloudObjectMetadata.size() >= numFilesLimit\` recalc
heuristic (addresses IMPORTANT #3)**
\`getObjectMetadata\` dedupes via \`.distinct()\`, so this comparison was
unreliable when duplicate events existed. With the limit applied in the window,
the recalc branch isn't needed at all.
**4. Removed \`IncrSourceHelper.getCheckpointFromLastRow\`**
No longer called anywhere.
**5. Reverted unrelated \`isNullOrEmpty → .isEmpty()\` cleanup in
\`CloudObjectsSelectorCommon\` (NIT #8)**
Kept the PR scoped to the fix.
**6. Tests added in \`TestIncrSourceHelper\`** (addresses IMPORTANT #4):
- \`testFilesLimitContiguousAcrossManyPartitions\` — 50 files spread across
8 Spark partitions, iterate 5 syncs with \`numFilesLimit = 10\`, assert all 50
files come out in strict insertion order with no gaps. This is the test that
would have caught the \`.limit()\` non-determinism on a real cluster.
- \`testFilesLimitCrossesCommitBoundary\` — files-limit truncates
mid-commit; assert next sync resumes from the next file in the same commit, no
skips or re-reads.
- \`testFilesLimitBindingOverByteLimit\` — files-limit is binding while
byte-limit is slack.
- \`testFilesLimitLargerThanAvailable\` — files-limit > dataset; byte-path
unaffected (guards the recalc-heuristic removal).
The existing \`testFilesLimitCheckpointConsistency\` in
\`TestS3EventsHoodieIncrSource\` still passes unchanged.
### Test runs (local)
- \`TestIncrSourceHelper\`: 11 / 11 passing (7 pre-existing + 4 new).
- \`TestS3EventsHoodieIncrSource\`: 16 / 16 passing.
### Not addressed in this commit (leaving for you / discussion)
- **IMPORTANT #5 (GCS mirror test)** — \`TestGcsEventsHoodieIncrSource\`
mirror of the S3 files-limit test. The behavior is covered at the helper layer
by the new tests, but a GCS-path integration test would still be valuable. Let
me know if you want me to add it.
- **SUGGESTION #6 (default limit lowering)** — left at 10M as you had it.
Reasonable people will disagree; happy to push a smaller default if you'd like.
- **SUGGESTION #7 (log-message ordering)** — folded a single \`log.info\`
showing both \`sourceLimit\` and \`numFilesLimit\` up-front, which seemed
sufficient.
Diff stats: 4 files, +175 / -50.
Net effect on the PR: same user-facing config + behavior, deterministic
across executor partitions, simpler control flow, no leak surface. Let me know
if you want any of this reshaped.
--
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]
