You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "claudevdm (via GitHub)" <gi...@apache.org> on 2026/09/15 03:55:33 UTC
[PR] AddFiles: create the table from the file schemas when it does not exist [beam]
claudevdm opened a new pull request, #40124:
URL: https://github.com/apache/beam/pull/40124
Before this change the pre-pass required an existing table. AddFiles without evolution creates a missing table from whichever file happens to reach ConvertToDataFile first; with the pre-pass every schema of the window is known before anything is created.
Create path (CommitSchemaUnion.create), taken when loadTable throws NoSuchTableException:
- The window's schemas are folded into one union on a scratch create transaction that is never committed (the same staging loop as the evolve path, so a schema conflicting with another is reported or skipped per the handling, and the create is abandoned under FAIL_PIPELINE before anything exists).
- The real table is built directly from createdSchema(union, config): every column optional at every level - one lucky file's declared or proven required-ness must not become a table constraint that the next file violates - except pinned paths and their ancestors, which are created required. Creation is the schema-authoring moment; evolution never tightens columns afterwards. A pin that no file schema carries fails the create under FAIL_PIPELINE (later windows could only add the column optional, so the pin would stay inert forever) and warns under ROUTE_TO_ERRORS. The created table is born with a single schema version.
- Creation is not gated on any particular evolution option: the options guard an existing table's schema, and there is none to guard yet.
- Partition spec, sort order and table properties come from TableCreation (the AddFiles constructor arguments), resolved against the union, so a partition or sort column carried by any schema of the window works; the name mapping is written in the same transaction.
- Create race: two workers (or a concurrent non-evolution AddFiles) may create the table at once. AlreadyExistsException joins CommitFailedException in the retry, and the next attempt takes the evolve path against the table the other party created.
- Empty window against a missing table returns NO_TABLE (-1) and creates nothing.
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: create the table from the file schemas when it does not exist [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm merged PR #40124:
URL: https://github.com/apache/beam/pull/40124
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: create the table from the file schemas when it does not exist [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on PR #40124:
URL: https://github.com/apache/beam/pull/40124#issuecomment-5674494125
R: @ahmedabu98
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: create the table from the file schemas when it does not exist [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #40124:
URL: https://github.com/apache/beam/pull/40124#discussion_r4026017190
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java:
##########
@@ -255,57 +256,302 @@ private static long commitOnce(
return table.schema().schemaId();
}
- private static final class Accepted {
- final Schema schema;
- final String json;
- final long files;
- final SchemaDelta delta;
-
- Accepted(Schema schema, String json, long files, SchemaDelta delta) {
- this.schema = schema;
- this.json = json;
- this.files = files;
- this.delta = delta;
+ /**
+ * Sorts the window's schemas into the ones the table must change for (accepted) and the ones it
+ * must not ({@code incompatible}, with the reason); schemas the table already covers drop out.
+ */
+ private static List<Accepted> classify(
+ Table table,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ List<Incompatible> incompatible) {
+ List<Accepted> accepted = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema =
+ FileSchemas.markRequired(
+ SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns());
+ SchemaDelta delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ continue;
+ }
+ if (!delta.allowedBy(config)) {
+ incompatible.add(
+ new Incompatible(
+ group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config)));
+ continue;
+ }
+ accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta));
}
+ return accepted;
}
/**
- * Folds one union per accepted schema into a scratch transaction the caller must never commit;
- * its intermediate schema versions exist only in memory. A schema can conflict with another
- * schema's additions, which only surfaces while staging and poisons the transaction, so on a
- * conflict the offender moves to {@code incompatible} and the transaction is rebuilt without it.
+ * Unions the accepted schemas into the table schema on scratch transactions that are never
+ * committed, relaxing every field the window adds; a schema that conflicts with another only
+ * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the
+ * folded schema, or null when nothing needs to change.
*/
- private static Transaction stageAll(
+ private static @Nullable Schema fold(
Table table,
Schema base,
TableIdentifier tableId,
List<Accepted> accepted,
List<Incompatible> incompatible) {
while (true) {
- Transaction txn = newTransactionOn(table, base, tableId);
- Accepted failed = null;
- for (Accepted item : accepted) {
- // Both caught types carry staging conflicts: ValidationException from Schema
- // construction at apply ("multiple fields for name"), IllegalArgumentException from
- // SchemaUpdate preconditions ("Cannot change column type").
- try {
- stage(txn, item);
- } catch (ValidationException | IllegalArgumentException e) {
- failed = item;
- incompatible.add(
- new Incompatible(
- item.json,
- item.files,
- "conflicts with another file schema in the same window: "
- + AddFiles.errorMessage(e)));
- break;
- }
+ Transaction scratch = newTransactionOn(table, base, tableId);
+ Accepted failed = stageAll(scratch, accepted, incompatible);
+ if (failed != null) {
+ accepted.remove(failed);
+ continue;
+ }
+ if (accepted.isEmpty()) {
+ return null;
}
+ relaxNewRequiredFields(scratch, base);
+ return scratch.table().schema();
+ }
+ }
+
+ /**
+ * One union replays the fold's net effect (additions, promotions, relaxations) so the table gains
+ * a single schema version instead of one per folded schema. The checkState is a pure bug
+ * detector: concurrent changes are caught earlier, by newTransactionOn.
+ */
+ private static void replay(Transaction txn, Schema merged, TableIdentifier tableId) {
+ txn.updateSchema().unionByNameWith(merged).commit();
+ // toString of the args runs only on failure
+ Schema foldResult = TypeUtil.assignIncreasingFreshIds(merged);
+ Schema replayResult = TypeUtil.assignIncreasingFreshIds(txn.table().schema());
+ checkState(
+ replayResult.sameSchema(foldResult),
+ "replaying the folded schema union for %s diverged from the fold; fold: %s replay: %s",
+ tableId,
+ foldResult,
+ replayResult);
+ }
+
+ /**
+ * Creates the table from the union of the window's schemas, with every column optional at every
+ * level so that one lucky file cannot impose required columns on the table - except pinned
+ * columns and their ancestors, which are created required.
+ */
+ private static long create(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ TableCreation creation,
+ Committer committer) {
+ if (schemas.isEmpty()) {
+ LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId);
+ return NO_TABLE;
+ }
+ List<Incompatible> incompatible = new ArrayList<>();
+ Schema merged = foldForCreate(catalog, tableId, schemas, incompatible);
+ if (!incompatible.isEmpty()) {
+ reportIncompatible(tableId, incompatible, handling, "no table was created");
+ }
+ // The real table is built from the folded result directly.
+ Schema created = createdSchema(merged, config);
+ reportUnenforceablePins(tableId, created, config, handling);
+ Map<String, String> properties =
+ creation.properties == null ? new HashMap<>() : new HashMap<>(creation.properties);
+ Transaction txn =
+ catalog
+ .buildTable(tableId, created)
+ .withPartitionSpec(PartitionUtils.toPartitionSpec(creation.partitionFields, created))
+ .withSortOrder(SortOrderUtils.toSortOrder(creation.sortFields, created))
+ .withProperties(properties)
+ .createTransaction();
+ stageNameMapping(txn);
+ committer.commit(txn);
+ Table table = catalog.loadTable(tableId);
+ LOG.info(
+ "Created table {} from {} file schema(s), schema id {}",
+ tableId,
+ schemas.size() - incompatible.size(),
+ table.schema().schemaId());
+ return table.schema().schemaId();
Review Comment:
nit: The `schemaId` should be available using `txn.table().schema().schemaId()` instead of having to call loadTable
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java:
##########
@@ -255,57 +256,302 @@ private static long commitOnce(
return table.schema().schemaId();
}
- private static final class Accepted {
- final Schema schema;
- final String json;
- final long files;
- final SchemaDelta delta;
-
- Accepted(Schema schema, String json, long files, SchemaDelta delta) {
- this.schema = schema;
- this.json = json;
- this.files = files;
- this.delta = delta;
+ /**
+ * Sorts the window's schemas into the ones the table must change for (accepted) and the ones it
+ * must not ({@code incompatible}, with the reason); schemas the table already covers drop out.
+ */
+ private static List<Accepted> classify(
+ Table table,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ List<Incompatible> incompatible) {
+ List<Accepted> accepted = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema =
+ FileSchemas.markRequired(
+ SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns());
+ SchemaDelta delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ continue;
+ }
+ if (!delta.allowedBy(config)) {
+ incompatible.add(
+ new Incompatible(
+ group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config)));
+ continue;
+ }
+ accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta));
}
+ return accepted;
}
/**
- * Folds one union per accepted schema into a scratch transaction the caller must never commit;
- * its intermediate schema versions exist only in memory. A schema can conflict with another
- * schema's additions, which only surfaces while staging and poisons the transaction, so on a
- * conflict the offender moves to {@code incompatible} and the transaction is rebuilt without it.
+ * Unions the accepted schemas into the table schema on scratch transactions that are never
+ * committed, relaxing every field the window adds; a schema that conflicts with another only
+ * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the
+ * folded schema, or null when nothing needs to change.
*/
- private static Transaction stageAll(
+ private static @Nullable Schema fold(
Table table,
Schema base,
TableIdentifier tableId,
List<Accepted> accepted,
List<Incompatible> incompatible) {
while (true) {
- Transaction txn = newTransactionOn(table, base, tableId);
- Accepted failed = null;
- for (Accepted item : accepted) {
- // Both caught types carry staging conflicts: ValidationException from Schema
- // construction at apply ("multiple fields for name"), IllegalArgumentException from
- // SchemaUpdate preconditions ("Cannot change column type").
- try {
- stage(txn, item);
- } catch (ValidationException | IllegalArgumentException e) {
- failed = item;
- incompatible.add(
- new Incompatible(
- item.json,
- item.files,
- "conflicts with another file schema in the same window: "
- + AddFiles.errorMessage(e)));
- break;
- }
+ Transaction scratch = newTransactionOn(table, base, tableId);
+ Accepted failed = stageAll(scratch, accepted, incompatible);
+ if (failed != null) {
+ accepted.remove(failed);
+ continue;
+ }
+ if (accepted.isEmpty()) {
+ return null;
}
+ relaxNewRequiredFields(scratch, base);
+ return scratch.table().schema();
+ }
+ }
+
+ /**
+ * One union replays the fold's net effect (additions, promotions, relaxations) so the table gains
+ * a single schema version instead of one per folded schema. The checkState is a pure bug
+ * detector: concurrent changes are caught earlier, by newTransactionOn.
+ */
+ private static void replay(Transaction txn, Schema merged, TableIdentifier tableId) {
+ txn.updateSchema().unionByNameWith(merged).commit();
+ // toString of the args runs only on failure
+ Schema foldResult = TypeUtil.assignIncreasingFreshIds(merged);
+ Schema replayResult = TypeUtil.assignIncreasingFreshIds(txn.table().schema());
+ checkState(
+ replayResult.sameSchema(foldResult),
+ "replaying the folded schema union for %s diverged from the fold; fold: %s replay: %s",
+ tableId,
+ foldResult,
+ replayResult);
+ }
+
+ /**
+ * Creates the table from the union of the window's schemas, with every column optional at every
+ * level so that one lucky file cannot impose required columns on the table - except pinned
+ * columns and their ancestors, which are created required.
+ */
+ private static long create(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ TableCreation creation,
+ Committer committer) {
+ if (schemas.isEmpty()) {
+ LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId);
+ return NO_TABLE;
+ }
+ List<Incompatible> incompatible = new ArrayList<>();
+ Schema merged = foldForCreate(catalog, tableId, schemas, incompatible);
+ if (!incompatible.isEmpty()) {
+ reportIncompatible(tableId, incompatible, handling, "no table was created");
+ }
+ // The real table is built from the folded result directly.
+ Schema created = createdSchema(merged, config);
+ reportUnenforceablePins(tableId, created, config, handling);
+ Map<String, String> properties =
+ creation.properties == null ? new HashMap<>() : new HashMap<>(creation.properties);
+ Transaction txn =
+ catalog
+ .buildTable(tableId, created)
+ .withPartitionSpec(PartitionUtils.toPartitionSpec(creation.partitionFields, created))
+ .withSortOrder(SortOrderUtils.toSortOrder(creation.sortFields, created))
+ .withProperties(properties)
+ .createTransaction();
+ stageNameMapping(txn);
+ committer.commit(txn);
+ Table table = catalog.loadTable(tableId);
+ LOG.info(
+ "Created table {} from {} file schema(s), schema id {}",
+ tableId,
+ schemas.size() - incompatible.size(),
+ table.schema().schemaId());
+ return table.schema().schemaId();
+ }
+
+ /**
+ * Unions the window's schemas into one on scratch create transactions that are never committed,
+ * seeded by the most common schema; conflicts move to {@code incompatible} and the fold restarts
+ * without the offender.
+ */
+ private static Schema foldForCreate(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ List<Incompatible> incompatible) {
+ Schema seed = SchemaParser.fromJson(schemas.get(0).getSchemaJson());
+ List<Accepted> rest = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas.subList(1, schemas.size())) {
+ Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson());
+ rest.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), null));
+ }
+ while (true) {
+ Transaction scratch = catalog.buildTable(tableId, seed).createTransaction();
+ Accepted failed = stageAll(scratch, rest, incompatible);
if (failed == null) {
- return txn;
+ return scratch.table().schema();
+ }
+ rest.remove(failed);
+ }
+ }
+
+ /**
+ * A pin the created schema did not end up enforcing - the column appears in no file schema, or
+ * the configured spelling resolves to a field the pin walk did not reach (a short container
+ * spelling like a.b for a.element.b, or a path inside a map key) - would stay inert forever,
+ * since later windows only add columns optional: a config error under FAIL_PIPELINE, a warning
+ * under ROUTE_TO_ERRORS (streaming may see the column later).
+ */
+ private static void reportUnenforceablePins(
+ TableIdentifier tableId,
+ Schema created,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling) {
+ List<String> unenforceable = new ArrayList<>();
+ for (String pin : config.getRequiredColumns()) {
+ Types.NestedField field = created.findField(pin);
+ if (field == null || field.isOptional()) {
+ unenforceable.add(pin);
+ }
+ }
+ if (unenforceable.isEmpty()) {
+ return;
+ }
+ Collections.sort(unenforceable);
+ if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) {
+ throw new IncompatibleSchemaException(
+ "Pinned column(s) "
+ + unenforceable
+ + " appear in none of the file schemas creating "
+ + tableId
+ + ", or their spelling does not match the column path; the created table cannot"
+ + " make them required");
+ }
+ LOG.warn(
+ "Pinned column(s) {} appear in none of the file schemas creating {}, or their spelling"
+ + " does not match the column path; the created table cannot make them required",
+ unenforceable,
+ tableId);
+ }
+
+ /**
+ * The created schema: every field optional at every level, list elements and map values included,
+ * except pinned paths and their ancestors, which stay required so the schema advertises the
+ * guarantee the per-file pin check enforces (a null ancestor nulls the pinned leaf). Map key
+ * subtrees keep their declared shape (keys are required by definition; pins inside them are not
+ * honored). Nothing depends on a created table's schema yet, so this is the schema-authoring
+ * moment; evolution never tightens columns afterwards.
+ */
+ static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) {
+ Pins pins = new Pins(config.getRequiredColumns());
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : merged.asStruct().fields()) {
+ fields.add(createdField(field, field.name(), pins));
+ }
+ return new Schema(fields);
+ }
+
+ private static Types.NestedField createdField(Types.NestedField field, String path, Pins pins) {
+ boolean required = pins.isPinned(path) || pins.pinnedColumnBeneath(path) != null;
+ return Types.NestedField.from(field)
+ .ofType(createdType(field.type(), path, pins))
+ .isOptional(!required)
+ .build();
+ }
+
+ private static Type createdType(Type type, String path, Pins pins) {
+ if (type.isStructType()) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : type.asStructType().fields()) {
+ fields.add(createdField(field, path + "." + field.name(), pins));
+ }
+ return Types.StructType.of(fields);
+ }
+ if (type.isListType()) {
+ Types.ListType list = type.asListType();
+ String elementPath = path + ".element";
+ Type elementType = createdType(list.elementType(), elementPath, pins);
+ boolean required =
+ pins.isPinned(elementPath) || pins.pinnedColumnBeneath(elementPath) != null;
+ return required
+ ? Types.ListType.ofRequired(list.elementId(), elementType)
+ : Types.ListType.ofOptional(list.elementId(), elementType);
+ }
+ if (type.isMapType()) {
+ Types.MapType map = type.asMapType();
+ String valuePath = path + ".value";
+ Type valueType = createdType(map.valueType(), valuePath, pins);
+ boolean required = pins.isPinned(valuePath) || pins.pinnedColumnBeneath(valuePath) != null;
+ return required
+ ? Types.MapType.ofRequired(map.keyId(), map.valueId(), map.keyType(), valueType)
+ : Types.MapType.ofOptional(map.keyId(), map.valueId(), map.keyType(), valueType);
+ }
+ return type;
+ }
+
+ private static void reportIncompatible(
+ TableIdentifier tableId,
+ List<Incompatible> incompatible,
+ IncompatibleSchemaHandling handling,
+ String consequence) {
+ long files = 0;
+ for (Incompatible item : incompatible) {
+ files += item.files;
+ }
+ if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) {
+ throw new IncompatibleSchemaException(
+ "Incompatible schemas for "
+ + tableId
+ + " ("
+ + incompatible.size()
+ + " schema(s), "
+ + files
+ + " file(s)); "
+ + consequence
+ + ":\n "
+ + joinLines(incompatible));
+ }
+ LOG.warn(
+ "Skipping {} incompatible schema(s) ({} file(s)) for {}; their files will be routed to"
+ + " the error output:\n {}",
+ incompatible.size(),
+ files,
+ tableId,
+ joinLines(incompatible));
+ }
+
+ /**
+ * Stages one union per accepted schema onto {@code txn}: a scratch transaction on the evolve path
+ * (its per-schema versions stay in memory; only the folded result is ever committed), the create
+ * transaction on the create path. A schema can conflict with another schema's additions, which
+ * only surfaces while staging and poisons the transaction, so on a conflict the offender is
+ * returned for the caller to drop and retry with a fresh transaction.
+ */
+ private static @Nullable Accepted stageAll(
+ Transaction txn, List<Accepted> accepted, List<Incompatible> incompatible) {
+ for (Accepted item : accepted) {
+ // Both caught types carry staging conflicts: ValidationException from Schema
+ // construction at apply ("multiple fields for name"), IllegalArgumentException from
+ // SchemaUpdate preconditions ("Cannot change column type").
+ try {
+ stage(txn, item);
Review Comment:
Before sending to stage, should we run `ColumnNameChecks.findCaseCollisions` against `txn.table().schema().asStruct()` and each item's schema?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java:
##########
@@ -255,57 +256,302 @@ private static long commitOnce(
return table.schema().schemaId();
}
- private static final class Accepted {
- final Schema schema;
- final String json;
- final long files;
- final SchemaDelta delta;
-
- Accepted(Schema schema, String json, long files, SchemaDelta delta) {
- this.schema = schema;
- this.json = json;
- this.files = files;
- this.delta = delta;
+ /**
+ * Sorts the window's schemas into the ones the table must change for (accepted) and the ones it
+ * must not ({@code incompatible}, with the reason); schemas the table already covers drop out.
+ */
+ private static List<Accepted> classify(
+ Table table,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ List<Incompatible> incompatible) {
+ List<Accepted> accepted = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema =
+ FileSchemas.markRequired(
+ SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns());
+ SchemaDelta delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ continue;
+ }
+ if (!delta.allowedBy(config)) {
+ incompatible.add(
+ new Incompatible(
+ group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config)));
+ continue;
+ }
+ accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta));
}
+ return accepted;
}
/**
- * Folds one union per accepted schema into a scratch transaction the caller must never commit;
- * its intermediate schema versions exist only in memory. A schema can conflict with another
- * schema's additions, which only surfaces while staging and poisons the transaction, so on a
- * conflict the offender moves to {@code incompatible} and the transaction is rebuilt without it.
+ * Unions the accepted schemas into the table schema on scratch transactions that are never
+ * committed, relaxing every field the window adds; a schema that conflicts with another only
+ * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the
+ * folded schema, or null when nothing needs to change.
*/
- private static Transaction stageAll(
+ private static @Nullable Schema fold(
Table table,
Schema base,
TableIdentifier tableId,
List<Accepted> accepted,
List<Incompatible> incompatible) {
while (true) {
- Transaction txn = newTransactionOn(table, base, tableId);
- Accepted failed = null;
- for (Accepted item : accepted) {
- // Both caught types carry staging conflicts: ValidationException from Schema
- // construction at apply ("multiple fields for name"), IllegalArgumentException from
- // SchemaUpdate preconditions ("Cannot change column type").
- try {
- stage(txn, item);
- } catch (ValidationException | IllegalArgumentException e) {
- failed = item;
- incompatible.add(
- new Incompatible(
- item.json,
- item.files,
- "conflicts with another file schema in the same window: "
- + AddFiles.errorMessage(e)));
- break;
- }
+ Transaction scratch = newTransactionOn(table, base, tableId);
+ Accepted failed = stageAll(scratch, accepted, incompatible);
+ if (failed != null) {
+ accepted.remove(failed);
+ continue;
+ }
+ if (accepted.isEmpty()) {
+ return null;
}
+ relaxNewRequiredFields(scratch, base);
+ return scratch.table().schema();
+ }
+ }
+
+ /**
+ * One union replays the fold's net effect (additions, promotions, relaxations) so the table gains
+ * a single schema version instead of one per folded schema. The checkState is a pure bug
+ * detector: concurrent changes are caught earlier, by newTransactionOn.
+ */
+ private static void replay(Transaction txn, Schema merged, TableIdentifier tableId) {
+ txn.updateSchema().unionByNameWith(merged).commit();
+ // toString of the args runs only on failure
+ Schema foldResult = TypeUtil.assignIncreasingFreshIds(merged);
+ Schema replayResult = TypeUtil.assignIncreasingFreshIds(txn.table().schema());
+ checkState(
+ replayResult.sameSchema(foldResult),
+ "replaying the folded schema union for %s diverged from the fold; fold: %s replay: %s",
+ tableId,
+ foldResult,
+ replayResult);
+ }
+
+ /**
+ * Creates the table from the union of the window's schemas, with every column optional at every
+ * level so that one lucky file cannot impose required columns on the table - except pinned
+ * columns and their ancestors, which are created required.
+ */
+ private static long create(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ TableCreation creation,
+ Committer committer) {
+ if (schemas.isEmpty()) {
+ LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId);
+ return NO_TABLE;
+ }
+ List<Incompatible> incompatible = new ArrayList<>();
+ Schema merged = foldForCreate(catalog, tableId, schemas, incompatible);
+ if (!incompatible.isEmpty()) {
+ reportIncompatible(tableId, incompatible, handling, "no table was created");
+ }
+ // The real table is built from the folded result directly.
+ Schema created = createdSchema(merged, config);
+ reportUnenforceablePins(tableId, created, config, handling);
+ Map<String, String> properties =
+ creation.properties == null ? new HashMap<>() : new HashMap<>(creation.properties);
+ Transaction txn =
+ catalog
+ .buildTable(tableId, created)
+ .withPartitionSpec(PartitionUtils.toPartitionSpec(creation.partitionFields, created))
+ .withSortOrder(SortOrderUtils.toSortOrder(creation.sortFields, created))
+ .withProperties(properties)
+ .createTransaction();
+ stageNameMapping(txn);
+ committer.commit(txn);
+ Table table = catalog.loadTable(tableId);
+ LOG.info(
+ "Created table {} from {} file schema(s), schema id {}",
+ tableId,
+ schemas.size() - incompatible.size(),
+ table.schema().schemaId());
+ return table.schema().schemaId();
+ }
+
+ /**
+ * Unions the window's schemas into one on scratch create transactions that are never committed,
+ * seeded by the most common schema; conflicts move to {@code incompatible} and the fold restarts
+ * without the offender.
+ */
+ private static Schema foldForCreate(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ List<Incompatible> incompatible) {
+ Schema seed = SchemaParser.fromJson(schemas.get(0).getSchemaJson());
+ List<Accepted> rest = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas.subList(1, schemas.size())) {
+ Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson());
+ rest.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), null));
+ }
+ while (true) {
+ Transaction scratch = catalog.buildTable(tableId, seed).createTransaction();
+ Accepted failed = stageAll(scratch, rest, incompatible);
if (failed == null) {
- return txn;
+ return scratch.table().schema();
+ }
+ rest.remove(failed);
+ }
+ }
+
+ /**
+ * A pin the created schema did not end up enforcing - the column appears in no file schema, or
+ * the configured spelling resolves to a field the pin walk did not reach (a short container
+ * spelling like a.b for a.element.b, or a path inside a map key) - would stay inert forever,
+ * since later windows only add columns optional: a config error under FAIL_PIPELINE, a warning
+ * under ROUTE_TO_ERRORS (streaming may see the column later).
+ */
+ private static void reportUnenforceablePins(
+ TableIdentifier tableId,
+ Schema created,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling) {
+ List<String> unenforceable = new ArrayList<>();
+ for (String pin : config.getRequiredColumns()) {
+ Types.NestedField field = created.findField(pin);
+ if (field == null || field.isOptional()) {
+ unenforceable.add(pin);
+ }
+ }
+ if (unenforceable.isEmpty()) {
+ return;
+ }
+ Collections.sort(unenforceable);
+ if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) {
+ throw new IncompatibleSchemaException(
+ "Pinned column(s) "
+ + unenforceable
+ + " appear in none of the file schemas creating "
+ + tableId
+ + ", or their spelling does not match the column path; the created table cannot"
+ + " make them required");
+ }
+ LOG.warn(
+ "Pinned column(s) {} appear in none of the file schemas creating {}, or their spelling"
+ + " does not match the column path; the created table cannot make them required",
+ unenforceable,
+ tableId);
+ }
+
+ /**
+ * The created schema: every field optional at every level, list elements and map values included,
+ * except pinned paths and their ancestors, which stay required so the schema advertises the
+ * guarantee the per-file pin check enforces (a null ancestor nulls the pinned leaf). Map key
+ * subtrees keep their declared shape (keys are required by definition; pins inside them are not
+ * honored). Nothing depends on a created table's schema yet, so this is the schema-authoring
+ * moment; evolution never tightens columns afterwards.
+ */
+ static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) {
+ Pins pins = new Pins(config.getRequiredColumns());
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : merged.asStruct().fields()) {
+ fields.add(createdField(field, field.name(), pins));
+ }
+ return new Schema(fields);
+ }
+
+ private static Types.NestedField createdField(Types.NestedField field, String path, Pins pins) {
+ boolean required = pins.isPinned(path) || pins.pinnedColumnBeneath(path) != null;
Review Comment:
nit: This `pins.isPinned || pins.pinnedColumnBeneath` line is repeated multiple times. Make it a method in Pins?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java:
##########
@@ -255,57 +256,302 @@ private static long commitOnce(
return table.schema().schemaId();
}
- private static final class Accepted {
- final Schema schema;
- final String json;
- final long files;
- final SchemaDelta delta;
-
- Accepted(Schema schema, String json, long files, SchemaDelta delta) {
- this.schema = schema;
- this.json = json;
- this.files = files;
- this.delta = delta;
+ /**
+ * Sorts the window's schemas into the ones the table must change for (accepted) and the ones it
+ * must not ({@code incompatible}, with the reason); schemas the table already covers drop out.
+ */
+ private static List<Accepted> classify(
+ Table table,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ List<Incompatible> incompatible) {
+ List<Accepted> accepted = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema =
+ FileSchemas.markRequired(
+ SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns());
+ SchemaDelta delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ continue;
+ }
+ if (!delta.allowedBy(config)) {
+ incompatible.add(
+ new Incompatible(
+ group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config)));
+ continue;
+ }
+ accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta));
}
+ return accepted;
}
/**
- * Folds one union per accepted schema into a scratch transaction the caller must never commit;
- * its intermediate schema versions exist only in memory. A schema can conflict with another
- * schema's additions, which only surfaces while staging and poisons the transaction, so on a
- * conflict the offender moves to {@code incompatible} and the transaction is rebuilt without it.
+ * Unions the accepted schemas into the table schema on scratch transactions that are never
+ * committed, relaxing every field the window adds; a schema that conflicts with another only
+ * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the
+ * folded schema, or null when nothing needs to change.
*/
- private static Transaction stageAll(
+ private static @Nullable Schema fold(
Table table,
Schema base,
TableIdentifier tableId,
List<Accepted> accepted,
List<Incompatible> incompatible) {
while (true) {
- Transaction txn = newTransactionOn(table, base, tableId);
- Accepted failed = null;
- for (Accepted item : accepted) {
- // Both caught types carry staging conflicts: ValidationException from Schema
- // construction at apply ("multiple fields for name"), IllegalArgumentException from
- // SchemaUpdate preconditions ("Cannot change column type").
- try {
- stage(txn, item);
- } catch (ValidationException | IllegalArgumentException e) {
- failed = item;
- incompatible.add(
- new Incompatible(
- item.json,
- item.files,
- "conflicts with another file schema in the same window: "
- + AddFiles.errorMessage(e)));
- break;
- }
+ Transaction scratch = newTransactionOn(table, base, tableId);
+ Accepted failed = stageAll(scratch, accepted, incompatible);
+ if (failed != null) {
+ accepted.remove(failed);
+ continue;
+ }
+ if (accepted.isEmpty()) {
+ return null;
}
+ relaxNewRequiredFields(scratch, base);
+ return scratch.table().schema();
+ }
+ }
+
+ /**
+ * One union replays the fold's net effect (additions, promotions, relaxations) so the table gains
+ * a single schema version instead of one per folded schema. The checkState is a pure bug
+ * detector: concurrent changes are caught earlier, by newTransactionOn.
+ */
+ private static void replay(Transaction txn, Schema merged, TableIdentifier tableId) {
+ txn.updateSchema().unionByNameWith(merged).commit();
+ // toString of the args runs only on failure
+ Schema foldResult = TypeUtil.assignIncreasingFreshIds(merged);
+ Schema replayResult = TypeUtil.assignIncreasingFreshIds(txn.table().schema());
+ checkState(
+ replayResult.sameSchema(foldResult),
+ "replaying the folded schema union for %s diverged from the fold; fold: %s replay: %s",
+ tableId,
+ foldResult,
+ replayResult);
+ }
+
+ /**
+ * Creates the table from the union of the window's schemas, with every column optional at every
+ * level so that one lucky file cannot impose required columns on the table - except pinned
+ * columns and their ancestors, which are created required.
+ */
+ private static long create(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ TableCreation creation,
+ Committer committer) {
+ if (schemas.isEmpty()) {
+ LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId);
+ return NO_TABLE;
+ }
+ List<Incompatible> incompatible = new ArrayList<>();
+ Schema merged = foldForCreate(catalog, tableId, schemas, incompatible);
+ if (!incompatible.isEmpty()) {
+ reportIncompatible(tableId, incompatible, handling, "no table was created");
+ }
+ // The real table is built from the folded result directly.
+ Schema created = createdSchema(merged, config);
+ reportUnenforceablePins(tableId, created, config, handling);
+ Map<String, String> properties =
+ creation.properties == null ? new HashMap<>() : new HashMap<>(creation.properties);
+ Transaction txn =
+ catalog
+ .buildTable(tableId, created)
+ .withPartitionSpec(PartitionUtils.toPartitionSpec(creation.partitionFields, created))
+ .withSortOrder(SortOrderUtils.toSortOrder(creation.sortFields, created))
+ .withProperties(properties)
+ .createTransaction();
+ stageNameMapping(txn);
+ committer.commit(txn);
+ Table table = catalog.loadTable(tableId);
+ LOG.info(
+ "Created table {} from {} file schema(s), schema id {}",
+ tableId,
+ schemas.size() - incompatible.size(),
+ table.schema().schemaId());
+ return table.schema().schemaId();
+ }
+
+ /**
+ * Unions the window's schemas into one on scratch create transactions that are never committed,
+ * seeded by the most common schema; conflicts move to {@code incompatible} and the fold restarts
+ * without the offender.
+ */
+ private static Schema foldForCreate(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ List<Incompatible> incompatible) {
Review Comment:
Should we run `ColumnNameChecks.findInvalidNames()` on these file schemas first?
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java:
##########
@@ -759,10 +774,386 @@ public void testPersistentCommitFailurePropagates() {
Arrays.asList(files(file, 1)),
ALL,
IncompatibleSchemaHandling.FAIL_PIPELINE,
+ NO_CREATION,
alwaysFails));
assertEquals(CommitSchemaUnion.MAX_ATTEMPTS, attempts.get());
}
+ // ---- create path
+
+ private TableIdentifier missing() {
+ return TableIdentifier.of("default", testName.getMethodName() + "_new");
+ }
+
+ private long commitTo(
+ TableIdentifier id,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ CommitSchemaUnion.TableCreation creation,
+ CollectDistinctSchemas.SchemaGroup... schemas) {
+ return CommitSchemaUnion.commit(
+ catalog,
+ id,
+ Arrays.asList(schemas),
+ config,
+ handling,
+ creation,
+ CommitSchemaUnion.DEFAULT_COMMITTER);
+ }
+
+ @Test
+ public void testMissingTableIsCreatedFromTheUnion() {
+ TableIdentifier id = missing();
+ Schema seed =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "email", Types.StringType.get()));
+ Schema other =
+ new Schema(
+ required(1, "id", Types.LongType.get()), optional(2, "extra", Types.LongType.get()));
+ CommitSchemaUnion.TableCreation creation =
+ new CommitSchemaUnion.TableCreation(
+ Arrays.asList("region"), null, java.util.Collections.singletonMap("k", "v"));
+ long schemaId =
+ commitTo(
+ id,
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ creation,
+ files(seed, 5),
+ files(other, 1));
+ Table table = catalog.loadTable(id);
+ assertEquals(table.schema().schemaId(), schemaId);
+ // canonical (sorted) seed columns first, the union's addition last
+ assertSameSchema(
+ new Schema(
+ optional(1, "email", Types.StringType.get()),
+ optional(2, "id", Types.LongType.get()),
+ optional(3, "region", Types.StringType.get()),
+ optional(4, "extra", Types.LongType.get())),
+ table.schema());
Review Comment:
Not sure if it's documented yet, but would be important to note that a newly created table will have lexicographically sorted column names
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java:
##########
@@ -255,57 +256,302 @@ private static long commitOnce(
return table.schema().schemaId();
Review Comment:
nit: also can skip `table.refresh();` and just use `txn.table().schema().schemaId()` here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: create the table from the file schemas when it does not exist [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on PR #40124:
URL: https://github.com/apache/beam/pull/40124#issuecomment-5687752842
Test failures are unrelated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: create the table from the file schemas when it does not exist [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on PR #40124:
URL: https://github.com/apache/beam/pull/40124#issuecomment-5681808973
Failing test is unrelated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: create the table from the file schemas when it does not exist [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40124:
URL: https://github.com/apache/beam/pull/40124#issuecomment-5674498524
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org