Repository navigation
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40852 +/- ##
===========================================
+ Coverage 0 79.91% +79.91%
===========================================
Files 0 493 +493
Lines 0 83479 +83479
===========================================
+ Hits 0 66709 +66709
- Misses 0 16770 +16770
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Collaborator
Author
|
R: @ahmedabu98 |
Contributor
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
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
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.
Completes dynamic destinations for
IcebergIO.AddFiles: one transform registers files into manytables, choosing each file's table from its own path.
Builds on #40749 (
PathPattern,RouteFiles) and #40817 (the destination-table key threadedthrough the registration chain), both merged. Those added the primitives and the plumbing with the
key held constant; this makes the key vary, and adds the YAML surface.
SchemaCommitVerdict(53 lines): the value the pre-pass produces per table.CommitSchemaUnion:commitnow returns that verdict instead of along;tryCommitwraps it so no failure escapes.
CommitSchemaPerTable(87): runs the pre-pass per table, emits one verdict each.TableGate(79): the barrier and filter. The three-outcomeprocessmethod is the heartof the change.
AddFiles.expand: theif (pathPattern != null)branch wiring the above together; theelseis the existing single-table path.AddFilesSchemaTransformProvider: thepath_patternoption and its three validations.How it fits together
The pre-pass is the same
CollectDistinctSchemasCombineFn, lifted withCombine.perKey. Thedynamic path is the existing pipeline with a key that varies, not a second implementation.
The error output is unchanged for existing pipelines.
ERROR_SCHEMAkeeps its two fields; asecond
DYNAMIC_ERROR_SCHEMAaddstableand is chosen per instance inexpand. An earlierdraft widened the single schema, which broke the public
errorsshape for every user and neededa breaking-change entry plus a fix to Beam's own
iceberg_add_files.yaml. The price of the splitis that the output shape depends on configuration, so the
path_patternfield description saysso explicitly.
FAIL_PIPELINEbecomes "refuse this table" underpath_pattern.tryCommitcatchesRuntimeException, so anIncompatibleSchemaExceptionrefuses one table and reports its filesrather than failing the run — registering 1000 tables and losing all of them to one bad table
isn't a usable feature. The option name is then a mild overstatement in this mode, which the
field description states. A distinct
REFUSE_TABLEvalue would be more honest and more surface;I'm happy to go that way instead.
error_handlingis mandatory withpath_pattern. Two of the three error producers arestructural, an unmatched path, and a refused table, so dropping the error collection can lose a
whole table's files silently. That's the one outcome this feature must not have, so the provider
refuses the combination rather than warning.
A table with no verdict is released, not refused. No verdict means no file of that table
yielded a readable schema, so the pre-pass never saw it. Releasing is safe because
getOrCreateTablerethrowsNoSuchTableExceptionwhen evolution is on, so each file becomes itsown
MISSING_TABLE_ERRORrow instead of one generic reason for the whole table, and no table isever created from a single file's schema. Only the pre-pass creates tables in dynamic mode.
Why
catch (RuntimeException)rather than an enumerated set.RESTExceptionparents bothpermanent (
Forbidden,BadRequest,UnprocessableEntity) and transient (ServiceUnavailable,ServiceFailure) subclasses, so there's no clean permanent set to list, and anything anenumeration missed would fail the pipeline — the outcome this exists to prevent.
NoSuchNamespaceExceptionextendsRuntimeExceptionrather thanNotFoundException, socatch-by-parent would miss it too. It does also catch bugs in this code, which is why the
LOG.warncarries the throwable.No retry wraps it, and the comment in
tryCommitrecords why: Iceberg's REST client alreadyretries 429 and 503 (
ExponentialHttpRequestRetryStrategy, 5 attempts by default) anddeliberately does not retry 500/502/504, since the commit may have landed.
withRetryis forstale state, where reloading the table is what makes a retry meaningful.
Dynamic destinations is batch-only, and that follows from three existing checks rather than a
new one:
path_patternrequires schema evolution, andexpandalready refuses evolution onunbounded input. Stated here because no single check says it.