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 c4cf7e11f133..08d5e6ef3587 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -68,6 +68,7 @@ * (Java) FirestoreIO: added `RpcQosOptions.Builder.withRampupThrottlingDisabled()`, mirroring the existing DatastoreIO option, to opt out of the 500/50/5 write ramp-up ([#27170](https://github.com/apache/beam/issues/27170)). * Added support for attributing BigQuery API quota and billing to a specific GCP project (quota project): `quota_project_id` parameter in `ReadFromBigQuery` or `--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)). +* (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 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 c97a20c6221b..2d22aa3b4bbd 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,10 +56,12 @@ 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; @@ -68,8 +70,10 @@ 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; @@ -179,10 +183,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; @@ -246,6 +264,20 @@ public AddFiles( this.locationPrefix = locationPrefix; } + /** + * Registers each file in the table its own path names. The pattern is a glob with {@code {name}} + * placeholders, and {@code table} becomes a template over what it captures. + * + *

Both checks run here rather than per file: a mistyped pattern or a template naming something + * the pattern never captures is one exception at submit time, not one error row per file. + */ + public AddFiles withPathPattern(String pathPattern) { + PathPattern compiled = PathPattern.compile(pathPattern); + compiled.checkRenderable(tableIdentifier); + this.pathPattern = pathPattern; + return this; + } + @Override public PCollectionRowTuple expand(PCollection input) { if (input.isBounded().equals(UNBOUNDED)) { @@ -269,28 +301,69 @@ 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> 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> windowed = + routed + .get(RouteFiles.ROUTED) + .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())) + .apply("PrePassGlobalWindow", Window.into(new GlobalWindows())); + + PCollectionView> verdicts = + verdictView(distinctSchemasPerTable(windowed), settings); + + PCollectionTuple gated = + windowed.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 = 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); + if (evolution.getDryRun()) { + return report(schemas, settings); + } + PCollection committed = commitSchema(schemas, settings); + paths = windowed.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( @@ -298,6 +371,7 @@ public PCollectionRowTuple expand(PCollection input) { ParDo.of( new ConvertToDataFile( catalogConfig, + pathPattern != null, locationPrefix, partitionFields, sortFields, @@ -358,8 +432,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. */ @@ -384,8 +465,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( @@ -449,6 +555,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; @@ -484,12 +591,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, @@ -499,12 +608,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; @@ -1020,21 +1131,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 c5cc758c555b..0c6a5f186706 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 @@ -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(); + @SchemaFieldDescription("Properties used to set up the Iceberg catalog.") public abstract @Nullable Map getCatalogProperties(); @@ -166,7 +187,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( @@ -189,6 +213,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); @@ -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 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); + } + PCollectionRowTuple result = input .getSinglePCollection() @@ -321,17 +379,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/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..7188f1099255 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesDynamicDestinationsTest.java @@ -0,0 +1,261 @@ +/* + * 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.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.values.PCollectionRowTuple; +import org.apache.beam.sdk.values.Row; +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.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 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 = + new AddFiles(catalogConfig, template, null, null, null, null, null, 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"))); + } +} 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..9eb32f2f1242 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 @@ -1869,7 +1869,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 9dc3eff38b1e..0ff31883f9e7 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: ""