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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
