rahil-c commented on PR #19698: URL: https://github.com/apache/hudi/pull/19698#issuecomment-5364895180
Self-review pass on this stack before it comes out of draft. Nothing blocking; the materializer seam looks like the right abstraction. Five things worth a second look, most significant first. **1. `selectDistinctObjects` changes behaviour for every existing cloud source, not just the unstructured one** (`CloudObjectsSelectorCommon.java`, `selectDistinctObjects`) This used to be `select(bucket, key, size).distinct()`. It now keeps only the newest event per `(bucket, key)` when the events carry a timestamp, so an object written twice inside one batch with different sizes is read once instead of twice. That looks like a genuine fix, but it lands in a PR about unstructured ingestion and it adds a `Window` shuffle to the hottest path in cloud ingestion. Worth calling out explicitly in the description, or splitting into its own commit, so a reviewer of the structured path sees it. **2. An unparseable notification timestamp degrades silently and noisily, per object** (`CloudObjectsSelectorCommon.epochMillis`) `epochMillis` runs per row inside `mapPartitions`, so each unparseable value logs a WARN. It then returns `UNKNOWN_MODIFICATION_TIME`, which makes `buildRow` fall back to `fs.getFileStatus` - turning the "avoid a metadata request per object" optimisation into exactly one metadata request per object. A 100k-object batch whose timestamp column is typed as Spark `TimestampType` (whose `toString` is `2026-08-20 10:00:00.0`, not ISO-8601) would produce 100k warnings and 100k HEAD requests, with nothing in the logs explaining the slowdown. Could this be detected once per batch and logged once? **3. Partition count is sized by bytes, but work is distributed by object count** (`UnstructuredFileRows.toDataset`) `UnstructuredFileMaterializer.partitionCount` computes the partition count from parseable bytes, then `jsc.parallelize(objects, partitions)` splits the list by element count. A batch of one 100 MiB document plus 10,000 small files can land the large document and its share of small ones in a single task. `CloudObjectMetadata` now carries `size`, so bucketing into partitions by bytes is nearly free here. **4. `partitionCount` and `toDataset` are untested** (`UnstructuredFileMaterializer`, `UnstructuredFileRows`) The predicate and dedup paths are covered, which is the right instinct. The gap is `partitionCount`, whose `Math.max(byWork, Math.min(floor, Math.max(objects.size(), 1)))` silently decides parallelism - a regression there surfaces only as a slow or skewed job. Two cases would cover it: every object above `parse.max.bytes`, and a batch smaller than `defaultParallelism`. **5. Is the materializer ever serialized?** (`ColumnarFileMaterializer`) `selectorCommon` is `transient` on a `Serializable` type, held by a `CloudDataFetcher` that declares a `serialVersionUID`. Driver-only today, so this is latent rather than live, but if the fetcher ever travels to an executor `materialize()` would NPE. A line on the class stating the driver-only contract would make that safe to rely on. **Candidates I checked and discarded**, recorded so nobody re-raises them: - The extension-filter default is unchanged: `ColumnarFileMaterializer.objectKeyPredicate` keeps `.orElse(fileFormat)`, and `extensionPredicate` renders a byte-identical string for the single-extension case. - `modification_time` is the precombine field, so a `0` from a notification would be a real hazard - but `buildRow(fs, CloudObjectMetadata)` falls back to `fs.getFileStatus` on `UNKNOWN_MODIFICATION_TIME`. Handled deliberately. - The positional `row.get(3)` for the timestamp holds on all three paths: with the exists check (`repartition` preserves columns), without it, and when the metadata table has no timestamp column (three columns, caught by the `row.size() > 3` guard). - `partitionCount` ignoring bytes that are inlined but not parsed is only reachable if `parse.max.bytes` is configured below `blob.inline.max.bytes` (defaults are 128 MiB and 1 MiB), and the `floor` holds parallelism at `defaultParallelism` regardless. Rows are lazy, so peak task memory stays at one blob. -- 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]
