ahmedabu98 commented on code in PR #40150:
URL: https://github.com/apache/beam/pull/40150#discussion_r4039034453
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java:
##########
@@ -115,6 +119,48 @@ public static Builder builder() {
+ "For more information on sort orders, please visit
https://iceberg.apache.org/spec/#sort-orders.";)
public abstract @Nullable List<String> getSortFields();
+ @SchemaFieldDescription(
+ "Schema changes the transform may make so that every file's columns
are covered by the"
+ + " table schema, from: ALLOW_FIELD_ADDITION (add file columns
absent from the table,"
+ + " as optional), ALLOW_FIELD_RELAXATION (make a required table
column optional when"
+ + " a file may contain nulls in it or lacks it),
ALLOW_TYPE_PROMOTION (widen a column"
+ + " type, e.g. int to long). Empty or absent: the table schema is
never changed."
+ + " With any option set, every Parquet file's footer is read and
the allowed changes"
+ + " are committed before files are registered, so every registered
file carries stats"
+ + " for every column it has. A file needing a change that is not
allowed is"
+ + " incompatible: see incompatible_schema_handling. Only Parquet
files can be checked:"
+ + " with options set, ORC and Avro files are routed to the error
output unless"
+ + " unverifiable_file_handling is ACCEPT. Routed files reach the
error output only"
+ + " when error_handling is set. Requires a batch pipeline;
streaming with schema"
+ + " evolution is not yet supported.")
+ public abstract @Nullable List<String> getSchemaEvolutionOptions();
+
+ @SchemaFieldDescription(
+ "Columns (dotted paths for nested fields) that must stay required
whatever the options"
+ + " say. A file that lacks such a column or holds nulls in it is
routed to the error"
+ + " output (see error_handling); so is one whose footer has no
null-count statistics"
Review Comment:
```suggestion
+ " output (see error_handling); so is one whose footer marks it
as optional and has no null-count statistics"
```
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java:
##########
@@ -112,8 +117,38 @@
import org.slf4j.LoggerFactory;
/**
- * A transform that takes in a stream of file paths, converts them to Iceberg
{@link DataFile}s with
- * partition metadata and metrics, then commits them to an Iceberg {@link
Table}.
+ * Registers existing Parquet, ORC or Avro files in an Iceberg table without
rewriting them: each
+ * path becomes a {@link DataFile} with partition metadata and column stats,
batched into manifests
+ * and committed as snapshots.
+ *
+ * <p>Outputs: {@code snapshots} (one row per commit), {@code errors} (one row
per file that could
+ * not be registered: {@code file}, {@code error}).
+ *
+ * <p><b>Schema evolution.</b> With a {@link SchemaEvolutionConfig} whose
options are set, a
+ * pre-pass reads every Parquet footer, classifies the change each distinct
file schema needs on the
+ * table (add a column, relax a required column, promote a type), commits the
allowed changes in one
+ * transaction, and only then registers the files. Manifest entries are
immutable, so this ordering
+ * is what guarantees that every registered file carries stats for every
column it has. Files whose
+ * schema needs a change that is not allowed, or that conflicts with the table
or with another file,
+ * are incompatible: by default the pipeline fails before committing anything,
or routes them to
+ * {@code errors} (see {@link
SchemaEvolutionConfig.IncompatibleSchemaHandling}). The per-file
+ * checks read Parquet footers: an ORC or Avro file, or a pinned column the
footer has no null count
+ * for, cannot be verified and goes to {@code errors} unless {@link
+ * SchemaEvolutionConfig.UnverifiableFileHandling#ACCEPT} registers it on
trust. Schema evolution
+ * currently requires bounded input; unbounded input with options set is
rejected at construction.
+ *
+ * <pre>{@code
+ * SchemaEvolutionConfig evolution =
+ * SchemaEvolutionConfig.builder()
+ * .setOptions(EnumSet.of(ALLOW_FIELD_ADDITION, ALLOW_TYPE_PROMOTION))
+ * .setRequiredColumns(Collections.singleton("id"))
+ * .build();
+ * paths.apply(new AddFiles(catalog, "db.sales", null, null, null, null, null,
null, evolution));
+ * }</pre>
+ *
+ * <p>Without options the table schema is never changed and files register
as-is: columns the table
+ * does not have get no stats and are not readable, and a nested column the
table does not know can
+ * make that file, and any scan that includes it, fail in Iceberg's reader.
Review Comment:
```suggestion
* make that file, and any scan that includes it, fail in an Iceberg reader.
```
--
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]
