feat(utilities): ingest unstructured files announced by GCS notifications - #19699
Draft
rahil-c wants to merge 4 commits into
Draft
feat(utilities): ingest unstructured files announced by GCS notifications#19699rahil-c wants to merge 4 commits into
rahil-c wants to merge 4 commits into
Conversation
…ntal sources hoodie.streamer.source.cloud.data.select.file.extension was interpolated into a single SQL predicate, so a comma separated value produced `s3.object.key like '%json,jsonl'`: a literal match on the joined string that selects nothing, with no error to explain it. A comma separated value now renders one predicate per extension, combined with OR inside parentheses so it composes with the size and path filters. A single value renders exactly the predicate it did before, so existing pipelines are unaffected, as is the long standing fallback to the data file format when the config is unset. generateFilter had no test coverage. Six cases were added to the existing suite, covering the single and multiple forms, blank and empty entries, the format fallback, the GCS column names, and composition with the path prefix filter.
…fications Unstructured file ingestion could only discover files by listing a prefix, which costs a full recursive walk on every sync whether or not anything arrived, and whose cost grows with the size of the bucket rather than with what is new. The cloud incremental sources already avoid that for data files by reading a Hudi table of S3 or GCS notifications, but their read step was hardwired to a Spark datasource, so documents could not use it. 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 the default, so existing pipelines are unaffected; generateFilter keeps its old two-argument form and delegates through it. 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 so costs almost nothing, which makes a byte count of the whole batch a poor proxy for its cost. 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. Objects are now 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. Verified with the existing suites unchanged: no test file is modified by this commit, and TestS3EventsHoodieIncrSource, TestGcsEventsHoodieIncrSource, TestCloudObjectsSelectorCommon and TestUnstructuredFileDFSSource all pass.
The de-duplication windowed over the bucket and key columns after selecting them. Both are nested, so the projection renames them to their leaf names and the window could no longer resolve them, failing the batch with UNRESOLVED_COLUMN on s3.bucket.name. Ranking first and projecting after resolves against the original schema. Only reachable with a genuinely nested event schema, which the existing tests did not build, so it survived the suite and surfaced on a real S3 run. Adds a test over nested events that reproduces it, and one covering a metadata table with no timestamp column, which must keep working.
…ions The GCS twin of UnstructuredFileS3EventsHoodieIncrSource. Everything that differs between the two object stores already sits behind CloudObjectsSelectorCommon.Type, so the new class differs only in which base it extends: discovery, batching, checkpointing and row construction are all shared. Also makes the rendered object-key predicate deterministic. The allowed and ignored extension sets are hash sets, so the predicate came out in arbitrary order, which made it unstable across runs in logs and query plans and unassertable in a test. Sorting before rendering fixes all three. Adds predicate coverage for both stores: the default denylist of data-file extensions, an allowlist overriding it, and the case where nothing is filtered and no predicate should be added at all.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19699 +/- ##
============================================
- Coverage 77.83% 77.83% -0.01%
- Complexity 33140 33169 +29
============================================
Files 2527 2533 +6
Lines 139942 140056 +114
Branches 16849 16860 +11
============================================
+ Hits 108926 109010 +84
- Misses 23396 23428 +32
+ Partials 7620 7618 -2
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
Collaborator
Collaborator
Author
|
Reviewed as part of a pass over the whole stack; the substantive comments are on #19698 since that is where the materializer seam and the event de-duplication change live, and they apply here too. Nothing GCS-specific to raise. The |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe the issue this Pull Request addresses
closes #19697
Stacked on #19698, which should be reviewed and merged first; this PR completes the same gap for GCS. Until #19698 merges this PR shows its commits too.
#19698 made unstructured files ingestable from S3 notifications. GCS notifications reach Hudi through the same two-stage pattern and the same shared helpers, so the equivalent source was missing for no reason other than that nobody had added the class.
Summary and Changelog
Adds
UnstructuredFileGcsEventsHoodieIncrSource. Everything that differs between the two object stores already sits behindCloudObjectsSelectorCommon.Type, which selects the object key column, the size column, the notification timestamp column and the URL prefix. The new class therefore differs from its S3 counterpart only in which base it extends: discovery, batching, checkpointing and row construction are all shared.Also makes the rendered object-key predicate deterministic. The allowed and ignored extension sets are hash sets, so the predicate came out in arbitrary order, which made it unstable across runs in logs and query plans and unassertable in a test. Sorting before rendering fixes all three, and is what lets the tests below assert the predicate exactly.
Impact
Additive: one new optional source class. The ordering change alters only the order of
and-joined predicates within the generated filter, which is semantically identical.Risk Level
low. The class is a constructor; all behaviour it relies on is already covered. Predicate coverage was added for both stores: the default denylist of data-file extensions, an allowlist overriding it, and the case where nothing is filtered and no predicate should be added at all.
TestCloudObjectsSelectorCommon(32),TestGcsEventsHoodieIncrSource(11),TestS3EventsHoodieIncrSource(16) andTestUnstructuredFileDFSSource(2) pass, with checkstyle clean.Exercised end to end through the real Google GCS Hadoop connector (
gcs-connector-hadoop3-2.2.21) against a localfake-gcs-server: three objects ingested overgs://, all parsing successfully, withmodification_timepopulated from the GCSupdatedcolumn, which is the part that could not be inferred from the S3 run. The generated filter came out over the flat GCS columns (size > 0 and name not like ...) rather than S3's nested ones.Not yet run against a real GCS bucket, so bucket-level behaviour such as pagination and throttling is unverified. The emulator stands in for the storage API, not for GCS itself.
Documentation Update
Covered by the same website update as #19698; this adds one more source class name to it.
Contributor's checklist