diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index 37dd25bf9029..34a6e02150e7 100644 --- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json +++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 3 + "modification": 4 } diff --git a/CHANGES.md b/CHANGES.md index 44e7568cb8ee..86ad7451beb0 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -100,6 +100,7 @@ * Added support for attributing BigQuery API quota and billing to a specific GCP project (quota project): `quota_project_id` parameter in `ReadFromBigQuery` or `--bigquery_quota_project_id` pipeline option, honored by the Python BigQuery IO transforms (`WriteToBigQuery` with `FILE_LOADS` or `STREAMING_INSERTS`; not yet `STORAGE_WRITE_API`, which is Java-backed), `--bigQueryQuotaProjectId` pipeline option (Java), and `bigqueryio.WithQuotaProject` read/query option (Go) ([#37431](https://github.com/apache/beam/issues/37431)). * (Java/Python/Go) DebeziumIO: `readAsJson()` can now include `op` and the remaining Debezium `source` fields (e.g. MySQL `file`, `pos`, `row`, `ts_ms`; PostgreSQL `lsn`, `txId`, `ts_ms`) in `metadata` via `withFullMetadata(true)` / `full_metadata=True` / `debeziumio.FullMetadata()`. Off by default ([#40701](https://github.com/apache/beam/issues/40701)). * [IcebergIO] AddFiles is now available through the Managed API with `Managed.write(Managed.ICEBERG_ADD_FILES)` (Java/Python) ([#40827](https://github.com/apache/beam/issues/40827)). +* (Java) IcebergIO: `AddFiles` can register files in several tables in one transform, naming each file's table from its path. Set `path_pattern` (`withPathPattern` in Java) to a glob with `{name}` placeholders, such as `gs://lake/export/{table}/**`, and make `table` a template over what it captures, such as `default.{table}`. The schema pre-pass then runs per table, so a table whose files conflict is refused and its files reported while every other table is still registered. Requires `schema_evolution_options` and `error_handling`; not supported with `dry_run`. Error rows gain a `table` field only when `path_pattern` is set, so an existing pipeline's error output is unchanged ([#40849](https://github.com/apache/beam/issues/40849)). ## New Features / Improvements @@ -133,6 +134,7 @@ * (Python) The `xgboost` extras now install `xgboost>=2.0.0,<4` and `datatable>=1.1.0,<2`, moving both the lower and upper bounds for both. The previous minimum `xgboost` and `datatable` versions (`1.6.0` and `1.0.0` respectively) are no longer allowed ([#40694](https://github.com/apache/beam/issues/40694)). * (Java) [IcebergIO] AddFiles (`IcebergAddFiles` in YAML) now carries the destination table alongside each file through the registration chain, so that one transform can register files into many tables. The coders of its internal `PCollection`s change, which makes this an update incompatible change for streaming pipelines that use it ([#40817](https://github.com/apache/beam/pull/40817)). * (YAML) `IcebergAddFiles` will be automatically replaced with the compatible managed transform. For older Beam versions, streaming update compatibility can be maintained by specifying the pipeline option `update_compatibility_version` ([#40827](https://github.com/apache/beam/issues/40827)). +* (Java) [IcebergIO] AddFiles now collapses batch input into the global window. Manifests and snapshots are sized per key and window, so windowing applied upstream of AddFiles only divided `manifest_file_size` into smaller manifests and more snapshots; its outputs are now globally windowed in batch. Streaming pipelines keep the windowing they had ([#40849](https://github.com/apache/beam/issues/40849)). * Twister2 runner supported is removed ([#35905](https://github.com/apache/beam/issues/35905))). ## Deprecations diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index 06bbee362a65..9829c8537636 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -56,20 +56,25 @@ import org.apache.beam.sdk.transforms.Combine; import org.apache.beam.sdk.transforms.Create; import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.Flatten; import org.apache.beam.sdk.transforms.GroupIntoBatches; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.ParDo; import org.apache.beam.sdk.transforms.Values; +import org.apache.beam.sdk.transforms.View; import org.apache.beam.sdk.transforms.Wait; import org.apache.beam.sdk.transforms.WithKeys; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; import org.apache.beam.sdk.transforms.windowing.GlobalWindows; import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollectionList; import org.apache.beam.sdk.values.PCollectionRowTuple; import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; @@ -194,10 +199,24 @@ public class AddFiles extends PTransform, PCollectionRowTupl private static final int DEFAULT_MAX_MANIFESTS_PER_SNAPSHOT = 100; static final Schema ERROR_SCHEMA = Schema.builder().addStringField("file").addStringField("error").build(); + + /** + * Error rows from the dynamic path, which names the table the file was headed for; empty when + * none could be determined. Single-table mode keeps {@link #ERROR_SCHEMA} as it was, so adding + * dynamic destinations does not change the shape of an existing pipeline's error output. + */ + static final Schema DYNAMIC_ERROR_SCHEMA = + Schema.builder() + .addStringField("file") + .addStringField("error") + .addStringField("table") + .build(); + private static final long MANIFEST_PREFIX = UUID.randomUUID().getMostSignificantBits(); private final IcebergCatalogConfig catalogConfig; private final String tableIdentifier; private @Nullable Duration intervalTrigger; + private @Nullable String pathPattern; private final int manifestFileSize; private final @Nullable String locationPrefix; private final @Nullable List partitionFields; @@ -261,6 +280,59 @@ public AddFiles( this.locationPrefix = locationPrefix; } + /** + * A rule that routes each file to a destination table, read from the file's own path: a glob with + * {@code {name}} placeholders, whose captures fill in the table identifier as a template. So + * {@code gs://lake/export/{tbl}/**} with table {@code sales.{tbl}} routes a file under {@code + * export/orders/} to {@code sales.orders}. + * + *

Everything that can be settled without the input is settled here: a mistyped pattern, a + * template naming something the pattern never captures, or an option this cannot be combined with + * is one exception at submit time, not one error row per file. + */ + public AddFiles withPathPattern(String pathPattern) { + // these are the only copies; the YAML keys are named because the provider defers to them. + // checkArgument, not checkState: PathPattern below throws the same for a bad pattern, and the + // provider used to throw it for these. + Preconditions.checkArgument( + evolution.isEnabled(), + "A path pattern needs schema evolution (schema_evolution_options in YAML): the pre-pass is" + + " what decides each table."); + Preconditions.checkArgument( + !evolution.getDryRun(), + "A dry run of a path pattern (dry_run in YAML) is not supported yet."); + Preconditions.checkArgument( + Strings.isNullOrEmpty(locationPrefix), + "A path pattern cannot be combined with a location prefix (location_prefix in YAML): one" + + " fixed prefix cannot stop short of each file's own table folder, so the folder ends" + + " up inside the partition path ('orders/dt=2024-01-15/' where the table is" + + " partitioned only by 'dt')."); + PathPattern compiled = PathPattern.compile(pathPattern); + compiled.checkRenderable(tableIdentifier); + this.pathPattern = pathPattern; + return this; + } + + /** + * Collapses batch input to one window, for two reasons: the schema pre-pass is one pass over the + * whole input, and {@link GroupIntoBatches} sizes per key and window, so upstream windowing only + * divides {@code manifestFileSize} into smaller manifests and more snapshots. The default trigger + * keeps a triggered input from firing the pre-pass more than once. + * + *

Streaming keeps the caller's windowing: the snapshots output descends from this, and a + * global window under the default trigger never closes. + * + *

Applied before the input branches rather than after, so the error output of each branch + * shares this windowing. {@code FlattenErrors} rejects a mix. + */ + private static PCollection collapseToOneWindow(PCollection input) { + return input.apply( + "CollapseToOneWindow", + Window.into(new GlobalWindows()) + .triggering(DefaultTrigger.of()) + .discardingFiredPanes()); + } + @Override public PCollectionRowTuple expand(PCollection input) { if (input.isBounded().equals(UNBOUNDED)) { @@ -284,28 +356,63 @@ public PCollectionRowTuple expand(PCollection input) { "AddFiles configured to build partition metadata after the prefix: '{}'", locationPrefix); } - PCollection paths = input; - if (evolution.isEnabled()) { - // one commit per window, and for bounded input the window is the whole input - PCollection windowed = - input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows())); - PCollection> 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> errorSources = new ArrayList<>(); + + PCollection files = + input.isBounded().equals(UNBOUNDED) ? input : collapseToOneWindow(input); + + PCollection> keyedPaths; + + if (pathPattern != null) { + PCollectionTuple routed = + files.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> routedPaths = + routed + .get(RouteFiles.ROUTED) + .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())); + + PCollectionView> verdicts = + verdictView(distinctSchemasPerTable(routedPaths), settings); + + PCollectionTuple gated = + routedPaths.apply( + "TableGate", + ParDo.of(new TableGate(verdicts)) + .withSideInputs(verdicts) + .withOutputTags(TableGate.RELEASED, TupleTagList.of(TableGate.ERRORS))); + errorSources.add(gated.get(TableGate.ERRORS).setRowSchema(DYNAMIC_ERROR_SCHEMA)); + keyedPaths = + gated + .get(TableGate.RELEASED) + .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())); + } else { + PCollection paths = files; + if (evolution.isEnabled()) { + PCollection> schemas = distinctSchemas(files); + if (evolution.getDryRun()) { + return report(schemas, settings); + } + PCollection committed = commitSchema(schemas, settings); + paths = files.apply("WaitForSchemaCommit", Wait.on(committed)); } - PCollection committed = commitSchema(schemas, settings); - paths = windowed.apply("WaitForSchemaCommit", Wait.on(committed)); - } - PCollection> keyedPaths = - paths.apply( - "KeyByTable", - WithKeys.of(tableIdentifier).withKeyType(TypeDescriptors.strings())); + keyedPaths = + paths.apply( + "KeyByTable", + WithKeys.of(tableIdentifier).withKeyType(TypeDescriptors.strings())); + } PCollectionTuple dataFiles = keyedPaths.apply( @@ -313,6 +420,7 @@ public PCollectionRowTuple expand(PCollection input) { ParDo.of( new ConvertToDataFile( catalogConfig, + pathPattern != null, locationPrefix, partitionFields, sortFields, @@ -373,8 +481,15 @@ public void process( .apply("CommitManifests", ParDo.of(new CommitManifestFilesDoFn(catalogConfig))) .setRowSchema(SnapshotInfo.getSchema()); - return PCollectionRowTuple.of( - OUTPUT_TAG, snapshots, ERROR_TAG, dataFiles.get(ERRORS).setRowSchema(ERROR_SCHEMA)); + errorSources.add( + dataFiles + .get(ERRORS) + .setRowSchema(pathPattern == null ? ERROR_SCHEMA : DYNAMIC_ERROR_SCHEMA)); + + PCollection errors = + PCollectionList.of(errorSources).apply("FlattenErrors", Flatten.pCollections()); + + return PCollectionRowTuple.of(OUTPUT_TAG, snapshots, ERROR_TAG, errors); } /** The footer read is keyed, so the single-table path supplies its one table as the key. */ @@ -399,8 +514,33 @@ private PCollection> distinctSchemas( Combine.globally(new CollectDistinctSchemas()).withoutDefaults()); } + /** One schema union per table: the same CombineFn, lifted to a per-key combine. */ + private PCollection>> distinctSchemasPerTable( + PCollection> routed) { + return fileSchemas(routed) + .apply("CollectDistinctSchemasPerTable", Combine.perKey(new CollectDistinctSchemas())); + } + + /** The verdicts, as a map the gate can look a table up in. */ + private PCollectionView> verdictView( + PCollection>> schemas, + CommitSchemaUnion.Settings settings) { + SchemaCoder verdictCoder; + try { + verdictCoder = SchemaRegistry.createDefault().getSchemaCoder(SchemaCommitVerdict.class); + } catch (Exception e) { + throw new RuntimeException(e); + } + return schemas + .apply( + "CommitSchemaPerTable", + ParDo.of(new CommitSchemaPerTable(catalogConfig, settings, committer))) + .setCoder(KvCoder.of(StringUtf8Coder.of(), verdictCoder)) + .apply("VerdictsByTable", View.asMap()); + } + /** Commits the plan for the window's schemas once; the signal releases the gated paths. */ - private PCollection commitSchema( + private PCollection commitSchema( PCollection> schemas, CommitSchemaUnion.Settings settings) { return schemas.apply( @@ -464,6 +604,7 @@ AddFiles withSchemaCommitter(CommitSchemaUnion.Committer committer) { static class ConvertToDataFile extends DoFn, KV> { private final IcebergCatalogConfig catalogConfig; + private final boolean dynamic; public static final TupleTag ERRORS = new TupleTag<>(); public static final TupleTag> DATA_FILES = new TupleTag<>(); private final @Nullable String prefix; @@ -499,12 +640,14 @@ private TableState stateFor(String tableName) { public ConvertToDataFile( IcebergCatalogConfig catalogConfig, + boolean dynamic, @Nullable String prefix, @Nullable List partitionFields, @Nullable List sortFields, @Nullable Map tableProps) { this( catalogConfig, + dynamic, prefix, partitionFields, sortFields, @@ -514,12 +657,14 @@ public ConvertToDataFile( public ConvertToDataFile( IcebergCatalogConfig catalogConfig, + boolean dynamic, @Nullable String prefix, @Nullable List partitionFields, @Nullable List sortFields, @Nullable Map tableProps, SchemaEvolutionConfig evolution) { this.catalogConfig = catalogConfig; + this.dynamic = dynamic; this.prefix = prefix; this.partitionFields = partitionFields; this.sortFields = sortFields; @@ -1035,21 +1180,21 @@ private boolean firstTime(String tableName, String key) { return checkStateNotNull(warned).add(tableName + ":" + key); } - private static ProcessResult errorResult( + private ProcessResult errorResult( String tableName, String filePath, String message, Instant timestamp, BoundedWindow window, PaneInfo pane) { - return new ProcessResult( - tableName, - null, - Row.withSchema(ERROR_SCHEMA).addValues(filePath, message).build(), - null, - timestamp, - window, - pane); + Row errorRow; + if (dynamic) { + errorRow = + Row.withSchema(DYNAMIC_ERROR_SCHEMA).addValues(filePath, message, tableName).build(); + } else { + errorRow = Row.withSchema(ERROR_SCHEMA).addValues(filePath, message).build(); + } + return new ProcessResult(tableName, null, errorRow, null, timestamp, window, pane); } static T transformValue(Transform transform, Type type, ByteBuffer bytes) { diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java index dced27a3aa2b..fd70d0b18515 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java @@ -70,9 +70,28 @@ 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: 'sales.{tbl}', or" + + " 'sales_{country}.{tbl}' to put a captured value inside a name. Every placeholder" + + " it uses has to be one the pattern captures.") public abstract String getTable(); + @SchemaFieldDescription( + "A rule that routes each file to a destination table, read from the file's own path." + + " Without it, every file is registered in the single table named by 'table'. The" + + " rule is a glob over the whole file path, with '{name}' placeholders that each" + + " capture one folder, and those captures fill in the 'table' template. For example" + + " path_pattern 'gs://lake/export/{tbl}/**' with table 'sales.{tbl}' routes" + + " 'gs://lake/export/orders/part-0.parquet' to table 'sales.orders', and" + + " 'gs://lake/export/customers/part-0.parquet' to 'sales.customers'. A '*' stays" + + " inside one folder; '**' crosses folders and may only appear after the last" + + " placeholder. A file whose path matches nothing is reported instead of registered," + + " so this needs error_handling, and it needs schema_evolution_options. It also adds" + + " a 'table' field to the error output, naming the table each reported file was" + + " headed for. Batch pipelines only; not with dry_run or location_prefix.") + public abstract @Nullable String getPathPattern(); + @SchemaFieldDescription("Properties used to set up the Iceberg catalog.") public abstract @Nullable Map getCatalogProperties(); @@ -171,7 +190,10 @@ public static Builder builder() { + " that is not allowed, or it conflicts with the table or with another file." + " FAIL_PIPELINE (the default) fails the pipeline before any schema change is" + " committed. ROUTE_TO_ERRORS commits the changes for the other files and sends the" - + " incompatible files to the error output; it requires error_handling.") + + " incompatible files to the error output; it requires error_handling." + + " When path_pattern is set neither value fails the pipeline: FAIL_PIPELINE refuses" + + " only the table whose files conflict, that table's files go to the error output," + + " and every other table is still registered.") public abstract @Nullable String getIncompatibleSchemaHandling(); @SchemaFieldDescription( @@ -194,6 +216,8 @@ public static Builder builder() { public abstract static class Builder { public abstract Builder setTable(String table); + public abstract Builder setPathPattern(String pathPattern); + public abstract Builder setCatalogProperties(Map catalogProperties); public abstract Builder setConfigProperties(Map confProperties); @@ -332,6 +356,30 @@ 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) { + // the one check withPathPattern cannot make: AddFiles has no notion of error handling. + // 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); + } + PCollectionRowTuple result = input .getSinglePCollection() @@ -340,17 +388,7 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { "ExtractPaths", MapElements.into(TypeDescriptors.strings()) .via(row -> checkStateNotNull(row.getString(0)))) - .apply( - new AddFiles( - configuration.getIcebergCatalog(), - configuration.getTable(), - configuration.getLocationPrefix(), - configuration.getPartitionFields(), - configuration.getSortFields(), - configuration.getTableProperties(), - configuration.getManifestFileSize(), - frequency != null ? Duration.standardSeconds(frequency) : null, - configuration.getSchemaEvolution())); + .apply(addFiles); PCollectionRowTuple output = PCollectionRowTuple.of("snapshots", result.get(OUTPUT_TAG)); ErrorHandling errorHandling = configuration.getErrorHandling(); diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java index 768fb564de9b..d99e0a45209a 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java @@ -30,7 +30,7 @@ * Commits one schema union per window (the combine output is one element per window) and emits the * resulting schema id as the signal for the Wait.on gate ahead of file registration. */ -class CommitSchemaOnce extends DoFn, Long> { +class CommitSchemaOnce extends DoFn, SchemaCommitVerdict> { static final String COMMITS_COUNTER = "numSchemaCommits"; private static final Counter numSchemaCommits = counter(CommitSchemaOnce.class, COMMITS_COUNTER); @@ -58,19 +58,19 @@ class CommitSchemaOnce extends DoFn, Lo @ProcessElement public void process( - @Element List schemas, OutputReceiver out) { + @Element List schemas, + OutputReceiver out) { if (catalog == null) { catalog = catalogConfig.catalog(); } TableIdentifier tableId = IcebergUtils.parseTableIdentifier(identifier); // The committer runs only when a transaction is actually committed, so wrapping it counts // real commits and skips no-op windows. - CommitSchemaUnion.Committer counting = + CommitSchemaUnion.Committer countingCommitter = txn -> { committer.commit(txn); numSchemaCommits.inc(); }; - long schemaId = CommitSchemaUnion.commit(catalog, tableId, schemas, settings, counting); - out.output(schemaId); + out.output(CommitSchemaUnion.commit(catalog, tableId, schemas, settings, countingCommitter)); } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaPerTable.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaPerTable.java new file mode 100644 index 000000000000..b521128dcc0a --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaPerTable.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.metrics.Metrics.counter; + +import java.util.List; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.values.KV; +import org.apache.iceberg.catalog.Catalog; +import org.checkerframework.checker.nullness.qual.MonotonicNonNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Commits one schema union per table per window and emits a verdict for each, so that a table that + * cannot be registered safely is refused on its own while the others load. + * + *

Never throws: a refusal is a value. The gate downstream reads these verdicts as a side input + * and decides which tables' files may proceed. + */ +class CommitSchemaPerTable + extends DoFn< + KV>, KV> { + + private static final Logger LOG = LoggerFactory.getLogger(CommitSchemaPerTable.class); + + static final String COMMITS_COUNTER = "numSchemaCommits"; + static final String REFUSALS_COUNTER = "numTablesRefused"; + private static final Counter numSchemaCommits = + counter(CommitSchemaPerTable.class, COMMITS_COUNTER); + private static final Counter numTablesRefused = + counter(CommitSchemaPerTable.class, REFUSALS_COUNTER); + + private final IcebergCatalogConfig catalogConfig; + private final CommitSchemaUnion.Settings settings; + private final CommitSchemaUnion.Committer committer; + private transient @MonotonicNonNull Catalog catalog; + + CommitSchemaPerTable( + IcebergCatalogConfig catalogConfig, + CommitSchemaUnion.Settings settings, + CommitSchemaUnion.Committer committer) { + this.catalogConfig = catalogConfig; + this.settings = settings; + this.committer = committer; + } + + @ProcessElement + public void process( + @Element KV> element, + OutputReceiver> out) { + if (catalog == null) { + catalog = catalogConfig.catalog(); + } + String table = element.getKey(); + CommitSchemaUnion.Committer countingCommitter = + txn -> { + committer.commit(txn); + numSchemaCommits.inc(); + }; + SchemaCommitVerdict verdict = + CommitSchemaUnion.tryCommit( + catalog, table, element.getValue(), settings, countingCommitter); + if (verdict.refused()) { + numTablesRefused.inc(); + LOG.warn("Table {} refused by the schema pre-pass: {}", table, verdict.getRefusal()); + } + out.output(KV.of(table, verdict)); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java index 093749798780..83d4e0020e63 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java @@ -82,9 +82,6 @@ final class CommitSchemaUnion { private static final Counter throttledMillis = Metrics.counter(Metrics.THROTTLE_TIME_NAMESPACE, Metrics.THROTTLE_TIME_COUNTER_NAME); - /** Returned when the table does not exist and there is no schema to create it from. */ - static final long NO_TABLE = -1L; - /** What a table is created with when it does not exist yet; unused on an existing table. */ static final class NewTableSettings implements Serializable { final @Nullable List partitionFields; @@ -137,12 +134,12 @@ static final class IncompatibleSchemaException extends IllegalStateException { private CommitSchemaUnion() {} /** - * Applies the schemas and returns the table's schema id after the call, or {@link #NO_TABLE} when - * the table is missing and there is no schema to create it from. + * Applies the schemas and returns the table's schema id after the call, or a refusal naming why + * no table could be created. * * @param schemas the window's distinct schema groups, most common first */ - static long commit( + static SchemaCommitVerdict commit( Catalog catalog, TableIdentifier tableId, List schemas, @@ -152,7 +149,7 @@ static long commit( } /** Test hook: the sleeper keeps the retry tests from waiting out the backoff. */ - static long commit( + static SchemaCommitVerdict commit( Catalog catalog, TableIdentifier tableId, List schemas, @@ -166,6 +163,35 @@ static long commit( table -> commitOnce(catalog, tableId, table, schemas, settings, committer)); } + /** + * As {@link #commit}, except that no failure escapes: one table's problem refuses that table and + * the rest of the window's tables load. + */ + static SchemaCommitVerdict tryCommit( + Catalog catalog, + String table, + List schemas, + Settings settings, + Committer committer) { + try { + // the parse belongs inside the catch: a rendered name the catalog cannot accept is this + // table's problem, not the run's + return commit( + catalog, IcebergUtils.parseTableIdentifier(table), schemas, settings, committer); + } catch (RuntimeException e) { + // Deliberately no retry here, and none added to withRetry. Iceberg's REST client already + // retries 429 and 503 itself (ExponentialHttpRequestRetryStrategy, 5 attempts by default), + // so a transient blip never reaches this. It does not retry 500/502/504, because the commit + // may have landed and retrying could commit twice. Either way what arrives here has + // genuinely failed, and retrying it would only multiply the backoff. + // + // Logged with the throwable because this also catches bugs in this code, which would + // otherwise surface only as an unexplained error row. + LOG.warn("Refusing {}: the schema pre-pass failed", table, e); + return SchemaCommitVerdict.refused(AddFiles.errorMessage(e)); + } + } + /** One attempt at a plan or a commit, against the table as it is now; null when missing. */ private interface Attempt { T run(@Nullable Table table); @@ -174,6 +200,9 @@ private interface Attempt { /** * Runs an attempt again, against a fresh load of the table, when a concurrent commit or a create * race invalidates the state it started from. Gives up when the backoff time is spent. + * + *

Only stale state belongs here: reloading the table is what makes the retry meaningful. A + * flaky catalog is a different problem, handled in {@link #tryCommit}. */ private static T withRetry( Catalog catalog, TableIdentifier tableId, Sleeper sleeper, Attempt once) { @@ -255,7 +284,7 @@ static SchemaPlan plan( table -> SchemaPlan.compute(catalog, tableId, table, schemas, settings)); } - private static long commitOnce( + private static SchemaCommitVerdict commitOnce( Catalog catalog, TableIdentifier tableId, @Nullable Table table, @@ -266,7 +295,8 @@ private static long commitOnce( if (plan instanceof SchemaPlan.Creation) { return create((SchemaPlan.Creation) plan, catalog, tableId, settings, committer); } - return evolve((SchemaPlan.Evolution) plan, tableId, settings.handling, committer); + return SchemaCommitVerdict.committed( + evolve((SchemaPlan.Evolution) plan, tableId, settings.handling, committer)); } /** Replays the planned union onto the table and repairs the name mapping, in one commit. */ @@ -330,7 +360,7 @@ private static void replay(Transaction txn, Schema merged, TableIdentifier table * canonical order (sorted by name at every level), not in any file's declared order; later unions * append after them. */ - private static long create( + private static SchemaCommitVerdict create( SchemaPlan.Creation plan, Catalog catalog, TableIdentifier tableId, @@ -338,13 +368,15 @@ private static long create( Committer committer) { if (plan.schemasToMerge.isEmpty() && plan.incompatibleSchemas.isEmpty()) { LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId); - return NO_TABLE; + return SchemaCommitVerdict.refused( + "the table does not exist and no file schema was read to create it from"); } failOrWarnOnIncompatibleSchemas( tableId, plan.incompatibleSchemas, settings.handling, "no table was created"); if (plan.newSchema == null) { LOG.info("Table {} does not exist and no file schema can seed it; not creating it", tableId); - return NO_TABLE; + return SchemaCommitVerdict.refused( + "the table does not exist and every file schema read was refused"); } if (plan.problem != null) { throw new IllegalStateException(plan.problem); @@ -371,7 +403,7 @@ private static long create( tableId, plan.schemasToMerge.size(), schemaId); - return schemaId; + return SchemaCommitVerdict.committed(schemaId); } private static void failOrWarnOnIncompatibleSchemas( diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PathPattern.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PathPattern.java index 33ed4c608521..e98350b7f555 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PathPattern.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PathPattern.java @@ -193,8 +193,8 @@ String remainder() { } /** - * Fills a table template from a path's captures, so {@code warehouse.{db}.{tbl}} with {@code - * db=sales, tbl=orders} gives {@code warehouse.sales.orders}. + * Fills a table template from a path's captures, so {@code {db}.{tbl}} with {@code db=sales, + * tbl=orders} gives {@code sales.orders}. */ static String render(String template, Map captures) { // split with -1 gives one more literal part than there are placeholders, so literals[i + 1] diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RouteFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RouteFiles.java index 5a6d7c047220..bd77eeb3f533 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RouteFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RouteFiles.java @@ -37,7 +37,7 @@ final class RouteFiles extends DoFn> { static final TupleTag ERRORS = new TupleTag<>(); /** The same rows as the transform's other errors, so one output can carry both. */ - static final Schema ERROR_SCHEMA = AddFiles.ERROR_SCHEMA; + static final Schema ERROR_SCHEMA = AddFiles.DYNAMIC_ERROR_SCHEMA; private final String pathPattern; private final String tableTemplate; @@ -57,14 +57,15 @@ public void process(@Element String path, MultiOutputReceiver output) { if (match == null) { String error = "did not match path_pattern '" + pathPattern + "'"; - output.get(ERRORS).output(errorRow(path, error)); + // empty table: nothing matched, so there is none to attribute the file to + output.get(ERRORS).output(errorRow(path, error, "")); return; } String table = PathPattern.render(tableTemplate, match.captures()); output.get(ROUTED).output(KV.of(table, path)); } - private static Row errorRow(String file, String error) { - return Row.withSchema(ERROR_SCHEMA).addValues(file, error).build(); + private static Row errorRow(String file, String error, String table) { + return Row.withSchema(ERROR_SCHEMA).addValues(file, error, table).build(); } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaCommitVerdict.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaCommitVerdict.java new file mode 100644 index 000000000000..88706e9cbad4 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaCommitVerdict.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import com.google.auto.value.AutoValue; +import org.apache.beam.sdk.schemas.AutoValueSchema; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What the schema pre-pass decided for one table: a schema id when it committed, or a reason when + * it refused. Exactly one of the two is set. + */ +@DefaultSchema(AutoValueSchema.class) +@AutoValue +abstract class SchemaCommitVerdict { + + /** The table's schema id after the commit; null when refused. */ + @SchemaFieldNumber("0") + abstract @Nullable Long getSchemaId(); + + /** Why the table was refused; null when committed. */ + @SchemaFieldNumber("1") + abstract @Nullable String getRefusal(); + + static SchemaCommitVerdict committed(long schemaId) { + return new AutoValue_SchemaCommitVerdict(schemaId, null); + } + + static SchemaCommitVerdict refused(String reason) { + return new AutoValue_SchemaCommitVerdict(null, reason); + } + + boolean refused() { + return getRefusal() != null; + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableGate.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableGate.java new file mode 100644 index 000000000000..64affbbb3b16 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableGate.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.metrics.Metrics.counter; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; + +import java.util.Map; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollectionView; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.TupleTag; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Holds every file back until the pre-pass has a verdict for every table, then lets through only + * the files of tables it did not refuse. + * + *

Two jobs in one step. The barrier is the side input: reading it cannot complete until the + * whole verdict collection exists, which is the ordering {@code Wait.on} manufactures for a stage + * that reads nothing. The filter is the per-key lookup, which {@code Wait.on} cannot express at + * all, having no key dimension and discarding element values. + */ +final class TableGate extends DoFn, KV> { + + static final TupleTag> RELEASED = new TupleTag<>(); + static final TupleTag ERRORS = new TupleTag<>(); + + static final String HELD_COUNTER = "numFilesHeldBackByRefusal"; + private static final Counter numFilesHeldBack = counter(TableGate.class, HELD_COUNTER); + + private final PCollectionView> verdicts; + + TableGate(PCollectionView> verdicts) { + this.verdicts = verdicts; + } + + @ProcessElement + public void process(ProcessContext context) { + KV element = context.element(); + Map byTable = context.sideInput(verdicts); + @Nullable SchemaCommitVerdict verdict = byTable.get(element.getKey()); + if (verdict == null) { + // No verdict means no file of this table contributed a readable schema, so the pre-pass + // never saw it. Released: registration reports each file with its own reason. + context.output(element); + return; + } + if (verdict.refused()) { + numFilesHeldBack.inc(); + // the reason alone; the table it was headed for is the row's third field + context.output( + ERRORS, + Row.withSchema(AddFiles.DYNAMIC_ERROR_SCHEMA) + .addValues( + element.getValue(), checkStateNotNull(verdict.getRefusal()), element.getKey()) + .build()); + return; + } + context.output(element); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesDynamicDestinationsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesDynamicDestinationsTest.java new file mode 100644 index 000000000000..113916e39378 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesDynamicDestinationsTest.java @@ -0,0 +1,363 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.EnumSet; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.windowing.AfterPane; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; +import org.apache.beam.sdk.transforms.windowing.Repeatedly; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.values.PCollectionRowTuple; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.TimestampedValue; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.data.parquet.GenericParquetWriter; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.parquet.Parquet; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** End-to-end dynamic destinations against a local Hadoop catalog. */ +@RunWith(JUnit4.class) +public class AddFilesDynamicDestinationsTest { + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + + @Rule public TemporaryFolder temp = new TemporaryFolder(); + @Rule public TestPipeline pipeline = TestPipeline.create(); + + @Rule + public transient TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default"); + + private String root; + private HadoopCatalog catalog; + private IcebergCatalogConfig catalogConfig; + + private static final Schema ORDERS = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "item", Types.StringType.get())); + + @Before + public void setup() { + root = temp.getRoot().getAbsolutePath() + "/"; + catalog = new HadoopCatalog(new Configuration(), warehouse.location); + catalogConfig = + IcebergCatalogConfig.builder() + .setCatalogProperties( + ImmutableMap.of("type", "hadoop", "warehouse", warehouse.location)) + .build(); + } + + private SchemaEvolutionConfig evolution() { + return SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .build(); + } + + private SchemaEvolutionConfig dryRunEvolution() { + return SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .setDryRun(true) + .build(); + } + + private AddFiles addFiles( + String template, @Nullable String locationPrefix, @Nullable SchemaEvolutionConfig evolution) { + return new AddFiles( + catalogConfig, template, locationPrefix, null, null, null, null, null, evolution); + } + + private String write(String relative, Schema schema, Record... records) throws IOException { + String file = root + relative; + new File(file).getParentFile().mkdirs(); + DataWriter writer = + Parquet.writeData(org.apache.iceberg.Files.localOutput(file)) + .schema(schema) + .withSpec(PartitionSpec.unpartitioned()) + .createWriterFunc(GenericParquetWriter::create) + .build(); + try { + for (Record record : records) { + writer.write(record); + } + } finally { + writer.close(); + } + return file; + } + + private static Record order(long id, String item) { + Record record = GenericRecord.create(ORDERS); + record.setField("id", id); + record.setField("item", item); + return record; + } + + private PCollectionRowTuple run(String pattern, String template, List files) { + AddFiles addFiles = addFiles(template, null, evolution()).withPathPattern(pattern); + return pipeline.apply("Files", Create.of(files)).apply(addFiles); + } + + private long rowCount(TableIdentifier id) { + if (!catalog.tableExists(id)) { + return -1; + } + Table table = catalog.loadTable(id); + if (table.currentSnapshot() == null) { + return 0; + } + long total = 0; + for (org.apache.iceberg.DataFile file : table.currentSnapshot().addedDataFiles(table.io())) { + total += file.recordCount(); + } + return total; + } + + /** Three tables under one pattern: all three load, each only its own files. */ + @Test + public void threeTablesEachGetOnlyItsOwnFiles() throws Exception { + catalog.createNamespace(Namespace.of("lake")); + List files = + Arrays.asList( + write("export/orders/a.parquet", ORDERS, order(1, "x"), order(2, "y")), + write("export/orders/b.parquet", ORDERS, order(3, "z")), + write("export/customers/a.parquet", ORDERS, order(4, "c")), + write("export/events/a.parquet", ORDERS, order(5, "e"))); + + PCollectionRowTuple out = run(root + "export/{tbl}/**", "lake.{tbl}", files); + PAssert.that(out.get(AddFiles.ERROR_TAG)).empty(); + pipeline.run().waitUntilFinish(); + + assertEquals(3, rowCount(TableIdentifier.of("lake", "orders"))); + assertEquals(1, rowCount(TableIdentifier.of("lake", "customers"))); + assertEquals(1, rowCount(TableIdentifier.of("lake", "events"))); + } + + /** A path that matches nothing is an error row, and does not stop the others. */ + @Test + public void anUnmatchedPathDoesNotStopTheRest() throws Exception { + catalog.createNamespace(Namespace.of("lake")); + String stray = write("stray.parquet", ORDERS, order(9, "s")); + List files = + Arrays.asList(write("export/orders/a.parquet", ORDERS, order(1, "x")), stray); + + PCollectionRowTuple out = run(root + "export/{tbl}/**", "lake.{tbl}", files); + PAssert.that(out.get(AddFiles.ERROR_TAG)) + .satisfies( + rows -> { + List all = new ArrayList<>(); + rows.forEach(all::add); + assertEquals("expected exactly the stray: " + all, 1, all.size()); + assertEquals(stray, all.get(0).getString("file")); + assertEquals("", all.get(0).getString("table")); + return null; + }); + pipeline.run().waitUntilFinish(); + + assertEquals(1, rowCount(TableIdentifier.of("lake", "orders"))); + } + + /** The invariant that matters: every input file either registers or is reported. */ + @Test + public void everyFileEitherRegistersOrIsReported() throws Exception { + catalog.createNamespace(Namespace.of("lake")); + Set inputs = new HashSet<>(); + inputs.add(write("export/orders/a.parquet", ORDERS, order(1, "x"))); + inputs.add(write("export/orders/b.parquet", ORDERS, order(2, "y"))); + inputs.add(write("nope/c.parquet", ORDERS, order(3, "z"))); + + run(root + "export/{tbl}/**", "lake.{tbl}", new ArrayList<>(inputs)); + pipeline.run().waitUntilFinish(); + + Table orders = catalog.loadTable(TableIdentifier.of("lake", "orders")); + Set registered = new HashSet<>(); + for (org.apache.iceberg.DataFile file : orders.currentSnapshot().addedDataFiles(orders.io())) { + registered.add(file.path().toString()); + } + assertEquals(2, registered.size()); + assertTrue(registered.contains(root + "export/orders/a.parquet")); + assertFalse(registered.contains(root + "nope/c.parquet")); + } + + /** The central question: one table refused, the other two load completely. */ + @Test + public void aRefusedTableDoesNotStopTheOthers() throws Exception { + catalog.createNamespace(Namespace.of("lake")); + // customers exists with id as a long; its file declares id as a string, which cannot merge + Schema conflicting = + new Schema( + required(1, "id", Types.StringType.get()), optional(2, "item", Types.StringType.get())); + catalog.createTable(TableIdentifier.of("lake", "customers"), ORDERS); + + Record bad = GenericRecord.create(conflicting); + bad.setField("id", "not-a-long"); + bad.setField("item", "c"); + + List files = + Arrays.asList( + write("export/orders/a.parquet", ORDERS, order(1, "x")), + write("export/customers/a.parquet", conflicting, bad), + write("export/events/a.parquet", ORDERS, order(5, "e"))); + + run(root + "export/{tbl}/**", "lake.{tbl}", files); + pipeline.run().waitUntilFinish(); + + assertEquals("orders loaded", 1, rowCount(TableIdentifier.of("lake", "orders"))); + assertEquals("events loaded", 1, rowCount(TableIdentifier.of("lake", "events"))); + Table customers = catalog.loadTable(TableIdentifier.of("lake", "customers")); + assertEquals("customers has no snapshot", null, customers.currentSnapshot()); + assertEquals("customers schema untouched", ORDERS.asStruct(), customers.schema().asStruct()); + } + + /** A table whose files are all unreadable is refused rather than failing the run. */ + @Test + public void aTableWithNothingReadableIsRefusedAlone() throws Exception { + catalog.createNamespace(Namespace.of("lake")); + File notParquet = new File(root + "export/junk/a.txt"); + notParquet.getParentFile().mkdirs(); + java.nio.file.Files.write( + notParquet.toPath(), "not parquet".getBytes(java.nio.charset.StandardCharsets.UTF_8)); + + List files = + Arrays.asList( + write("export/orders/a.parquet", ORDERS, order(1, "x")), notParquet.getAbsolutePath()); + + run(root + "export/{tbl}/**", "lake.{tbl}", files); + pipeline.run().waitUntilFinish(); + + assertEquals("orders still loaded", 1, rowCount(TableIdentifier.of("lake", "orders"))); + assertFalse( + "junk table was not created", catalog.tableExists(TableIdentifier.of("lake", "junk"))); + } + + /** + * A custom-windowed input used to fail at construction: the error output taken off RouteFiles + * kept the input's windowing while every other branch had been moved to the global window, and + * the Flatten over them refused the mix. + */ + @Test + public void aCustomWindowedInputStillRegistersAndReports() throws Exception { + catalog.createNamespace(Namespace.of("lake")); + String routed = write("export/orders/a.parquet", ORDERS, order(1, "x")); + String stray = write("stray.parquet", ORDERS, order(9, "s")); + + AddFiles addFiles = + addFiles("lake.{tbl}", null, evolution()).withPathPattern(root + "export/{tbl}/**"); + PCollectionRowTuple out = + pipeline + .apply( + "Files", + Create.timestamped( + TimestampedValue.of(routed, new Instant(0)), + TimestampedValue.of(stray, new Instant(0)))) + .apply("FixedWindows", Window.into(FixedWindows.of(Duration.standardMinutes(1)))) + .apply(addFiles); + + PAssert.that(out.get(AddFiles.ERROR_TAG)) + .satisfies( + rows -> { + List all = new ArrayList<>(); + rows.forEach(all::add); + assertEquals("expected exactly the stray: " + all, 1, all.size()); + assertEquals(stray, all.get(0).getString("file")); + return null; + }); + pipeline.run().waitUntilFinish(); + + assertEquals(1, rowCount(TableIdentifier.of("lake", "orders"))); + } + + /** + * A triggered input used to fire the pre-pass per pane, and a pane that carried no readable + * schema refused the table outright: the gate then held back every file of a table that was fine. + */ + @Test + public void aTriggeredInputCommitsTheSchemaOnce() throws Exception { + catalog.createNamespace(Namespace.of("lake")); + List files = + Arrays.asList( + write("export/orders/a.parquet", ORDERS, order(1, "x")), + write("export/orders/b.parquet", ORDERS, order(2, "y"))); + + AddFiles addFiles = + addFiles("lake.{tbl}", null, evolution()).withPathPattern(root + "export/{tbl}/**"); + PCollectionRowTuple out = + pipeline + .apply("Files", Create.of(files)) + .apply( + "FirePerElement", + Window.into(new GlobalWindows()) + .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1))) + .discardingFiredPanes()) + .apply(addFiles); + PAssert.that(out.get(AddFiles.ERROR_TAG)).empty(); + pipeline.run().waitUntilFinish(); + + assertEquals(2, rowCount(TableIdentifier.of("lake", "orders"))); + } + + /** What the pattern cannot be combined with fails where it is set, not once per file. */ + @Test + public void optionsThePatternCannotBeCombinedWithFailWhereItIsSet() { + String pattern = root + "export/{tbl}/**"; + + AddFiles noEvolution = addFiles("lake.{tbl}", null, null); + assertThrows(IllegalArgumentException.class, () -> noEvolution.withPathPattern(pattern)); + + AddFiles dryRun = addFiles("lake.{tbl}", null, dryRunEvolution()); + assertThrows(IllegalArgumentException.class, () -> dryRun.withPathPattern(pattern)); + + AddFiles locationPrefix = addFiles("lake.{tbl}", root, evolution()); + assertThrows(IllegalArgumentException.class, () -> locationPrefix.withPathPattern(pattern)); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java index 64ea7e873828..2c8dbd3999dc 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java @@ -1802,6 +1802,35 @@ public void testUpstreamWindowedBatchInputEvolvesAndRegisters() throws Exception assertEmailAddedAndFilesRegistered(2); } + /** + * Manifests and snapshots are sized per key and window, so upstream windowing used to divide + * them: two files 45s apart under 30s windows committed two snapshots. Batch input is collapsed + * to one window, so one snapshot, and this holds with no pre-pass to need it. + */ + @Test + public void testUpstreamWindowedBatchInputWithoutEvolutionCommitsOneSnapshot() throws Exception { + catalog.createTable(tableId, icebergSchema); + String first = writeOneRecord("first.parquet"); + String second = writeOneRecord("second.parquet"); + + PCollectionRowTuple output = + pipeline + .apply( + "Create Input", + Create.timestamped( + TimestampedValue.of(first, new Instant(0)), + TimestampedValue.of(second, new Instant(45_000)))) + .apply("UpstreamWindow", Window.into(FixedWindows.of(Duration.standardSeconds(30)))) + .apply(addFiles(null)); + PAssert.that(output.get("errors")).empty(); + + pipeline.run().waitUntilFinish(); + + Table table = catalog.loadTable(tableId); + assertEquals(2, Iterables.size(table.newScan().planFiles())); + assertEquals(1, Iterables.size(table.snapshots())); + } + /** * The schema commit retries a CommitFailedException (another writer got in first). The committer * is serialized with the DoFn, so the retry shows in the table and in the time the backoff @@ -1869,7 +1898,7 @@ private PCollectionTuple convert(SchemaEvolutionConfig config, String... files) .apply( ParDo.of( new AddFiles.ConvertToDataFile( - catalogConfig, null, null, null, null, config)) + catalogConfig, false, null, null, null, null, config)) .withOutputTags( AddFiles.ConvertToDataFile.DATA_FILES, TupleTagList.of(AddFiles.ConvertToDataFile.ERRORS))); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java index 07a0adeb6cfb..10a829eecc4e 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java @@ -80,7 +80,7 @@ public void testCommitsTheWindowsSchemasAndEmitsTheSchemaId() { CollectDistinctSchemas.SchemaGroup.of( SchemaParser.toJson(FileSchemas.canonical(file)), 3L, Arrays.asList())); - PCollection schemaIds = + PCollection verdicts = pipeline .apply( Create.of(Arrays.asList(schemas)).withCoder(CollectDistinctSchemas.outputCoder())) @@ -93,7 +93,7 @@ public void testCommitsTheWindowsSchemasAndEmitsTheSchemaId() { SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), IncompatibleSchemaHandling.FAIL_PIPELINE, new CommitSchemaUnion.NewTableSettings(null, null, null))))); - PAssert.that(schemaIds).containsInAnyOrder(1L); + PAssert.that(verdicts).containsInAnyOrder(SchemaCommitVerdict.committed(1L)); PipelineResult result = pipeline.run(); result.waitUntilFinish(); assertEquals(1, commitsCounted(result)); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java index ce226f37f9f0..09e42d0d8199 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.iceberg; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; import static org.hamcrest.MatcherAssert.assertThat; @@ -140,6 +141,13 @@ private long commit( SchemaEvolutionConfig config, IncompatibleSchemaHandling handling, CollectDistinctSchemas.SchemaGroup... schemas) { + return checkStateNotNull(commitVerdict(config, handling, schemas).getSchemaId()); + } + + private SchemaCommitVerdict commitVerdict( + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + CollectDistinctSchemas.SchemaGroup... schemas) { return CommitSchemaUnion.commit( catalog, tableId, @@ -811,6 +819,16 @@ private long commitTo( IncompatibleSchemaHandling handling, CommitSchemaUnion.NewTableSettings creation, CollectDistinctSchemas.SchemaGroup... schemas) { + return checkStateNotNull( + commitVerdictTo(id, config, handling, creation, schemas).getSchemaId()); + } + + private SchemaCommitVerdict commitVerdictTo( + TableIdentifier id, + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + CommitSchemaUnion.NewTableSettings creation, + CollectDistinctSchemas.SchemaGroup... schemas) { return CommitSchemaUnion.commit( catalog, id, @@ -1005,14 +1023,17 @@ public void testCreationIsNotGatedOnAnyParticularOption() { @Test public void testMissingTableWithoutSchemasIsNotCreated() { TableIdentifier id = missing(); - long result = + SchemaCommitVerdict result = CommitSchemaUnion.commit( catalog, id, new ArrayList<>(), settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION), CommitSchemaUnion.DEFAULT_COMMITTER); - assertEquals(CommitSchemaUnion.NO_TABLE, result); + assertTrue(result.refused()); + assertEquals( + "the table does not exist and no file schema was read to create it from", + result.getRefusal()); assertFalse(catalog.tableExists(id)); } @@ -1180,15 +1201,17 @@ public void testInvalidColumnNamesCannotSeedACreatedTable() { assertFalse(catalog.tableExists(id)); // nothing valid to seed from: routed, and no table - long none = - commitTo( + SchemaCommitVerdict none = + commitVerdictTo( id, ALL, IncompatibleSchemaHandling.ROUTE_TO_ERRORS, NO_CREATION, files(dotted, 1), files(caseDuplicate, 2)); - assertEquals(CommitSchemaUnion.NO_TABLE, none); + assertTrue(none.refused()); + assertEquals( + "the table does not exist and every file schema read was refused", none.getRefusal()); assertFalse(catalog.tableExists(id)); // the valid schema alone seeds the table diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RouteFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RouteFilesTest.java index f10e9abc870b..6c80e75614bc 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RouteFilesTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RouteFilesTest.java @@ -87,7 +87,8 @@ public void aPathThatDoesNotMatchBecomesAnErrorRow() { Row.withSchema(RouteFiles.ERROR_SCHEMA) .addValues( "gs://lake/stray.parquet", - "did not match path_pattern 'gs://lake/{db}/{tbl}/**'") + "did not match path_pattern 'gs://lake/{db}/{tbl}/**'", + "") .build()); pipeline.run().waitUntilFinish(); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/TableGateTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/TableGateTest.java new file mode 100644 index 000000000000..dd2708e706cf --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/TableGateTest.java @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.metrics.MetricNameFilter; +import org.apache.beam.sdk.metrics.MetricResult; +import org.apache.beam.sdk.metrics.MetricsFilter; +import org.apache.beam.sdk.schemas.NoSuchSchemaException; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.View; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.PCollectionView; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.TupleTagList; +import org.junit.Rule; +import org.junit.Test; + +/** The gate's three outcomes, and that a refusal holds back only its own table. */ +public class TableGateTest { + + @Rule public final TestPipeline pipeline = TestPipeline.create(); + + private static final String ORDERS = "default.orders"; + private static final String EVENTS = "default.events"; + private static final String ORDERS_FILE = "gs://lake/export/orders/a.parquet"; + private static final String EVENTS_FILE = "gs://lake/export/events/b.parquet"; + private static final String REASON = "every file schema read was refused"; + + @Test + public void aCommittedTableReleasesItsFiles() { + PCollectionTuple out = + gate( + Arrays.asList(KV.of(ORDERS, SchemaCommitVerdict.committed(7L))), + Arrays.asList(KV.of(ORDERS, ORDERS_FILE))); + + PAssert.that(out.get(TableGate.RELEASED)).containsInAnyOrder(KV.of(ORDERS, ORDERS_FILE)); + PAssert.that(out.get(TableGate.ERRORS)).empty(); + pipeline.run().waitUntilFinish(); + } + + /** The subtle branch: no verdict means the pre-pass never saw the table, not that it refused. */ + @Test + public void aTableWithNoVerdictIsReleased() { + PCollectionTuple out = + gate( + Arrays.asList(KV.of(ORDERS, SchemaCommitVerdict.committed(7L))), + Arrays.asList(KV.of(EVENTS, EVENTS_FILE))); + + PAssert.that(out.get(TableGate.RELEASED)).containsInAnyOrder(KV.of(EVENTS, EVENTS_FILE)); + PAssert.that(out.get(TableGate.ERRORS)).empty(); + pipeline.run().waitUntilFinish(); + } + + /** Fields are read by name: a transposed row would pass an equality check against a twin. */ + @Test + public void aRefusedTableReportsEveryFileWithTheReasonAndTheTable() { + PCollectionTuple out = + gate( + Arrays.asList(KV.of(ORDERS, SchemaCommitVerdict.refused(REASON))), + Arrays.asList(KV.of(ORDERS, ORDERS_FILE), KV.of(ORDERS, ORDERS_FILE + "2"))); + + PAssert.that(out.get(TableGate.RELEASED)).empty(); + PAssert.that(out.get(TableGate.ERRORS)) + .satisfies( + rows -> { + int seen = 0; + for (Row row : rows) { + seen++; + assertEquals(REASON, row.getString("error")); + assertEquals(ORDERS, row.getString("table")); + assertTrue( + "unexpected file: " + row.getString("file"), + ORDERS_FILE.equals(row.getString("file")) + || (ORDERS_FILE + "2").equals(row.getString("file"))); + } + assertEquals(2, seen); + return null; + }); + + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + // files, not tables: two files of one refused table + assertEquals(2, filesHeldBack(result)); + } + + /** The point of the feature: one table's refusal must not cost another table its files. */ + @Test + public void oneRefusedTableDoesNotHoldBackAnother() { + PCollectionTuple out = + gate( + Arrays.asList( + KV.of(ORDERS, SchemaCommitVerdict.refused(REASON)), + KV.of(EVENTS, SchemaCommitVerdict.committed(3L))), + Arrays.asList(KV.of(ORDERS, ORDERS_FILE), KV.of(EVENTS, EVENTS_FILE))); + + PAssert.that(out.get(TableGate.RELEASED)).containsInAnyOrder(KV.of(EVENTS, EVENTS_FILE)); + PAssert.that(out.get(TableGate.ERRORS)) + .satisfies( + rows -> { + for (Row row : rows) { + assertEquals(ORDERS, row.getString("table")); + assertEquals(ORDERS_FILE, row.getString("file")); + } + return null; + }); + pipeline.run().waitUntilFinish(); + } + + private PCollectionTuple gate( + List> verdicts, List> files) { + PCollectionView> view = + pipeline + .apply("Verdicts", Create.of(verdicts).withCoder(verdictCoder())) + .apply("AsView", View.asMap()); + PCollectionTuple out = + pipeline + .apply("Files", Create.of(files)) + .apply( + ParDo.of(new TableGate(view)) + .withSideInputs(view) + .withOutputTags(TableGate.RELEASED, TupleTagList.of(TableGate.ERRORS))); + out.get(TableGate.ERRORS).setRowSchema(AddFiles.DYNAMIC_ERROR_SCHEMA); + return out; + } + + private static KvCoder verdictCoder() { + try { + return KvCoder.of( + StringUtf8Coder.of(), + SchemaRegistry.createDefault().getSchemaCoder(SchemaCommitVerdict.class)); + } catch (NoSuchSchemaException e) { + throw new RuntimeException(e); + } + } + + private static long filesHeldBack(PipelineResult result) { + long total = 0; + for (MetricResult counter : + result + .metrics() + .queryMetrics( + MetricsFilter.builder() + .addNameFilter(MetricNameFilter.named(TableGate.class, TableGate.HELD_COUNTER)) + .build()) + .getCounters()) { + total += counter.getAttempted(); + } + return total; + } +} diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml index e3a9180be9e9..17945c18c068 100644 --- a/sdks/python/apache_beam/yaml/standard_io.yaml +++ b/sdks/python/apache_beam/yaml/standard_io.yaml @@ -693,6 +693,7 @@ mappings: 'IcebergAddFiles': table: 'table' + path_pattern: 'path_pattern' catalog_properties: 'catalog_properties' config_properties: 'config_properties' triggering_frequency_seconds: 'triggering_frequency_seconds' diff --git a/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dynamic.yaml b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dynamic.yaml new file mode 100644 index 000000000000..75f9abce7885 --- /dev/null +++ b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dynamic.yaml @@ -0,0 +1,157 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# One IcebergAddFiles registers files in several tables, naming each table from the file's +# own path. Note on '{{tbl}}': this harness runs str.format over the spec to fill +# '{TEMP_DIR}', so a literal brace has to be doubled here. A real YAML pipeline writes +# '{tbl}' once. + +fixtures: + - name: TEMP_DIR + type: "tempfile.TemporaryDirectory" + + +pipelines: + # Pipeline 1: one table's files, in a folder named after the table + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - type: WriteToParquet + config: + path: "{TEMP_DIR}/export/orders/data" + file_name_suffix: ".parquet" + num_shards: 1 + + # Pipeline 2: a second table, same layout + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "389a", rank: 2} + - type: WriteToParquet + config: + path: "{TEMP_DIR}/export/events/data" + file_name_suffix: ".parquet" + num_shards: 1 + + # Pipeline 3: a file outside the pattern, so the error output has something in it + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "stray", rank: 9} + - type: WriteToParquet + config: + path: "{TEMP_DIR}/misc/stray/data" + file_name_suffix: ".parquet" + num_shards: 1 + + # Pipeline 4: register everything in one pass. path_pattern decides the table per file; + # the file under misc/ matches nothing and is reported instead of registered. + - pipeline: + type: composite + transforms: + - type: ReadMatchFiles + name: Match + config: + file_pattern: "{TEMP_DIR}/*/*/*.parquet" + - type: MapToFields + name: Paths + input: Match + config: + language: python + fields: + path: + callable: "lambda row: row.path" + output_type: string + - type: IcebergAddFiles + name: AddFiles + input: Paths + config: + table: "default.{{tbl}}" + path_pattern: "{TEMP_DIR}/export/{{tbl}}/**" + catalog_properties: + type: "hadoop" + warehouse: "{TEMP_DIR}/dir" + schema_evolution_options: [ALLOW_FIELD_ADDITION] + error_handling: + output: error_output + - type: WriteToJson + name: WriteErrorsToJson + input: AddFiles.error_output + config: + path: "{TEMP_DIR}/error.json" + + providers: + - type: python + config: { } + transforms: + ReadMatchFiles: 'apache_beam.io.fileio.MatchFiles' + + # Pipeline 5: the first table holds only its own rows + - pipeline: + type: chain + transforms: + - type: ReadFromIceberg + config: + table: "default.orders" + catalog_properties: + type: "hadoop" + warehouse: "{TEMP_DIR}/dir" + - type: AssertEqual + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + + # Pipeline 6: and the second table only its own + - pipeline: + type: chain + transforms: + - type: ReadFromIceberg + config: + table: "default.events" + catalog_properties: + type: "hadoop" + warehouse: "{TEMP_DIR}/dir" + - type: AssertEqual + config: + elements: + - {label: "389a", rank: 2} + + # Pipeline 7: the unroutable file is reported, with the table it could not be given + - pipeline: + type: chain + transforms: + - type: ReadFromJson + config: + path: "{TEMP_DIR}/error.json*" + - type: AssertEqual + config: + elements: + - file: "{TEMP_DIR}/misc/stray/data-00000-of-00001.parquet" + error: "did not match path_pattern '{TEMP_DIR}/export/{{tbl}}/**'" + table: ""