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:
|
|
R: @ahmedabu98 |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
| routed | ||
| .get(RouteFiles.ROUTED) | ||
| .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())) | ||
| .apply("PrePassGlobalWindow", Window.into(new GlobalWindows())); |
There was a problem hiding this comment.
Maybe move PrePassGlobalWindow up before RouteFiles so that its error output also lines up with the windowing of the other error sources
Also reset the trigger (triggering(DefaultTrigger.of()).discardingFiredPanes())
This would show up if a user passes an input with custom windowing
| "path_pattern needs error_handling to receive the files it could not route or" | ||
| + " register"); | ||
| addFiles = addFiles.withPathPattern(pathPattern); | ||
| } |
There was a problem hiding this comment.
Can we move these checks inside the AddFiles constructor? or into withPathPattern
| @SchemaFieldDescription( | ||
| "A fully-qualified table identifier. When path_pattern is set this is a template" | ||
| + " instead, filled from what the pattern captures, for example" | ||
| + " 'warehouse.{db}.{table}'; every placeholder it uses has to be one the pattern" |
There was a problem hiding this comment.
I would use a different example, maybe something like orders_{country}.{table}
"Warehouse" is not commonly used in the table name
| + " headed for and empty when none could be determined; without it the error output" | ||
| + " keeps the two fields it has always had. Cannot be combined with dry_run. Batch" | ||
| + " pipelines only.") | ||
| public abstract @Nullable String getPathPattern(); |
There was a problem hiding this comment.
Can we humanize this description a little :)
| + "The provided prefix should go up until the partition information:%n" | ||
| + "'gs://bucket/namespace/table/data/'.%n" | ||
| + "If not provided, will try determining each DataFile's partition from its metrics metadata.") | ||
| public abstract @Nullable String getLocationPrefix(); |
There was a problem hiding this comment.
I think we may need to also support templating for location prefix. Can be done in a future PR
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.