nsivabalan commented on PR #18076:
URL: https://github.com/apache/hudi/pull/18076#issuecomment-4921945018
Follow-ups from the review round — happy to leave these to you rather than
push onto the branch. Three items:
### 1. `sourceData.persist()` / `unpersist()` bracket only closes on the
happy path
`IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit`, lines 293
→ 347. The `persist` at 293 has one matching `unpersist` on the
empty-after-checkpoint-filter branch (307–309) and one on the main path (347),
but neither is in a `finally`. If `.first()` at 340 / 344 throws (executor
loss, driver interrupt, OOM), the cached blocks leak for the lifetime of the
SparkSession. Pre-existing, but worth tightening while the surrounding code is
being touched:
```java
sourceData.persist(StorageLevel.MEMORY_AND_DISK());
try {
// existing body from line 294 onward, minus the two unpersist() calls
return Pair.of(...);
} finally {
sourceData.unpersist();
}
```
### 2. Single-partition window collapse — worth an inline note (no fix)
`Window.orderBy(...)` with no `partitionBy` triggers Spark's
`SinglePartition` exchange (the `WindowExec: No Partition Defined for Window
operation!` WARN). Pre-existing — the bytes-limit path already required a total
ordering — and Catalyst folds the new `row_number()` into the same
`WindowExec`, so no extra shuffle. But this is the real scale ceiling for the
guard: the entire `filteredSourceData` (post checkpoint filter, pre limit) is
held on one executor for the window pass, which effectively caps how large
`numFilesLimit` can usefully go.
Not asking for a fix — just an inline comment on `windowSpec` at line 325
noting this so future readers don't try to raise the limit and get burned.
Something like:
```java
// Window is intentionally unpartitioned — running sum and row_number
require a total
// ordering, so Spark collapses this to a single-partition WindowExec. This
bounds how
// large numFilesLimit can usefully go; the whole filteredSourceData is held
on one
// executor here.
```
A follow-up JIRA to compute the prefix-sum via `mapPartitions` +
driver-merge (avoiding the collapse) would be worthwhile but out of scope for
this PR.
### 3. Default value: `10_000_000` → `10_000`
The whole point of the guard is bounding driver-side per-file metadata. 10M
is essentially "off" — a user has to opt in to get any protection. `10_000` is
a more defensive default: still generous for normal workloads, meaningfully
caps pathological small-file bursts, and users with genuinely small files can
raise it. Two-character change in `CloudSourceConfig.SOURCE_MAX_FILES_PER_SYNC`.
Thanks!
--
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]
