This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 706c76dc613 AddFiles: per-file coverage check and pinned-column
enforcement in ConvertToDataFile (#40143)
706c76dc613 is described below
commit 706c76dc6132b156e47bd6ab9bcc3169c5ea928f
Author: claudevdm <[email protected]>
AuthorDate: Thu Sep 17 07:23:31 2026 -0400
AddFiles: per-file coverage check and pinned-column enforcement in
ConvertToDataFile (#40143)
* AddFiles: per-file coverage check and pinned-column enforcement in
ConvertToDataFile
ConvertToDataFile gains a SchemaEvolutionConfig.
With options set, every Parquet file is checked against the table
before its DataFile is built, and every failure is one row on the
error output:
1. FileSchemas.effective(footer) (convert plus tighten, shared with the
read side so both sides see the same schema) is computed. A footer
whose schema cannot be converted (unannotated repeated leaf, legacy
MAP_KEY_VALUE, uint64, ...) yields "Could not read the file's
schema: <cause>".
2. Coverage: SchemaDelta.classify against the cached table. The
pre-pass commits the schema before paths reach this DoFn, so the
delta is expected to be empty. If it is not, the table is refreshed once
and classified again. A remaining delta is reported as "Table schema
does
not cover the file after refresh: <reason>", where the reason is the
delta's disallowed reason when the options forbid the change (the
ROUTE_TO_ERRORS path: this is where an incompatible file lands) or
"changes not applied: ..." when the change was allowed but somehow not
committed
(defensive catch all).
3. Pins: for each pinned column the table has, the file must contain
it and be provably null-free: the footer's own null count for the
column (FileSchemas.nullCount, summed over row groups) must be
present and zero. A missing count is a violation too: "cannot prove"
is not "proven".
Messages: "Pinned required column X is absent from the file" / "has no
null
count statistics in the file" / "has N null(s) in the file".
The pin evidence deliberately does NOT come from the Metrics object
built for the DataFile: the table's write.metadata.metrics
configuration shapes those (mode none, or the inferred-column cap on
wide schemas, drops null counts entirely), and pin enforcement must
not be configurable away. Two tests pin this down under
write.metadata.metrics.default=none: a null-free pinned file still
registers and a file with nulls is still caught.
With evolution enabled, a non-Parquet file (ORC, Avro) is routed to
the error output instead of registering unchecked: none of the checks
above can read those formats, and silently registering a file the
options promised to verify would be a hole in the guarantee. With
evolution disabled they register exactly as before.
SchemaEvolutionConfig.UnverifiableFileHandling relaxes exactly the
"cannot verify" cases, never a failed check: REJECT (default) routes
them to the error output as above; ACCEPT registers a non-Parquet file
unchecked, and a Parquet file whose footer has no null count for a
pinned column on trust (a writer with statistics disabled, or a pin
under a list or map, whose physical chunk path is never mapped). An
absent pinned column and a counted null stay violations under ACCEPT,
and the pin walk continues past an unproven pin so a counted one still
fails the file.
* comments
* spotbugs
---
.../IO_Iceberg_Integration_Tests.json | 2 +-
.../org/apache/beam/sdk/io/iceberg/AddFiles.java | 218 ++++++-
.../apache/beam/sdk/io/iceberg/FileSchemas.java | 115 +++-
.../beam/sdk/io/iceberg/SchemaEvolutionConfig.java | 46 +-
.../apache/beam/sdk/io/iceberg/AddFilesTest.java | 681 +++++++++++++++++++++
.../beam/sdk/io/iceberg/FileSchemasTest.java | 95 +++
.../sdk/io/iceberg/SchemaEvolutionConfigTest.java | 22 +
7 files changed, 1152 insertions(+), 27 deletions(-)
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index 5d04b2c0a8c..89e73b29da0 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": 5
+ "modification": 6
}
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 8254b2fc07d..daf61bf883f 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
@@ -34,13 +34,16 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
+import
org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.UnverifiableFileHandling;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.SchemaCoder;
@@ -120,6 +123,11 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
counter(AddFiles.class, "numManifestFilesAdded");
private static final Counter numDataFilesAdded = counter(AddFiles.class,
"numDataFilesAdded");
private static final Counter numErrorFiles = counter(AddFiles.class,
"numErrorFiles");
+ static final String UNCHECKED_FORMAT_COUNTER = "numUncheckedFormatFiles";
+ static final String UNPROVEN_PINS_COUNTER = "numUnprovenPinFiles";
+ private static final Counter numUncheckedFormatFiles =
+ counter(AddFiles.class, UNCHECKED_FORMAT_COUNTER);
+ private static final Counter numUnprovenPinFiles = counter(AddFiles.class,
UNPROVEN_PINS_COUNTER);
private static final Logger LOG = LoggerFactory.getLogger(AddFiles.class);
private static final int DEFAULT_DATAFILES_PER_MANIFEST = 10_000;
private static final int DEFAULT_MAX_MANIFESTS_PER_SNAPSHOT = 100;
@@ -260,8 +268,10 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
private final @Nullable List<String> partitionFields;
private final @Nullable List<String> sortFields;
private final @Nullable Map<String, String> tableProps;
+ private final SchemaEvolutionConfig evolution;
private transient @MonotonicNonNull BoundedAsyncTasks<ProcessResult> tasks;
private transient volatile @MonotonicNonNull Table table;
+ private transient @MonotonicNonNull Set<String> warned;
// Number of parallel threads processing incoming files
private static final int THREAD_POOL_SIZE = 10;
@@ -274,21 +284,80 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
@Nullable List<String> partitionFields,
@Nullable List<String> sortFields,
@Nullable Map<String, String> tableProps) {
+ this(
+ catalogConfig,
+ identifier,
+ prefix,
+ partitionFields,
+ sortFields,
+ tableProps,
+ SchemaEvolutionConfig.disabled());
+ }
+
+ public ConvertToDataFile(
+ IcebergCatalogConfig catalogConfig,
+ String identifier,
+ @Nullable String prefix,
+ @Nullable List<String> partitionFields,
+ @Nullable List<String> sortFields,
+ @Nullable Map<String, String> tableProps,
+ SchemaEvolutionConfig evolution) {
this.catalogConfig = catalogConfig;
this.identifier = identifier;
this.prefix = prefix;
this.partitionFields = partitionFields;
this.sortFields = sortFields;
this.tableProps = tableProps;
+ this.evolution = evolution;
}
static final String PREFIX_ERROR = "File path did not start with the
specified prefix";
private static final String UNKNOWN_FORMAT_ERROR = "Could not determine
the file's format";
static final String UNKNOWN_PARTITION_ERROR = "Could not determine the
file's partition: ";
+ static final String UNREADABLE_SCHEMA_ERROR = "Could not read the file's
schema: ";
+ static final String UNCOVERED_ERROR = "Table schema does not cover the
file after refresh: ";
+ static final String PINNED_COLUMN_ERROR = "Pinned required column ";
+ static final String UNCHECKED_FORMAT_ERROR =
+ "Schema evolution is enabled but coverage and pin checks support only
Parquet;"
+ + " refusing to register an unchecked file of format ";
+
+ /**
+ * What a file registered under {@link UnverifiableFileHandling#ACCEPT}
could not be checked
+ * for.
+ */
+ enum Unverified {
+ FORMAT,
+ PIN_STATISTICS
+ }
+
+ /** Verdict of the per-file checks: an error, or none plus what was left
unverified. */
+ private static final class Verdict {
+ static final Verdict OK = new Verdict(null, null);
+
+ final @Nullable String error;
+ final @Nullable Unverified unverified;
+
+ private Verdict(@Nullable String error, @Nullable Unverified unverified)
{
+ this.error = error;
+ this.unverified = unverified;
+ }
+
+ static Verdict error(String message) {
+ return new Verdict(message, null);
+ }
+
+ static Verdict unverified(Unverified what) {
+ return new Verdict(null, what);
+ }
+ }
private static class ProcessResult {
final @Nullable SerializableDataFile dataFile;
final @Nullable Row errorRow;
+
+ /** Counted on the processing thread: metrics touched from the executor
are lost. */
+ final @Nullable Unverified unverified;
+
final Instant timestamp;
final BoundedWindow window;
final PaneInfo paneInfo;
@@ -296,6 +365,7 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
ProcessResult(
@Nullable SerializableDataFile dataFile,
@Nullable Row errorRow,
+ @Nullable Unverified unverified,
Instant timestamp,
BoundedWindow window,
PaneInfo paneInfo) {
@@ -306,6 +376,7 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
errorRow);
this.dataFile = dataFile;
this.errorRow = errorRow;
+ this.unverified = unverified;
this.timestamp = timestamp;
this.window = window;
this.paneInfo = paneInfo;
@@ -315,6 +386,7 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
@Setup
public void setup() {
tasks = new BoundedAsyncTasks<>(THREAD_POOL_SIZE, MAX_IN_FLIGHT_TASKS);
+ warned = ConcurrentHashMap.newKeySet();
}
/** Clears anything left behind if the runner reuses this instance after a
failed bundle. */
@@ -361,6 +433,7 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
result.timestamp,
Collections.singleton(result.window),
result.paneInfo);
+ countUnverified(result);
}
}
@@ -375,6 +448,15 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
numErrorFiles.inc();
} else if (result.dataFile != null) {
context.output(DATA_FILES, result.dataFile, result.timestamp,
result.window);
+ countUnverified(result);
+ }
+ }
+
+ private static void countUnverified(ProcessResult result) {
+ if (result.unverified == Unverified.FORMAT) {
+ numUncheckedFormatFiles.inc();
+ } else if (result.unverified == Unverified.PIN_STATISTICS) {
+ numUnprovenPinFiles.inc();
}
}
@@ -411,6 +493,13 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
return errorResult(filePath, PREFIX_ERROR, timestamp, window,
paneInfo);
}
+ if (table.schema().columns().isEmpty() && firstTime("empty schema")) {
+ LOG.warn(
+ "Table {} has no columns: files register with no readable
columns and no stats."
+ + " Enable schema evolution to infer the schema from the
files.",
+ identifier);
+ }
+
// ---- Per-file phase: every failure below is one error row, never a
failed bundle.
@Nullable ParquetMetadata parquetFooter = null;
if (format.equals(FileFormat.PARQUET)) {
@@ -420,6 +509,13 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
return errorResult(filePath, errorMessage(e), timestamp, window,
paneInfo);
}
}
+ Verdict verdict = Verdict.OK;
+ if (evolution.isEnabled()) {
+ verdict = verify(filePath, format, parquetFooter);
+ }
+ if (verdict.error != null) {
+ return errorResult(filePath, verdict.error, timestamp, window,
paneInfo);
+ }
InputFile inputFile = table.io().newInputFile(filePath);
@@ -464,7 +560,12 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
.withPartitionPath(partitionPath)
.build();
return new ProcessResult(
- SerializableDataFile.from(df, table.spec()), null, timestamp,
window, paneInfo);
+ SerializableDataFile.from(df, table.spec()),
+ null,
+ verdict.unverified,
+ timestamp,
+ window,
+ paneInfo);
} catch (Exception e) {
// getLength is a per-file read (e.g. the file was deleted
mid-flight).
return errorResult(filePath, errorMessage(e), timestamp, window,
paneInfo);
@@ -472,11 +573,126 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
};
}
+ /**
+ * The checks the options promise, in order: a format the checks can read,
a convertible schema,
+ * coverage by the table, pins. The first failure is the verdict.
+ */
+ private Verdict verify(String filePath, FileFormat format, @Nullable
ParquetMetadata footer) {
+ if (!format.equals(FileFormat.PARQUET)) {
+ if (evolution.getUnverifiableFileHandling() ==
UnverifiableFileHandling.REJECT) {
+ return Verdict.error(UNCHECKED_FORMAT_ERROR + format.name());
+ }
+ if (firstTime("unchecked format")) {
+ LOG.warn(
+ "Registering {} files in table {} unchecked
(UnverifiableFileHandling.ACCEPT):"
+ + " coverage and pin checks read only Parquet, so a required
column such a file"
+ + " lacks or holds nulls in fails reads of the table, not
registration. First"
+ + " file: {}",
+ format,
+ identifier,
+ filePath);
+ }
+ return Verdict.unverified(Unverified.FORMAT);
+ }
+ ParquetMetadata parquetFooter = checkStateNotNull(footer, "Parquet
checks need the footer");
+ org.apache.iceberg.Schema fileSchema;
+ try {
+ fileSchema = FileSchemas.effective(parquetFooter);
+ } catch (Exception e) {
+ return Verdict.error(UNREADABLE_SCHEMA_ERROR + errorMessage(e));
+ }
+ @Nullable String uncovered = uncoveredReason(fileSchema);
+ if (uncovered != null) {
+ return Verdict.error(uncovered);
+ }
+ return checkPins(filePath, fileSchema, parquetFooter);
+ }
+
+ /**
+ * The pre-pass commits the schema before paths reach this stage, so the
cached table normally
+ * covers every file. If not, refresh once (a commit may have landed since
the table was cached)
+ * and report the remaining delta. Never changes the schema.
+ */
+ private @Nullable String uncoveredReason(org.apache.iceberg.Schema
fileSchema) {
+ Table table = checkStateNotNull(this.table);
+ SchemaDelta delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ return null;
+ }
+ synchronized (this) {
+ table.refresh();
+ }
+ delta = SchemaDelta.classify(table, fileSchema);
+ if (delta.isEmpty()) {
+ return null;
+ }
+ String reason = delta.disallowedReason(evolution);
+ if (reason.isEmpty()) {
+ reason = "changes not applied: " + String.join("; ",
delta.descriptions());
+ }
+ return UNCOVERED_ERROR + reason;
+ }
+
+ /**
+ * A pinned column must be present and provably null-free; a zero-row file
is vacuously fine.
+ * The evidence is the footer's own null counts read by the tighten rules
({@link
+ * FileSchemas#nullCount}), never the Metrics built for the DataFile: the
table's
+ * write.metadata.metrics configuration shapes those (mode none, or the
inferred-column cap on
+ * wide schemas, drops the counts) and must not be able to turn pin
enforcement off. A pin with
+ * no count is a violation under REJECT; under ACCEPT it is recorded as
unproven and the walk
+ * goes on, so a pin the footer does count nulls for still fails the file.
+ */
+ private Verdict checkPins(
+ String filePath, org.apache.iceberg.Schema fileSchema, ParquetMetadata
footer) {
+ Table table = checkStateNotNull(this.table);
+ List<String> unproven = new ArrayList<>();
+ for (String pinned : evolution.getRequiredColumns()) {
+ if (table.schema().findField(pinned) == null) {
+ continue;
+ }
+ if (fileSchema.findField(pinned) == null) {
+ return Verdict.error(PINNED_COLUMN_ERROR + pinned + " is absent from
the file");
+ }
+ @Nullable Long nulls = FileSchemas.nullCount(footer, fileSchema,
pinned);
+ if (nulls == null) {
+ if (evolution.getUnverifiableFileHandling() ==
UnverifiableFileHandling.REJECT) {
+ return Verdict.error(
+ PINNED_COLUMN_ERROR + pinned + " has no null count statistics
in the file");
+ }
+ unproven.add(pinned);
+ continue;
+ }
+ if (nulls > 0) {
+ return Verdict.error(
+ PINNED_COLUMN_ERROR + pinned + " has " + nulls + " null(s) in
the file");
+ }
+ }
+ if (unproven.isEmpty()) {
+ return Verdict.OK;
+ }
+ if (firstTime("unproven pins")) {
+ LOG.warn(
+ "Registering files in table {} whose footer has no null count
statistics for pinned"
+ + " column(s) {} on trust (UnverifiableFileHandling.ACCEPT):
nulls there fail"
+ + " reads of the table, not registration. First file: {}",
+ identifier,
+ unproven,
+ filePath);
+ }
+ return Verdict.unverified(Unverified.PIN_STATISTICS);
+ }
+
+ /** Once per instance per key: at volume a line per file would drown the
log. */
+ private boolean firstTime(String key) {
+ return checkStateNotNull(warned).add(key);
+ }
+
private static ProcessResult errorResult(
String filePath, String message, Instant timestamp, BoundedWindow
window, PaneInfo pane) {
return new ProcessResult(
null,
Row.withSchema(ERROR_SCHEMA).addValues(filePath, message).build(),
+ null,
timestamp,
window,
pane);
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java
index d95321153d5..79f87aa2fba 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java
@@ -21,8 +21,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
@@ -34,6 +36,7 @@ import org.apache.parquet.column.statistics.Statistics;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.checkerframework.checker.nullness.qual.Nullable;
/**
* What a file contributes to schema inference: the canonical form of the
schema it declares, and
@@ -66,6 +69,51 @@ final class FileSchemas {
SchemaParser.toJson(canonical(converted)), 1,
changedToRequired(converted, tightened));
}
+ /**
+ * This one file's schema in table terms: convert, then tighten using its
own footer. Uses the
+ * file's own null evidence, not its group's: a file that proves a column
null-free registers even
+ * when other files sharing its schema could not prove it.
+ */
+ static Schema effective(ParquetMetadata footer) {
+ Schema converted =
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
+ return tighten(converted, footer);
+ }
+
+ /**
+ * The file's own null count for one dotted column of its schema, or null
when the file cannot
+ * prove it. Pin evidence must come from the file itself, not from the
Metrics built for the
+ * DataFile: the table's write.metadata.metrics configuration shapes those
and must not be able to
+ * turn pin enforcement off. A column the schema declares required, with
every ancestor required,
+ * counts 0 without any statistic: Parquet's repetition is structural, so
such a column cannot
+ * encode a null. Otherwise the footer's counts are read by the {@link
#tighten} rules: a leaf's
+ * count is summed over row groups and unknown when any non-empty row group
lacks it; a struct
+ * counts 0 when a leaf beneath it (struct nesting only) has none, since a
null struct nulls every
+ * leaf, and is unknown otherwise, since a leaf's nulls include its
ancestors' and cannot be
+ * attributed; a column under a list or map is unknown; a file with no rows
proves every column.
+ */
+ static @Nullable Long nullCount(ParquetMetadata footer, Schema fileSchema,
String column) {
+ if (rowCount(footer) == 0) {
+ return 0L;
+ }
+ Types.NestedField field = fileSchema.findField(column);
+ if (field == null) {
+ return null;
+ }
+ if (requiredAlongPath(fileSchema, column)) {
+ return 0L;
+ }
+ List<String> path = new ArrayList<>(Arrays.asList(column.split("\\.",
-1)));
+ Map<List<String>, @Nullable Long> counts = nullCountsByLeaf(footer);
+ if (field.type().isPrimitiveType()) {
+ return counts.get(path);
+ }
+ if (field.type().isStructType()) {
+ Tightened struct = tightenStruct(field.type().asStructType(), path,
zeroNullLeaves(counts));
+ return struct.hasNullFreeLeaf ? Long.valueOf(0) : null;
+ }
+ return null;
+ }
+
/**
* Marks a declared-optional column required when every row group has a null
count of zero for it,
* so the file does not request a relaxation it does not need; an absent
count is not proof. A
@@ -164,33 +212,68 @@ final class FileSchemas {
return rows;
}
+ /** Whether the column and each of its ancestors are required: no null can
be encoded there. */
+ private static boolean requiredAlongPath(Schema schema, String column) {
+ Types.NestedField field = schema.findField(column);
+ if (field == null) {
+ return false;
+ }
+ Map<Integer, Integer> parents = TypeUtil.indexParents(schema.asStruct());
+ for (@Nullable Integer id = field.fieldId(); id != null; id =
parents.get(id)) {
+ if (schema.findField(id).isOptional()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /** Leaf paths proven null-free: the one source of null evidence for tighten
and for pins. */
+ private static Set<List<String>> leafPathsWithZeroNullCounts(ParquetMetadata
footer) {
+ return zeroNullLeaves(nullCountsByLeaf(footer));
+ }
+
+ private static Set<List<String>> zeroNullLeaves(Map<List<String>, @Nullable
Long> counts) {
+ Set<List<String>> proven = new HashSet<>();
+ for (Map.Entry<List<String>, @Nullable Long> entry : counts.entrySet()) {
+ Long count = entry.getValue();
+ if (count != null && count == 0) {
+ proven.add(entry.getKey());
+ }
+ }
+ return proven;
+ }
+
/**
- * Leaf paths proven null-free in every row group that has rows
(intersection over blocks). An
- * empty row group holds no nulls whatever its statistics say, so it
constrains nothing.
+ * Null count per leaf chunk path, summed over the row groups that have
rows; null where any such
+ * row group lacks the statistic (an absent count proves nothing). An empty
row group holds no
+ * nulls whatever its statistics say, so it constrains nothing.
*/
- private static Set<List<String>> leafPathsWithZeroNullCounts(ParquetMetadata
footer) {
- Set<List<String>> proven = null;
+ private static Map<List<String>, @Nullable Long>
nullCountsByLeaf(ParquetMetadata footer) {
+ Map<List<String>, @Nullable Long> counts = new HashMap<>();
for (BlockMetaData block : footer.getBlocks()) {
if (block.getRowCount() == 0) {
continue;
}
- Set<List<String>> provenHere = new HashSet<>();
for (ColumnChunkMetaData chunk : block.getColumns()) {
+ List<String> path = Arrays.asList(chunk.getPath().toArray());
Statistics<?> stats = chunk.getStatistics();
- if (stats != null && stats.isNumNullsSet() && stats.getNumNulls() ==
0) {
- provenHere.add(Arrays.asList(chunk.getPath().toArray()));
+ Long inChunk = null;
+ if (stats != null && stats.isNumNullsSet()) {
+ inChunk = stats.getNumNulls();
+ }
+ if (!counts.containsKey(path)) {
+ counts.put(path, inChunk);
+ continue;
+ }
+ Long soFar = counts.get(path);
+ if (soFar == null || inChunk == null) {
+ counts.put(path, null);
+ } else {
+ counts.put(path, soFar + inChunk);
}
- }
- if (proven == null) {
- proven = provenHere;
- } else {
- proven.retainAll(provenHere);
}
}
- if (proven == null) {
- return new HashSet<>();
- }
- return proven;
+ return counts;
}
private static final class Tightened {
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java
index 492b1cdad49..6b67b390ae5 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java
@@ -38,13 +38,18 @@ import org.checkerframework.checker.nullness.qual.Nullable;
* }</pre>
*
* <p><b>Pins.</b> Required columns are pinned: never made optional whatever
the options say, and
- * created required when this transform creates the table. A Parquet file that
lacks a pinned
- * column, has nulls in it, or carries no null-count statistics for it is
routed to the error
- * output; ORC and Avro files are not checked. Pins name canonical (table)
paths, dotted for nested
- * fields, with the container segment spelled out under lists and maps ({@code
+ * created required when this transform creates the table. A Parquet file that
lacks a pinned column
+ * or has nulls in it is routed to the error output. Pins name canonical
(table) paths, dotted for
+ * nested fields, with the container segment spelled out under lists and maps
({@code
* addresses.element.city}, {@code attributes.value.total}). A top-level
column whose own name
* contains a dot cannot be pinned.
*
+ * <p><b>Unverifiable files.</b> The per-file checks read Parquet footers. An
ORC or Avro file
+ * cannot be checked at all, and a Parquet file whose footer carries no
null-count statistics for a
+ * pinned column (a writer with statistics disabled, or a pin under a list or
map, whose physical
+ * chunk path the check does not map) cannot prove the pin. {@link
UnverifiableFileHandling} decides
+ * whether such a file is routed to the error output (the default) or
registered on trust.
+ *
* <p><b>Incompatible schemas.</b> A schema that needs a change the options do
not allow, or that
* conflicts with the table or with another file's schema. {@link
IncompatibleSchemaHandling}
* decides whether that fails the pipeline before any schema commit (the batch
default) or skips the
@@ -69,6 +74,23 @@ public abstract class SchemaEvolutionConfig implements
Serializable {
ROUTE_TO_ERRORS
}
+ /**
+ * What to do with a file the per-file checks cannot verify: a non-Parquet
file, or a Parquet file
+ * with no null-count statistics for a pinned column. A file that fails a
check is always routed
+ * to the error output.
+ */
+ public enum UnverifiableFileHandling {
+ /** Route the file to the error output. The default: "cannot prove" is not
"proven". */
+ REJECT,
+ /**
+ * Register the file unchecked, counted ({@code numUncheckedFormatFiles},
{@code
+ * numUnprovenPinFiles}) and logged. A trusted file that lacks a required
column or holds nulls
+ * in one breaks reads of the table at query time, not at registration. A
non-Parquet file never
+ * contributes to schema inference, so it cannot seed a missing table.
+ */
+ ACCEPT
+ }
+
public abstract Set<SchemaEvolutionOption> getOptions();
/**
@@ -86,6 +108,8 @@ public abstract class SchemaEvolutionConfig implements
Serializable {
*/
public abstract @Nullable IncompatibleSchemaHandling
getIncompatibleSchemaHandling();
+ public abstract UnverifiableFileHandling getUnverifiableFileHandling();
+
public IncompatibleSchemaHandling incompatibleSchemaHandling(boolean
bounded) {
IncompatibleSchemaHandling handling = getIncompatibleSchemaHandling();
if (handling != null) {
@@ -117,7 +141,8 @@ public abstract class SchemaEvolutionConfig implements
Serializable {
public static Builder builder() {
return new AutoValue_SchemaEvolutionConfig.Builder()
.setOptions(Collections.emptySet())
- .setRequiredColumns(Collections.emptySet());
+ .setRequiredColumns(Collections.emptySet())
+ .setUnverifiableFileHandling(UnverifiableFileHandling.REJECT);
}
@AutoValue.Builder
@@ -129,9 +154,11 @@ public abstract class SchemaEvolutionConfig implements
Serializable {
public abstract Builder setIncompatibleSchemaHandling(
@Nullable IncompatibleSchemaHandling handling);
+ public abstract Builder
setUnverifiableFileHandling(UnverifiableFileHandling handling);
+
abstract SchemaEvolutionConfig autoBuild();
- /** Pins and handling without an option would silently do nothing, so they
are rejected. */
+ /** Any setting without an option would silently do nothing, so they are
rejected. */
public SchemaEvolutionConfig build() {
SchemaEvolutionConfig config = autoBuild();
for (String column : config.getRequiredColumns()) {
@@ -143,9 +170,10 @@ public abstract class SchemaEvolutionConfig implements
Serializable {
Preconditions.checkArgument(
config.isEnabled()
|| (config.getRequiredColumns().isEmpty()
- && config.getIncompatibleSchemaHandling() == null),
- "required columns and incompatible schema handling need at least one
schema evolution"
- + " option");
+ && config.getIncompatibleSchemaHandling() == null
+ && config.getUnverifiableFileHandling() ==
UnverifiableFileHandling.REJECT),
+ "required columns, incompatible schema handling and unverifiable
file handling need at"
+ + " least one schema evolution option");
return config;
}
}
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 96da59f4505..85d82b84551 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
@@ -36,18 +36,31 @@ import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Comparator;
+import java.util.EnumSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.StringUtf8Coder;
+import
org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.UnverifiableFileHandling;
+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.testing.ExpectedLogs;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.TestStream;
+import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.TupleTagList;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
@@ -116,6 +129,7 @@ public class AddFilesTest {
private IcebergCatalogConfig catalogConfig;
@ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new
TemporaryFolder();
@Rule public TestName testName = new TestName();
+ @Rule public ExpectedLogs logs = ExpectedLogs.none(AddFiles.class);
@Rule public ExpectedException thrown = ExpectedException.none();
@Rule
@@ -949,4 +963,671 @@ public class AddFilesTest {
private Record record(int id, String name, int age) {
return GenericRecord.create(icebergSchema).copy("id", id, "name", name,
"age", age);
}
+
+ // ---- ConvertToDataFile coverage check and pinned columns
+
+ private static final SchemaEvolutionConfig ADDITIONS =
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION);
+
+ private PCollectionTuple convert(SchemaEvolutionConfig config, String...
files) {
+ PCollectionTuple out =
+ pipeline
+ .apply("Create Input", Create.of(Arrays.asList(files)))
+ .apply(
+ ParDo.of(
+ new AddFiles.ConvertToDataFile(
+ catalogConfig, tableId.toString(), null, null,
null, null, config))
+ .withOutputTags(
+ AddFiles.ConvertToDataFile.DATA_FILES,
+ TupleTagList.of(AddFiles.ConvertToDataFile.ERRORS)));
+
out.get(AddFiles.ConvertToDataFile.ERRORS).setRowSchema(AddFiles.ERROR_SCHEMA);
+ return out;
+ }
+
+ private String writeWithSchema(String name, Schema schema, Record...
records) throws IOException {
+ String file = root + name;
+ DataWriter<Record> writer =
+ Parquet.writeData(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 String writeOneRecord(String name) throws IOException {
+ String file = root + name;
+ DataWriter<Record> writer = createWriter(file);
+ writer.write(record(1, "a", 1));
+ writer.close();
+ return file;
+ }
+
+ private static final Schema WIDER =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "name", Types.StringType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()),
+ Types.NestedField.optional(4, "email", Types.StringType.get()));
+
+ private String writeWider(String name) throws IOException {
+ Record record = GenericRecord.create(WIDER);
+ record.setField("id", 1);
+ record.setField("name", "a");
+ record.setField("age", 1);
+ record.setField("email", "e");
+ return writeWithSchema(name, WIDER, record);
+ }
+
+ private void assertSingleError(PCollectionTuple out, String file, String
contains) {
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.DATA_FILES)).empty();
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS))
+ .satisfies(
+ rows -> {
+ Row row = Iterables.getOnlyElement(rows);
+ assertEquals(file, row.getString("file"));
+ assertThat(row.getString("error"), containsString(contains));
+ return null;
+ });
+ }
+
+ private void assertRegisters(PCollectionTuple out, long files) {
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS)).empty();
+
PAssert.thatSingleton(out.get(AddFiles.ConvertToDataFile.DATA_FILES).apply(Count.globally()))
+ .isEqualTo(files);
+ }
+
+ @Test
+ public void testEmptySchemaTableWarnsAndStillRegisters() throws Exception {
+ catalog.createTable(tableId, new Schema());
+ String file = writeOneRecord("data.parquet");
+
+ PCollectionTuple out = convert(SchemaEvolutionConfig.disabled(), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ logs.verifyWarn("has no columns");
+ }
+
+ /** Files without embedded field ids against a zero-column table still
register. */
+ @Test
+ public void testEmptySchemaTableWithIdLessParquetStillRegisters() throws
Exception {
+ catalog.createTable(tableId, new Schema());
+ File file = new File(temp.getRoot(), "idless.parquet");
+ org.apache.avro.Schema avro =
+ org.apache.avro.SchemaBuilder.record("r")
+ .fields()
+ .requiredInt("id")
+ .optionalString("name")
+ .name("address")
+ .type()
+ .record("address")
+ .fields()
+ .optionalString("city")
+ .endRecord()
+ .noDefault()
+ .endRecord();
+ try (org.apache.parquet.hadoop.ParquetWriter<Object> writer =
+ org.apache.parquet.avro.AvroParquetWriter.builder(
+ new org.apache.hadoop.fs.Path(file.getAbsolutePath()))
+ .withSchema(avro)
+ .build()) {
+ org.apache.avro.generic.GenericData.Record record =
+ new org.apache.avro.generic.GenericData.Record(avro);
+ record.put("id", 1);
+ record.put("name", "a");
+ org.apache.avro.generic.GenericData.Record address =
+ new
org.apache.avro.generic.GenericData.Record(avro.getField("address").schema());
+ address.put("city", "c");
+ record.put("address", address);
+ writer.write(record);
+ }
+
+ PCollectionTuple out = convert(SchemaEvolutionConfig.disabled(),
file.getAbsolutePath());
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testCoveredFileRegistersWithEvolutionEnabled() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeOneRecord("data.parquet");
+
+ PCollectionTuple out = convert(ADDITIONS, file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testUncoveredFileRoutesToErrorsWhenEvolutionEnabled() throws
Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeWider("wider.parquet");
+
+ PCollectionTuple out = convert(ADDITIONS, file);
+
+ assertSingleError(out, file, "does not cover the file");
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS))
+ .satisfies(
+ rows -> {
+ assertThat(
+ Iterables.getOnlyElement(rows).getString("error"),
+ containsString("add optional email string"));
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testExtraColumnsRegisterWhenEvolutionDisabled() throws Exception
{
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeWider("wider.parquet");
+
+ PCollectionTuple out = convert(SchemaEvolutionConfig.disabled(), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testUnreadableSchemaRoutesToErrorsWithConverterMessage() throws
Exception {
+ catalog.createTable(tableId, icebergSchema);
+ // a legacy unannotated repeated field: readable Parquet, rejected by
Iceberg's converter
+ org.apache.parquet.schema.MessageType legacy =
+ org.apache.parquet.schema.Types.buildMessage()
+
.required(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32)
+ .named("id")
+
.repeated(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32)
+ .named("vals")
+ .named("root");
+ File legacyFile = new File(temp.getRoot(), "legacy.parquet");
+ try
(org.apache.parquet.hadoop.ParquetWriter<org.apache.parquet.example.data.Group>
writer =
+ org.apache.parquet.hadoop.example.ExampleParquetWriter.builder(
+ new org.apache.hadoop.fs.Path(legacyFile.getAbsolutePath()))
+ .withType(legacy)
+ .build()) {
+ org.apache.parquet.example.data.Group group =
+ new
org.apache.parquet.example.data.simple.SimpleGroupFactory(legacy).newGroup();
+ group.add("id", 1);
+ group.add("vals", 2);
+ writer.write(group);
+ }
+ String file = legacyFile.getAbsolutePath();
+
+ PCollectionTuple out = convert(ADDITIONS, file);
+
+ assertSingleError(out, file,
AddFiles.ConvertToDataFile.UNREADABLE_SCHEMA_ERROR);
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS))
+ .satisfies(
+ rows -> {
+ assertThat(
+ Iterables.getOnlyElement(rows).getString("error"),
+ containsString("repetition REPEATED"));
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ }
+
+ // ---- pinned columns
+ //
+ // A pin is enforced in two layers. A pinned column the table holds as
REQUIRED is protected by
+ // the coverage check: a file that declares it optional with nulls, or lacks
it, needs a
+ // relaxation of a pinned column, which SchemaDelta refuses, so the file is
routed there
+ // (testPinnedRequiredColumnIsProtectedByCoverage). The per-file pin walk is
reached only for
+ // pinned columns the table holds as OPTIONAL: columns the pre-pass added
(it never creates them
+ // required) or pre-existing optional ones. The tests below therefore pin
optional columns.
+
+ private static SchemaEvolutionConfig pinned(String column) {
+ return SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(Collections.singleton(column))
+ .build();
+ }
+
+ private static final Schema OPTIONAL_NAME =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()));
+
+ private static final Schema WITHOUT_NAME =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()));
+
+ private String writeCleanName(String name) throws IOException {
+ return writeWithSchema(
+ name,
+ OPTIONAL_NAME,
+ GenericRecord.create(OPTIONAL_NAME).copy("id", 1, "name", "a", "age",
1));
+ }
+
+ private String writeOneNullName(String name) throws IOException {
+ return writeWithSchema(
+ name,
+ OPTIONAL_NAME,
+ GenericRecord.create(OPTIONAL_NAME).copy("id", 1, "name", "a", "age",
1),
+ GenericRecord.create(OPTIONAL_NAME).copy("id", 2, "age", 2));
+ }
+
+ private String writeWithoutName(String name) throws IOException {
+ return writeWithSchema(
+ name, WITHOUT_NAME, GenericRecord.create(WITHOUT_NAME).copy("id", 1,
"age", 1));
+ }
+
+ @Test
+ public void testPinnedRequiredColumnIsProtectedByCoverage() throws Exception
{
+ catalog.createTable(tableId, icebergSchema);
+ String withNull = writeOneNullName("nulls.parquet");
+ String absent = writeWithoutName("noname.parquet");
+
+ PCollectionTuple out = convert(pinned("name"), withNull, absent);
+
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.DATA_FILES)).empty();
+ PAssert.that(out.get(AddFiles.ConvertToDataFile.ERRORS))
+ .satisfies(
+ rows -> {
+ assertEquals(2, Iterables.size(rows));
+ for (Row row : rows) {
+ assertThat(row.getString("error"), containsString("does not
cover the file"));
+ assertThat(row.getString("error"), containsString("pinned as
required"));
+ }
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedColumnWithNullsRoutesToErrors() throws Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ String file = writeOneNullName("nulls.parquet");
+
+ PCollectionTuple out = convert(pinned("name"), file);
+
+ assertSingleError(out, file, "Pinned required column name has 1 null(s)");
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedColumnAbsentRoutesToErrors() throws Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ String file = writeWithoutName("noname.parquet");
+
+ PCollectionTuple out = convert(pinned("name"), file);
+
+ assertSingleError(out, file, "Pinned required column name is absent from
the file");
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedColumnProvenNullFreeRegisters() throws Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ String file = writeCleanName("clean.parquet");
+
+ PCollectionTuple out = convert(pinned("name"), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedColumnZeroRowFileRegisters() throws Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ // Iceberg's writer creates no file for zero rows; parquet-avro does.
+ File empty = new File(temp.getRoot(), "empty.parquet");
+ org.apache.parquet.avro.AvroParquetWriter.builder(
+ new org.apache.hadoop.fs.Path(empty.getAbsolutePath()))
+ .withSchema(AVRO_OPTIONAL_NAME)
+ .build()
+ .close();
+ String file = empty.getAbsolutePath();
+
+ PCollectionTuple out = convert(pinned("name"), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ /** Pin evidence comes from the footer: the table's metrics configuration
cannot disable it. */
+ @Test
+ public void testPinnedColumnProvenNullFreeRegistersUnderMetricsModeNone()
throws Exception {
+ catalog.createTable(
+ tableId,
+ OPTIONAL_NAME,
+ PartitionSpec.unpartitioned(),
+ ImmutableMap.of("write.metadata.metrics.default", "none"));
+ String file = writeCleanName("clean.parquet");
+
+ PCollectionTuple out = convert(pinned("name"), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedColumnNullsDetectedUnderMetricsModeNone() throws
Exception {
+ catalog.createTable(
+ tableId,
+ OPTIONAL_NAME,
+ PartitionSpec.unpartitioned(),
+ ImmutableMap.of("write.metadata.metrics.default", "none"));
+ String file = writeOneNullName("nulls.parquet");
+
+ PCollectionTuple out = convert(pinned("name"), file);
+
+ assertSingleError(out, file, "Pinned required column name has 1 null(s)");
+ pipeline.run().waitUntilFinish();
+ }
+
+ /** With evolution on, a format the checks cannot read must not register
unchecked. */
+ @Test
+ public void testNonParquetFileRoutesToErrorsWhenEvolutionEnabled() throws
Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeAvroFile("data.avro");
+
+ PCollectionTuple out = convert(ADDITIONS, file);
+
+ assertSingleError(out, file,
AddFiles.ConvertToDataFile.UNCHECKED_FORMAT_ERROR + "AVRO");
+ pipeline.run().waitUntilFinish();
+ }
+
+ // ---- UnverifiableFileHandling.ACCEPT
+
+ private static SchemaEvolutionConfig accepting(SchemaEvolutionConfig base) {
+ return SchemaEvolutionConfig.builder()
+ .setOptions(base.getOptions())
+ .setRequiredColumns(base.getRequiredColumns())
+ .setUnverifiableFileHandling(UnverifiableFileHandling.ACCEPT)
+ .build();
+ }
+
+ private static long counted(PipelineResult result, String counter) {
+ long total = 0;
+ for (MetricResult<Long> metric :
+ result
+ .metrics()
+ .queryMetrics(
+ MetricsFilter.builder()
+ .addNameFilter(MetricNameFilter.named(AddFiles.class,
counter))
+ .build())
+ .getCounters()) {
+ total += metric.getAttempted();
+ }
+ return total;
+ }
+
+ private String writeAvroFile(String name) throws IOException {
+ File avroFile = new File(temp.getRoot(), name);
+ org.apache.avro.Schema avro =
+
org.apache.avro.SchemaBuilder.record("r").fields().requiredInt("id").endRecord();
+ try
(org.apache.avro.file.DataFileWriter<org.apache.avro.generic.GenericRecord>
writer =
+ new org.apache.avro.file.DataFileWriter<>(
+ new org.apache.avro.generic.GenericDatumWriter<>(avro))) {
+ writer.create(avro, avroFile);
+ org.apache.avro.generic.GenericData.Record avroRecord =
+ new org.apache.avro.generic.GenericData.Record(avro);
+ avroRecord.put("id", 1);
+ writer.append(avroRecord);
+ }
+ return avroFile.getAbsolutePath();
+ }
+
+ private static final org.apache.avro.Schema AVRO_OPTIONAL_NAME =
+ org.apache.avro.SchemaBuilder.record("r")
+ .fields()
+ .requiredInt("id")
+ .optionalString("name")
+ .requiredInt("age")
+ .endRecord();
+
+ /**
+ * One OPTIONAL_NAME-shaped row written by parquet-avro with column
statistics switched off, for
+ * the named columns or for every column when none is named.
+ */
+ private String writeWithoutStatistics(String name, List<String>
statlessColumns, boolean nullName)
+ throws IOException {
+ File file = new File(temp.getRoot(), name);
+ org.apache.parquet.avro.AvroParquetWriter.Builder<Object> builder =
+ org.apache.parquet.avro.AvroParquetWriter.builder(
+ new org.apache.hadoop.fs.Path(file.getAbsolutePath()))
+ .withSchema(AVRO_OPTIONAL_NAME);
+ if (statlessColumns.isEmpty()) {
+ builder = builder.withStatisticsEnabled(false);
+ }
+ for (String column : statlessColumns) {
+ builder = builder.withStatisticsEnabled(column, false);
+ }
+ try (org.apache.parquet.hadoop.ParquetWriter<Object> writer =
builder.build()) {
+ org.apache.avro.generic.GenericData.Record record =
+ new org.apache.avro.generic.GenericData.Record(AVRO_OPTIONAL_NAME);
+ record.put("id", 1);
+ record.put("name", nullName ? null : "a");
+ record.put("age", 1);
+ writer.write(record);
+ }
+ return file.getAbsolutePath();
+ }
+
+ private static final org.apache.avro.Schema AVRO_REQUIRED_NAME =
+ org.apache.avro.SchemaBuilder.record("r")
+ .fields()
+ .requiredInt("id")
+ .requiredString("name")
+ .requiredInt("age")
+ .endRecord();
+
+ /** Parquet cannot encode a null in a required column, so no statistics are
needed to prove it. */
+ @Test
+ public void testPinnedColumnDeclaredRequiredRegistersWithoutStatistics()
throws Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ File file = new File(temp.getRoot(), "required_nostats.parquet");
+ try (org.apache.parquet.hadoop.ParquetWriter<Object> writer =
+ org.apache.parquet.avro.AvroParquetWriter.builder(
+ new org.apache.hadoop.fs.Path(file.getAbsolutePath()))
+ .withSchema(AVRO_REQUIRED_NAME)
+ .withStatisticsEnabled(false)
+ .build()) {
+ org.apache.avro.generic.GenericData.Record record =
+ new org.apache.avro.generic.GenericData.Record(AVRO_REQUIRED_NAME);
+ record.put("id", 1);
+ record.put("name", "a");
+ record.put("age", 1);
+ writer.write(record);
+ }
+
+ PCollectionTuple out = convert(pinned("name"), file.getAbsolutePath());
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedColumnWithoutStatisticsRoutesToErrors() throws
Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ String file = writeWithoutStatistics("nostats.parquet",
Collections.emptyList(), false);
+
+ PCollectionTuple out = convert(pinned("name"), file);
+
+ assertSingleError(
+ out, file, "Pinned required column name has no null count statistics
in the file");
+ pipeline.run().waitUntilFinish();
+ }
+
+ /** The trusted file does hold a null the footer cannot report; ACCEPT
registers it anyway. */
+ @Test
+ public void testPinnedColumnWithoutStatisticsRegistersWhenAccepted() throws
Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ String file = writeWithoutStatistics("nostats.parquet",
Collections.emptyList(), true);
+
+ PCollectionTuple out = convert(accepting(pinned("name")), file);
+
+ assertRegisters(out, 1);
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
+ assertEquals(1, counted(result, AddFiles.UNPROVEN_PINS_COUNTER));
+ assertEquals(0, counted(result, AddFiles.UNCHECKED_FORMAT_COUNTER));
+ logs.verifyWarn("no null count statistics for pinned column(s) [name]");
+ }
+
+ /**
+ * ACCEPT trusts only what the footer cannot say; a pin the footer does
count is still enforced.
+ */
+ @Test
+ public void testPinWithNullsStillCaughtWhenAnotherPinIsUnproven() throws
Exception {
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ String file = writeWithoutStatistics("mixed.parquet",
Collections.singletonList("age"), true);
+ SchemaEvolutionConfig config =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(new LinkedHashSet<>(Arrays.asList("age",
"name")))
+ .setUnverifiableFileHandling(UnverifiableFileHandling.ACCEPT)
+ .build();
+
+ PCollectionTuple out = convert(config, file);
+
+ assertSingleError(out, file, "Pinned required column name has 1 null(s)");
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedColumnAbsentStillCaughtWhenAccepted() throws Exception
{
+ catalog.createTable(tableId, OPTIONAL_NAME);
+ String file = writeWithoutName("noname.parquet");
+
+ PCollectionTuple out = convert(accepting(pinned("name")), file);
+
+ assertSingleError(out, file, "Pinned required column name is absent from
the file");
+ pipeline.run().waitUntilFinish();
+ }
+
+ private static final Schema WITH_ITEMS =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(
+ 2,
+ "items",
+ Types.ListType.ofOptional(
+ 3,
+ Types.StructType.of(
+ Types.NestedField.optional(4, "sku",
Types.StringType.get())))));
+
+ private String writeWithItems(String name) throws IOException {
+ Types.StructType item =
+
WITH_ITEMS.findField("items").type().asListType().elementType().asStructType();
+ Record row = GenericRecord.create(WITH_ITEMS);
+ row.setField("id", 1);
+ row.setField("items",
Collections.singletonList(GenericRecord.create(item).copy("sku", "x")));
+ return writeWithSchema(name, WITH_ITEMS, row);
+ }
+
+ /**
+ * The footer's chunk paths under a list are never mapped onto the pin, so
it cannot be proven.
+ */
+ @Test
+ public void testPinUnderListCannotBeProvenAndIsRejected() throws Exception {
+ catalog.createTable(tableId, WITH_ITEMS);
+ String file = writeWithItems("items.parquet");
+
+ PCollectionTuple out = convert(pinned("items.element.sku"), file);
+
+ assertSingleError(
+ out,
+ file,
+ "Pinned required column items.element.sku has no null count statistics
in the file");
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinUnderListRegistersWhenAccepted() throws Exception {
+ catalog.createTable(tableId, WITH_ITEMS);
+ String file = writeWithItems("items.parquet");
+
+ PCollectionTuple out = convert(accepting(pinned("items.element.sku")),
file);
+
+ assertRegisters(out, 1);
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
+ assertEquals(1, counted(result, AddFiles.UNPROVEN_PINS_COUNTER));
+ }
+
+ private static final Schema WITH_ADDRESS =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(
+ 2,
+ "address",
+ Types.StructType.of(
+ Types.NestedField.optional(3, "city",
Types.StringType.get()),
+ Types.NestedField.optional(4, "zip",
Types.IntegerType.get()))));
+
+ private String writeWithAddress(String name, @Nullable String city, boolean
nullAddress)
+ throws IOException {
+ Types.StructType addressType =
WITH_ADDRESS.findField("address").type().asStructType();
+ Record row = GenericRecord.create(WITH_ADDRESS);
+ row.setField("id", 1);
+ if (!nullAddress) {
+ row.setField("address", GenericRecord.create(addressType).copy("city",
city, "zip", 1));
+ }
+ return writeWithSchema(name, WITH_ADDRESS, row);
+ }
+
+ /** A struct pin is proven by any null-free leaf beneath it, exactly as
tighten proves it. */
+ @Test
+ public void testPinnedStructProvenByOneLeafRegisters() throws Exception {
+ catalog.createTable(tableId, WITH_ADDRESS);
+ String file = writeWithAddress("address.parquet", null, false);
+
+ PCollectionTuple out = convert(pinned("address"), file);
+
+ assertRegisters(out, 1);
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testPinnedStructNullInARowRoutesToErrors() throws Exception {
+ catalog.createTable(tableId, WITH_ADDRESS);
+ String file = writeWithAddress("noaddress.parquet", "c", true);
+
+ PCollectionTuple out = convert(pinned("address"), file);
+
+ assertSingleError(
+ out, file, "Pinned required column address has no null count
statistics in the file");
+ pipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testNonParquetFileRegistersUncheckedWhenAccepted() throws
Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeAvroFile("data.avro");
+
+ PCollectionTuple out = convert(accepting(ADDITIONS), file);
+
+ assertRegisters(out, 1);
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
+ assertEquals(1, counted(result, AddFiles.UNCHECKED_FORMAT_COUNTER));
+ assertEquals(0, counted(result, AddFiles.UNPROVEN_PINS_COUNTER));
+ logs.verifyWarn("unchecked (UnverifiableFileHandling.ACCEPT)");
+ }
+
+ @Test
+ public void testNonParquetFileRegistersWhenEvolutionDisabled() throws
Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeAvroFile("data.avro");
+
+ PCollectionTuple out = convert(SchemaEvolutionConfig.disabled(), file);
+
+ assertRegisters(out, 1);
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
+ assertEquals(0, counted(result, AddFiles.UNCHECKED_FORMAT_COUNTER));
+ }
}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java
index 6a6bef483b7..d72ccbf5974 100644
---
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java
@@ -21,6 +21,7 @@ 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.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
@@ -50,6 +51,7 @@ import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
import org.apache.parquet.schema.Type.Repetition;
+import org.checkerframework.checker.nullness.qual.Nullable;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -346,6 +348,99 @@ public class FileSchemasTest {
assertTrue(isRequired(schema, "address.zip"));
}
+ // ---- null counts (pin evidence), by the tighten rules
+
+ private static @Nullable Long nullCount(ParquetMetadata footer, String
column) {
+ return FileSchemas.nullCount(
+ footer,
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema()), column);
+ }
+
+ @Test
+ public void testNullCountOfLeafSumsRowGroups() throws IOException {
+ ParquetMetadata footer =
+ write(10, 2, true, new Nulls(r -> r == 1 || r == 7, r -> false, r ->
false, r -> false));
+ assertEquals(Long.valueOf(2), nullCount(footer, "name"));
+ assertEquals(Long.valueOf(0), nullCount(footer, "id"));
+ assertEquals(Long.valueOf(0), nullCount(footer, "address.city"));
+ }
+
+ @Test
+ public void testNullCountUnknownWithoutStatisticsOrUnderAList() throws
IOException {
+ assertNull(nullCount(write(10, 1, false, Nulls.NONE), "name"));
+ assertNull(nullCount(write(10, 1, true, Nulls.NONE), "tags.element"));
+ assertNull(nullCount(write(10, 1, true, Nulls.NONE), "missing"));
+ }
+
+ // root: required id, optional group home {required city}, required group
office {required city}
+ private static final MessageType REQUIRED_LEAVES =
+ org.apache.parquet.schema.Types.buildMessage()
+ .required(PrimitiveTypeName.INT64)
+ .named("id")
+ .addField(
+ org.apache.parquet.schema.Types.buildGroup(Repetition.OPTIONAL)
+ .required(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("city")
+ .named("home"))
+ .addField(
+ org.apache.parquet.schema.Types.buildGroup(Repetition.REQUIRED)
+ .required(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("city")
+ .named("office"))
+ .named("root");
+
+ /** A declared-required path cannot encode a null, so it needs no statistics
to be proven. */
+ @Test
+ public void testNullCountTrustsDeclaredRequiredPathsWithoutStatistics()
throws IOException {
+ File file = new File(tmp.getRoot(), "required.parquet");
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(new Path(file.getAbsolutePath()))
+ .withType(REQUIRED_LEAVES)
+ .withStatisticsEnabled(false)
+ .build()) {
+ Group group = new SimpleGroupFactory(REQUIRED_LEAVES).newGroup();
+ group.add("id", 1L);
+ group.addGroup("office").add("city", "c");
+ writer.write(group);
+ }
+ ParquetMetadata footer = ParquetFooters.read(file.getAbsolutePath());
+
+ assertEquals(Long.valueOf(0), nullCount(footer, "id"));
+ assertEquals(Long.valueOf(0), nullCount(footer, "office"));
+ assertEquals(Long.valueOf(0), nullCount(footer, "office.city"));
+ // required only relative to an optional parent: a null home nulls city,
and no count says
+ assertNull(nullCount(footer, "home"));
+ assertNull(nullCount(footer, "home.city"));
+ }
+
+ /** A struct is proven exactly when tighten would mark it required. */
+ @Test
+ public void testNullCountOfStructFollowsTighten() throws IOException {
+ ParquetMetadata oneLeafProven =
+ write(10, 1, true, new Nulls(r -> false, r -> false, r -> r == 2, r ->
false));
+ assertEquals(Long.valueOf(0), nullCount(oneLeafProven, "address"));
+ assertEquals(Long.valueOf(1), nullCount(oneLeafProven, "address.city"));
+
+ ParquetMetadata nullStruct =
+ write(10, 1, true, new Nulls(r -> false, r -> r == 5, r -> false, r ->
false));
+ assertNull(nullCount(nullStruct, "address"));
+ assertEquals(Long.valueOf(1), nullCount(nullStruct, "address.city"));
+
+ // every leaf has a null of its own, in different rows: the struct itself
was never null,
+ // but leaf counts cannot say so
+ ParquetMetadata leavesNulled =
+ write(10, 1, true, new Nulls(r -> false, r -> false, r -> r == 2, r ->
r == 4));
+ assertNull(nullCount(leavesNulled, "address"));
+ }
+
+ @Test
+ public void testNullCountZeroRowsProvesEverything() throws IOException {
+ ParquetMetadata footer = write(0, 1, false, Nulls.NONE);
+ assertEquals(Long.valueOf(0), nullCount(footer, "name"));
+ assertEquals(Long.valueOf(0), nullCount(footer, "address"));
+ }
+
@Test
public void testTightenPreservesIdsNamesAndTypes() throws IOException {
ParquetMetadata footer = write(10, 1, true, Nulls.NONE);
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java
index edc8b338bfc..4ccf844538f 100644
---
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java
@@ -26,6 +26,7 @@ import java.util.Collections;
import java.util.EnumSet;
import java.util.HashSet;
import
org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import
org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.UnverifiableFileHandling;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -66,13 +67,34 @@ public class SchemaEvolutionConfigTest {
SchemaEvolutionConfig.Builder handlingOnly =
SchemaEvolutionConfig.builder()
.setIncompatibleSchemaHandling(IncompatibleSchemaHandling.ROUTE_TO_ERRORS);
+ SchemaEvolutionConfig.Builder acceptOnly =
+ SchemaEvolutionConfig.builder()
+ .setUnverifiableFileHandling(UnverifiableFileHandling.ACCEPT);
IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
pinsOnly::build);
assertThrows(IllegalArgumentException.class, handlingOnly::build);
+ assertThrows(IllegalArgumentException.class, acceptOnly::build);
assertTrue(e.getMessage(), e.getMessage().contains("at least one schema
evolution option"));
}
+ @Test
+ public void testUnverifiableFileHandlingDefaultsToReject() {
+ SchemaEvolutionConfig unset =
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION);
+ SchemaEvolutionConfig accepting =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))
+ .setUnverifiableFileHandling(UnverifiableFileHandling.ACCEPT)
+ .build();
+
+ assertEquals(UnverifiableFileHandling.REJECT,
unset.getUnverifiableFileHandling());
+ assertEquals(
+ UnverifiableFileHandling.REJECT,
+ SchemaEvolutionConfig.disabled().getUnverifiableFileHandling());
+ assertEquals(UnverifiableFileHandling.ACCEPT,
accepting.getUnverifiableFileHandling());
+ }
+
@Test
public void testIncompatibleSchemaHandlingDefaultsByMode() {
SchemaEvolutionConfig unset =
