skywalker0618 opened a new issue, #20338: URL: https://github.com/apache/hudi/issues/20338
### Task Description **What needs to be done:** On each discovery pass, the Flink streaming reader (`IncrementalInputSplits` → `WriteProfiles.getFilesFromMetadata`) checks that every file written by the new commits still exists, because the cleaner may have removed some. It makes one `StreamerUtil.fileExists` call per file, serially, on the JobManager's split enumerator thread. Proposal: on a high level, group files by partitions, like group all files under 2026/10/14, and then do a single LIST call to this partition to get all files under this partition. This avoids having to issue 1 request per file. More details: - Group candidate files by parent directory. When two or more candidates share a directory, list it once and check membership against the listing. Directories with a single candidate keep a single `exists` call. - Process directories in parallel with a bounded pool, set by a new option such as `read.file-existence-check.parallelism` (default 8). - Keep the reverse-commit-order resolution, so a file touched by several commits resolves to its latest surviving version. Alternatives considered: - **Parallel per-file `exists` with a large pool (for example 100 threads).** Wall time is similar (~36 s estimated for the table below), but it still sends one request per file. That's ~197K requests per pass, about 5.5K/s at 100 threads, which risks object-store throttling and it is capped by filesystem connection-pool limits. Grouping by directory sends ~49× fewer requests with a small pool. - **Answering existence from the metadata table's `files` partition.** No storage calls at all, but it only works for tables with the metadata table enabled. The listing approach works for every table. **Why this task is needed:** On large COW tables with scattered upserts, almost every file group is rewritten by every commit, so each discovery pass checks one file per file group. Production example: a COW table with ~10B rows (~28 TB) on GCS, partitioned by date into ~4,000 partitions, with ~197K file groups. Each commit rewrites ~197K files. | | Before | After (prototype) | |---|---|---| | Candidate files per pass | 197,191 | 197,191 | | Storage calls | ~197K `exists`, serial | 4,027 directory listings, 8 in parallel | | Latency per call | ~18 ms average | p50 85 ms / p90 132 ms / p99 245 ms | | Existence check | ~3,600 s | 51.4 s | | Total discovery pass | ~60 min | 56.5 s (~64× faster) | In the prototype, the existence check is now 51 s of the 56.5 s pass. The rest is timeline analysis 2.5 s, commit metadata 0.9 s, collecting candidates 1.4 s, and building splits 1.0 s. Slow discovery also causes duplicate splits. Flink's `ExecutorNotifier` schedules discovery at a fixed rate, so a pass that takes longer than `read.streaming.check-interval` (300 s here) is followed immediately by another. That next pass starts from a position that hasn't advanced yet and rediscovers the same commits. Keeping discovery well under the interval avoids this. I have an implementation running in production and plan to open a PR. ### Task Type Code improvement/refactoring ### Related Issues **Parent feature issue:** (if applicable ) **Related issues:** NOTE: Use `Relationships` button to add parent/blocking issues after issue is created. -- 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]
