rahil-c opened a new pull request, #19698: URL: https://github.com/apache/hudi/pull/19698
### Describe the issue this Pull Request addresses issue: #19697 Stacked on #19672, which should be reviewed and merged first. Until it merges this PR shows both commits. Unstructured file ingestion can only discover files by listing a prefix, so every sync walks the whole prefix whether or not anything arrived. Hudi already avoids that for data files by reading a Hudi table of S3 or GCS notifications, but the step that turns the selected objects into rows was hardwired to a Spark datasource read, so documents could not use it. ### Summary and Changelog Introduces `CloudObjectMaterializer`, the step that turns the objects a batch has selected into rows. Three concerns move together and so live on one type: the predicate deciding which object keys are eligible, the partition count, and the read itself. `ColumnarFileMaterializer` holds the previous behaviour and is registered as the default, and `generateFilter` keeps its existing two-argument form and delegates through it, so nothing changes for current pipelines. `UnstructuredFileMaterializer` selects by document extension, sizes partitions by the bytes that will actually be parsed rather than the bytes referenced, and builds rows directly. An object above `parse.max.bytes` is referenced without being read and costs almost nothing, which makes a byte count of the whole batch a poor proxy for its cost. `UnstructuredFileS3EventsHoodieIncrSource` wires the two together. Row construction is now shared by both discovery paths through `UnstructuredFileRows`, so blob placement and parse handling exist once rather than once per source. `CloudObjectMetadata` carries the notification timestamp, which the table needs to order repeated writes to the same path, and objects are de-duplicated on bucket and key keeping the newest event rather than on size as well: an object written twice inside one batch was previously read once per write. Where a metadata table predates the timestamp column the events cannot be ordered, so that falls back to the previous behaviour with a warning rather than failing. ### Impact Additive. `ColumnarFileMaterializer` is the default and reproduces the previous read exactly, and no existing test file is modified by this PR, which is the evidence for that. The de-duplication change does affect existing cloud sources: an object rewritten inside a single batch is now read once instead of once per write. New user-facing config: `hoodie.streamer.source.unstructured.work.bytes.per.partition`. ### Risk Level low. The behaviour-preserving claim is checked by the existing suites rather than asserted: `TestS3EventsHoodieIncrSource` (16), `TestGcsEventsHoodieIncrSource` (11), `TestCloudObjectsSelectorCommon` (29), `TestUnstructuredFileDFSSource` (2) and `TestCloudObjectsSelector` (50) all pass unchanged, including the argument captor that pins the partition count and the tests that pin the generated filter string. Validated end to end against real S3 on a 4,032 object corpus spread over 1,008 prefixes: the new source produces a table identical in content to the folder-based source, 4,044 rows with 4,044 successful parses after 12 further objects were added. A discovery-cost comparison on the same corpus measured 121.8s by listing against 12.3s by notification for a sync with nothing to ingest. Two defects were found by that run rather than by the suite, and both are fixed here with tests that reproduce them: the de-duplication ranked over nested columns after projecting them, which renames them to their leaf names and fails the batch with `UNRESOLVED_COLUMN`; and the rendered object-key predicate was built from a hash set, so its order was arbitrary across runs. ### Documentation Update New config documented on the config class, so the reference tables generate from it. The website page describing unstructured ingestion will need the new source class name and a note that `hoodie.streamer.source.s3incr.fs.prefix` must match the registered filesystem scheme, since `s3` is the default while `s3a` is what `hadoop-aws` registers. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable -- 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]
