nsivabalan commented on PR #18076:
URL: https://github.com/apache/hudi/pull/18076#issuecomment-4827339221
Thanks for tackling this @lokeshj1703 — the driver-OOM problem is real and
the config shape (advanced, sensible default, versioned, dot-separated) is well
done. Sharing a principal-level review below. One blocking correctness concern,
otherwise net-positive.
## Mergeability summary
- Goal is sound; problem statement in #18075 is legitimate.
- One **🚨 BLOCKING** correctness risk around `.limit(n)` on an unordered
Dataset — described below. Fixable with a small targeted change.
- No storage / public-API / config-semantics breakage. Additive config with
backward-compatible default.
- Test coverage is thin: one happy-path S3 test that only exercises a single
Spark partition. Edge cases around multi-partition determinism, GCS path,
commit boundaries, and limit=1 are missing.
- Size is small (84/-4 across 5 files). No need to split.
- No timeline/metadata/concurrency surfaces touched.
## 🚨 BLOCKING
### 1. `.limit(n)` on `collectedRows` is non-deterministic — risk of silent
file skipping / data loss
`CloudDataFetcher.java` (the new block after
`filterAndGenerateCheckpointBasedOnSourceLimit`):
\`\`\`java
Dataset<Row> metadataRows = checkPointAndDataset.getRight().get()
.limit((int) Math.min(numFilesLimit, Integer.MAX_VALUE));
...
if (cloudObjectMetadata.size() >= numFilesLimit) {
checkpoint = IncrSourceHelper.getCheckpointFromLastRow(metadataRows,
queryInfo);
}
\`\`\`
The dataset returned by \`filterAndGenerateCheckpointBasedOnSourceLimit\` is
\`collectedRows\`, produced via
\`aggregatedData.filter(col(CUMULATIVE_COLUMN_NAME).leq(sourceLimit))\` on top
of a window-\`orderBy\`. Spark's logical plan does **not** guarantee that a
subsequent \`.limit(n)\` will return the first \`n\` rows of the window order
once the dataset spans multiple partitions — \`GlobalLimit\` operates over
partitions whose internal ordering is no longer guaranteed after the
window+filter. There is no explicit \`orderBy\` immediately preceding the
\`.limit()\` call.
Concrete failure mode:
1. \`collectedRows\` is ordered as {f1, f2, f3, f4, f5}.
2. \`.limit(3)\` returns an arbitrary subset, say {f1, f3, f5}.
3. \`getObjectMetadata(...)\` materializes {f1, f3, f5} → these are the
files actually ingested.
4. \`getCheckpointFromLastRow\` reorders this subset desc and picks f5 as
the new checkpoint.
5. Next sync starts strictly *after* f5 → **f2 and f4 are never processed**.
Silent data loss in an append-only ingest pipeline.
The new unit test passes only because in a single-node test setup with one
partition \`.limit()\` happens to be deterministic. This will not hold on real
clusters with multiple partitions or skew.
**Recommended fix**: push the files-limit into
\`IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit\`, alongside
the existing byte-based cumulative computation, using \`row_number()\` over the
same window:
\`\`\`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, row_number().over(windowSpec));
Dataset<Row> collectedRows = aggregatedData.filter(
col(CUMULATIVE_COLUMN_NAME).leq(sourceLimit)
.and(col(CUMULATIVE_COUNT_COLUMN).leq(filesLimit)));
\`\`\`
The existing \`row = collectedRows.select(...).orderBy(...desc).first()\` at
the end of \`filterAndGenerateCheckpointBasedOnSourceLimit\` then produces the
correct checkpoint with no second pass. This also:
- piggybacks on the existing sorted-window infra (returns a contiguous
prefix),
- removes the second \`orderBy().first()\` Spark job in
\`getCheckpointFromLastRow\`,
- removes the duplicate \`persist\`/\`unpersist\` block in
\`CloudDataFetcher\`,
- removes the need for the \`cloudObjectMetadata.size() >= numFilesLimit\`
heuristic (see ⚠️ #3),
- makes \`getCheckpointFromLastRow\` unnecessary.
### 2. Resource leak: persisted dataset not unpersisted on exception
In the new block in \`CloudDataFetcher\`:
\`\`\`java
metadataRows.persist(StorageLevel.MEMORY_AND_DISK());
...
List<CloudObjectMetadata> cloudObjectMetadata =
CloudObjectsSelectorCommon.getObjectMetadata(...);
...
metadataRows.unpersist();
\`\`\`
If \`getObjectMetadata\` or \`getCheckpointFromLastRow\` throws (S3 list
failure, executor OOM, schema mismatch), the cached block leaks for the
lifetime of the Spark application. With the Streamer running in a long-lived
loop this accumulates across syncs. Wrap in try/finally — or better, fold into
the existing persist/unpersist scope inside
\`filterAndGenerateCheckpointBasedOnSourceLimit\` (the fix in #1 does this for
free).
## ⚠️ IMPORTANT
### 3. Size-based recalculation condition can miss the actual files-limit
case
\`cloudObjectMetadata.size() >= numFilesLimit\` — \`cloudObjectMetadata\`
comes from \`.select(...).distinct().mapPartitions(...)\` inside
\`getObjectMetadata\`, i.e. **deduplicated** by (bucket, key, size). If
\`metadataRows\` contained duplicate rows (rare but possible if the same S3
event was emitted twice), \`size()\` < \`numFilesLimit\` even though
\`.limit(n)\` truncated the dataset → checkpoint is *not* recalculated → next
sync skips the truncated tail. The fix in #1 removes this branch entirely. If
keeping the current structure, compare against the row count of
\`metadataRows\` (after limit), not the deduped size.
### 4. New test doesn't cover the multi-partition non-determinism
\`testFilesLimitCheckpointConsistency\` runs on 5 rows in a single partition
where \`.limit()\` is incidentally deterministic. Add a test that:
- generates ≥ 50 rows,
- repartitions the source dataset into ≥ 4 partitions,
- sets \`SOURCE_MAX_FILES_PER_SYNC = 10\`,
- asserts that the next sync resumes *immediately* after the last processed
file, and that (batch1 ∪ batch2) equals the full ordered list with no gaps.
This is the test that would have caught #1.
### 5. Missing GCS coverage
The fix lives in \`CloudDataFetcher.fetchPartitionedSource\`, used by both
\`S3EventsHoodieIncrSource\` and \`GcsEventsHoodieIncrSource\`. Only an S3 test
was added. Add a mirror test in \`TestGcsEventsHoodieIncrSource\` so we don't
regress the GCS path silently.
## 💬 SUGGESTIONS
### 6. Default of 10M files may be too lenient as a guardrail
The motivating problem is driver OOM from per-file metadata held in the
driver. 10M \`CloudObjectMetadata\` objects (path string + size) at ~200 bytes
each is ~2 GB on the driver — already enough to OOM most default driver heaps.
Consider defaulting to 1M or 500K and letting power users tune upward.
Non-blocking; author's call.
### 7. Log message ordering
\`log.info(\"Total number of files to process :{}\", ...)\` fires before any
files-limit recalculation. Consider logging the *final* checkpoint and count
together after the recalculation block so operators see a single coherent line.
## 💅 NITS
### 8. Unrelated cleanup mixed in
The \`isNullOrEmpty → .isEmpty()\` swap in
\`CloudObjectsSelectorCommon.loadAsDataset\` is unrelated to the files-limit.
Fine as-is but ideally split.
### 9. Javadoc on \`getCheckpointFromLastRow\`
Worth clarifying the dataset is expected to already be ordered by the same
columns. Becomes moot if #1 is applied — the method is no longer needed.
## Suggested additional tests
- **Unit**: \`testFilesLimit_singleFileExceedsBytesLimit\` — first file is
larger than \`SOURCE_MAX_BYTES_PER_PARTITION\` AND \`SOURCE_MAX_FILES_PER_SYNC
= 1\`; assert exactly one file processed and checkpoint points to it.
- **Unit**: \`testFilesLimit_one\` — \`SOURCE_MAX_FILES_PER_SYNC = 1\`
across 5 syncs; assert checkpoint advances exactly one file per sync with no
duplicates/gaps.
- **Unit**: \`testFilesLimit_crossesCommitBoundary\` — 3 files in c1, 3 in
c2; \`SOURCE_MAX_FILES_PER_SYNC = 4\`; assert first checkpoint is
\`c2#<2nd-file-of-c2>\` and second sync picks up the remaining file.
- **Unit**: \`testFilesLimit_largerThanAvailable\` — limit = 1000, only 5
files; assert no recalculation path runs and checkpoint matches the byte-based
one.
- **Functional**: GCS mirror of \`testFilesLimitCheckpointConsistency\`.
- **Functional**: end-to-end Streamer with files-limit set, multi-batch
progression, asserting `sum(ingested files) == produced files` with no
duplicates.
--
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]
