ahmedabu98 commented on code in PR #40852:
URL: https://github.com/apache/beam/pull/40852#discussion_r4404211859
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java:
##########
@@ -65,9 +65,30 @@ public static Builder builder() {
return new
AutoValue_AddFilesSchemaTransformProvider_Configuration.Builder();
}
- @SchemaFieldDescription("A fully-qualified table identifier.")
+ @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"
+ + " captures.")
public abstract String getTable();
+ @SchemaFieldDescription(
+ "Registers each file in the table that its own path names, instead of
registering every"
+ + " file in one table. It is a glob over the whole path with
'{name}' placeholders"
+ + " that each capture one folder: 'gs://lake/export/{table}/**'
over"
+ + " 'gs://lake/export/orders/part-0.parquet' captures
table=orders, and that file is"
+ + " registered in whichever table the 'table' template then
renders. A '*' stays"
+ + " inside one folder, '**' crosses folders and may only appear
after the last"
+ + " placeholder, and a path the pattern does not match goes to the
error output"
+ + " instead of being registered anywhere. Requires
schema_evolution_options, because"
+ + " the schema pre-pass is what decides each table, and
error_handling, because that"
+ + " is where unroutable files and refused tables are reported.
Setting it also adds"
+ + " a 'table' field to the error output, naming the table each
reported file was"
+ + " 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();
Review Comment:
Can we humanize this description a little :)
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java:
##########
@@ -65,9 +65,30 @@ public static Builder builder() {
return new
AutoValue_AddFilesSchemaTransformProvider_Configuration.Builder();
}
- @SchemaFieldDescription("A fully-qualified table identifier.")
+ @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"
Review Comment:
I would use a different example, maybe something like
`orders_{country}.{table}`
"Warehouse" is not commonly used in the table name
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java:
##########
@@ -269,35 +301,77 @@ public PCollectionRowTuple expand(PCollection<String>
input) {
"AddFiles configured to build partition metadata after the prefix:
'{}'", locationPrefix);
}
- PCollection<String> paths = input;
- if (evolution.isEnabled()) {
- // one commit per window, and for bounded input the window is the whole
input
- PCollection<String> windowed =
- input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
- PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas =
distinctSchemas(windowed);
- CommitSchemaUnion.Settings settings =
- new CommitSchemaUnion.Settings(
- evolution,
- evolution.incompatibleSchemaHandlingFor(input.isBounded()),
- new CommitSchemaUnion.NewTableSettings(partitionFields,
sortFields, tableProps));
- if (evolution.getDryRun()) {
- return report(schemas, settings);
+ CommitSchemaUnion.Settings settings =
+ new CommitSchemaUnion.Settings(
+ evolution,
+ evolution.incompatibleSchemaHandlingFor(input.isBounded()),
+ new CommitSchemaUnion.NewTableSettings(partitionFields,
sortFields, tableProps));
+
+ // every stage that can reject a file appends here; one Flatten publishes
them together
+ List<PCollection<Row>> errorSources = new ArrayList<>();
+
+ PCollection<KV<String, String>> keyedPaths;
+
+ if (pathPattern != null) {
+ checkState(
+ evolution.isEnabled(),
+ "A path pattern needs schema evolution: the pre-pass is what decides
each table.");
+ checkState(!evolution.getDryRun(), "A dry run of a path pattern is not
supported yet.");
+
+ PCollectionTuple routed =
+ input.apply(
+ "RouteFiles",
+ ParDo.of(new RouteFiles(checkStateNotNull(pathPattern),
tableIdentifier))
+ .withOutputTags(RouteFiles.ROUTED,
TupleTagList.of(RouteFiles.ERRORS)));
+
errorSources.add(routed.get(RouteFiles.ERRORS).setRowSchema(DYNAMIC_ERROR_SCHEMA));
+
+ PCollection<KV<String, String>> windowed =
+ routed
+ .get(RouteFiles.ROUTED)
+ .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))
+ .apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
Review 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
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java:
##########
@@ -313,6 +339,38 @@ public PCollectionRowTuple expand(PCollectionRowTuple
input) {
@Nullable Integer frequency =
configuration.getTriggeringFrequencySeconds();
+ AddFiles addFiles =
+ new AddFiles(
+ configuration.getIcebergCatalog(),
+ configuration.getTable(),
+ configuration.getLocationPrefix(),
+ configuration.getPartitionFields(),
+ configuration.getSortFields(),
+ configuration.getTableProperties(),
+ configuration.getManifestFileSize(),
+ frequency != null ? Duration.standardSeconds(frequency) : null,
+ configuration.getSchemaEvolution());
+
+ String pathPattern = configuration.getPathPattern();
+ if (pathPattern != null) {
+ // AddFiles checks the first two against its own fields; these name
the YAML keys
+ List<String> evolutionOptions =
configuration.getSchemaEvolutionOptions();
+ Preconditions.checkArgument(
+ evolutionOptions != null && !evolutionOptions.isEmpty(),
+ "path_pattern needs schema_evolution_options: the schema pre-pass
is what decides"
+ + " which table each file belongs to");
+ Preconditions.checkArgument(
+ !Boolean.TRUE.equals(configuration.getDryRun()),
+ "path_pattern cannot be combined with dry_run");
+ // with no error output an unmatched path and a refused table's files
are dropped
+ // silently, which is the one outcome this transform must not have
+ Preconditions.checkArgument(
+ ErrorHandling.hasOutput(configuration.getErrorHandling()),
+ "path_pattern needs error_handling to receive the files it could
not route or"
+ + " register");
+ addFiles = addFiles.withPathPattern(pathPattern);
+ }
Review Comment:
Can we move these checks inside the AddFiles constructor? or into
`withPathPattern`
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java:
##########
Review Comment:
I think we may need to also support templating for location prefix. Can be
done in a future PR
--
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]
