ahmedabu98 commented on code in PR #39975:
URL: https://github.com/apache/beam/pull/39975#discussion_r3926025037
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java:
##########
@@ -18,29 +18,208 @@
package org.apache.beam.sdk.io.iceberg;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.parquet.ParquetSchemaUtil;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+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;
/**
- * Derives the schema a file contributes to schema inference. The canonical
form sorts struct fields
- * by name at every level and renumbers ids in deterministic order, so files
that differ only in
- * column order produce identical JSON. Ids are positional and meaningless:
the commit side
- * reconciles columns by name.
+ * What a file contributes to schema inference: the canonical form of the
schema it declares, and
+ * the columns its footer proves free of nulls.
+ *
+ * <p>The schema half depends only on the declared schema, never on the data,
so files written by
+ * the same job dedup to one entry no matter where their nulls fall. The null
evidence is combined
+ * per schema by {@link CollectDistinctSchemas} and reapplied by the commit
side via {@link
+ * #markRequired}.
+ *
+ * <p>The canonical form sorts struct fields by name at every level and
renumbers ids. Ids are
+ * positional and meaningless (the commit side reconciles columns by name), so
never diff two file
+ * schemas by id. Other field attributes (doc, defaults) are preserved,
matching what SchemaDelta
+ * compares. Column paths are dotted, like pins.
*/
final class FileSchemas {
private FileSchemas() {}
+ /** Canonical JSON of the schema the file declares. */
static String canonicalJson(ParquetMetadata footer) {
Schema converted =
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
return SchemaParser.toJson(canonical(converted));
}
+ /** This file as a schema group of one: its declared schema and its
null-free columns. */
+ static CollectDistinctSchemas.SchemaGroup schemaGroup(ParquetMetadata
footer) {
+ Schema converted =
ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
+ Schema tightened = tighten(converted, footer);
+ return new CollectDistinctSchemas.SchemaGroup(
+ SchemaParser.toJson(canonical(converted)), 1,
changedToRequired(converted, tightened));
+ }
+
+ /**
+ * 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
+ * struct is null-free when any leaf under it is (a null struct nulls all
its leaves). Nothing
+ * under lists or maps is tightened: a zero count there would be valid
evidence too, but mapping
+ * physical chunk paths (writer-dependent names like {@code list.element},
{@code array}) onto the
+ * converted schema is not worth it. A file with no row groups has no rows
and proves every
+ * column, matching how the pin check treats empty files.
+ */
+ static Schema tighten(Schema schema, ParquetMetadata footer) {
+ if (footer.getBlocks().isEmpty()) {
+ return new Schema(tightenAll(schema.asStruct()).fields());
+ }
+ Set<List<String>> zeroNullLeaves = leafPathsWithZeroNullCounts(footer);
+ if (zeroNullLeaves.isEmpty()) {
+ return schema;
+ }
+ return new Schema(
+ tightenStruct(schema.asStruct(), new ArrayList<>(),
zeroNullLeaves).struct.fields());
+ }
+
+ /** With no rows, nothing can hold a null: every leaf and struct outside
lists and maps. */
+ private static Types.StructType tightenAll(Types.StructType struct) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : struct.fields()) {
+ Type type = field.type();
+ if (type.isStructType()) {
+ fields.add(withOptionality(field, tightenAll(type.asStructType()),
false));
+ } else if (type.isPrimitiveType()) {
+ fields.add(withOptionality(field, type, false));
+ } else {
+ fields.add(field);
+ }
+ }
+ return Types.StructType.of(fields);
+ }
+
+ /**
+ * Returns the schema with the given dotted column paths made required. The
commit side parses a
+ * group's schema JSON (optionality as the writer declared it) and applies
the group's null-free
+ * columns with this before classifying, so only relaxations some file
actually needs remain.
+ */
+ static Schema markRequired(Schema declared, Collection<String> columns) {
+ if (columns.isEmpty()) {
+ return declared;
+ }
+ Types.StructType required = markRequiredStruct(declared.asStruct(), "",
new HashSet<>(columns));
+ return new Schema(required.fields());
+ }
+
+ private static Types.StructType markRequiredStruct(
+ Types.StructType struct, String prefix, Set<String> columns) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : struct.fields()) {
+ String path = prefix + field.name();
+ Type type = field.type();
+ if (type.isStructType()) {
+ type = markRequiredStruct(type.asStructType(), path + ".", columns);
+ }
+ boolean required = !field.isRequired() && columns.contains(path);
+ fields.add(withOptionality(field, type, field.isOptional() &&
!required));
+ }
+ return Types.StructType.of(fields);
+ }
+
+ /** Dotted paths of fields the tightened schema made required, sorted. */
+ private static List<String> changedToRequired(Schema declared, Schema
tightened) {
+ List<String> paths = new ArrayList<>();
+ collectChangedToRequired(declared.asStruct(), tightened.asStruct(), "",
paths);
+ Collections.sort(paths);
+ return paths;
+ }
+
+ private static void collectChangedToRequired(
+ Types.StructType declared, Types.StructType tightened, String prefix,
List<String> out) {
+ for (int i = 0; i < declared.fields().size(); i++) {
+ Types.NestedField before = declared.fields().get(i);
+ Types.NestedField after = tightened.fields().get(i);
+ String path = prefix + before.name();
+ if (before.isOptional() && after.isRequired()) {
+ out.add(path);
+ }
Review Comment:
ignore this: I guess it's not useful information if we know the column is
already required
--
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]
