claudevdm commented on code in PR #40852:
URL: https://github.com/apache/beam/pull/40852#discussion_r4413737491
##########
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:
I applied this to input for all batch pipelines. I think that is what this
transform is optimized for in batch WDYT?
--
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]
