amogh-jahagirdar commented on code in PR #17320:
URL: https://github.com/apache/iceberg/pull/17320#discussion_r3979766385
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java:
##########
@@ -231,6 +235,90 @@ public static ParquetValueReader<?>
replaceWithMetadataReader(
return reader;
}
+ /**
+ * Builds a struct reader from the expected fields, in field order. A field
present in the file
+ * uses its column reader; a field missing from the file uses a metadata or
partition constant, or
+ * its initial default. When no expected field reads a file column, the
struct reader is wrapped
+ * so a presence column supplies the definition level.
+ */
+ public static <T> ParquetValueReader<T> structReader(
+ MessageType fileSchema,
+ String[] structPath,
+ List<Types.NestedField> expectedFields,
+ Map<Integer, ParquetValueReader<?>> readersById,
+ Map<Integer, ?> idToConstant,
+ BiFunction<org.apache.iceberg.types.Type, Object, Object>
convertConstant,
+ Function<List<ParquetValueReader<?>>, ParquetValueReader<T>>
newStructReader) {
+ int constantDefinitionLevel = fileSchema.getMaxDefinitionLevel(structPath);
+
+ List<ParquetValueReader<?>> readers =
Lists.newArrayListWithExpectedSize(expectedFields.size());
+ for (Types.NestedField field : expectedFields) {
+ int id = field.fieldId();
+ ParquetValueReader<?> reader =
+ replaceWithMetadataReader(id, readersById.get(id), idToConstant,
constantDefinitionLevel);
+ readers.add(defaultReader(field, reader, constantDefinitionLevel,
convertConstant));
+ }
+
+ ParquetValueReader<T> reader = newStructReader.apply(readers);
+ ColumnDescriptor presence =
+ presenceColumn(
+ fileSchema, structPath, constantDefinitionLevel, expectedFields,
readersById);
+
+ return presence == null
+ ? reader
+ : withPresence(reader, presence,
fileSchema.getMaxRepetitionLevel(structPath));
Review Comment:
Done, inlined.
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java:
##########
@@ -129,12 +132,103 @@ public static Type fieldType(GroupType group, String
name) {
public static MessageType pruneColumns(MessageType fileSchema, Schema
expectedSchema) {
// column order must match the incoming type, so it doesn't matter that
the ids are unordered
- Set<Integer> selectedIds = TypeUtil.getProjectedIds(expectedSchema);
+ Set<Integer> selectedIds =
Sets.newHashSet(TypeUtil.getProjectedIds(expectedSchema));
+ // retain one real leaf under each struct that projects only constants
like default values,
+ // so its definition level still shows whether the struct is present
+ TypeWithSchemaVisitor.visit(
+ expectedSchema.asStruct(), fileSchema, new
PresenceColumnSelector(fileSchema, selectedIds));
return (MessageType)
TypeWithSchemaVisitor.visit(
expectedSchema.asStruct(), fileSchema, new
PruneColumns(selectedIds));
}
+ /**
+ * Adds one leaf id under each projected struct whose fields are all
constants and would otherwise
+ * retain no file leaf. That leaf's definition level is what still shows
whether the struct is
+ * present.
+ */
+ private static class PresenceColumnSelector extends
TypeWithSchemaVisitor<Void> {
+ private final MessageType fileSchema;
+ private final Set<Integer> selectedIds;
+
+ private PresenceColumnSelector(MessageType fileSchema, Set<Integer>
selectedIds) {
+ this.fileSchema = fileSchema;
+ this.selectedIds = selectedIds;
+ }
+
+ @Override
+ public Void struct(Types.StructType expected, GroupType struct, List<Void>
fields) {
+ // nothing projected under this struct, so there is nothing to track
+ if (expected == null || expected.fields().isEmpty()) {
+ return null;
+ }
+
+ String[] path = currentPath();
+ // add a presence column only if no real leaf under the struct is
already read
+ boolean readsRealLeaf =
+ leafColumns(fileSchema, path).stream()
+ .anyMatch(leaf -> selectedIds.contains(leafId(leaf)));
+ if (!readsRealLeaf) {
+ ColumnDescriptor presence = selectPresenceColumn(fileSchema, path);
+ if (presence != null) {
+ selectedIds.add(leafId(presence));
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Void variant(Types.VariantType expected, GroupType variantGroup,
Void result) {
+ return null;
+ }
+ }
+
+ /** Shallowest leaf under path; its definition level shows whether the
struct is present. */
+ static ColumnDescriptor selectPresenceColumn(MessageType fileSchema,
String[] path) {
+ if (fileSchema.getMaxDefinitionLevel(path) <= 0) {
+ return null;
+ }
+
+ return leafColumns(fileSchema, path).stream()
+ .min(Comparator.comparingInt(ColumnDescriptor::getMaxRepetitionLevel))
+ .orElse(null);
+ }
+
+ /** Returns the leaf columns with ids under the given path. */
+ static List<ColumnDescriptor> leafColumns(MessageType fileSchema, String[]
path) {
Review Comment:
Done, made `leafColumns` private.
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java:
##########
@@ -129,12 +132,103 @@ public static Type fieldType(GroupType group, String
name) {
public static MessageType pruneColumns(MessageType fileSchema, Schema
expectedSchema) {
// column order must match the incoming type, so it doesn't matter that
the ids are unordered
- Set<Integer> selectedIds = TypeUtil.getProjectedIds(expectedSchema);
+ Set<Integer> selectedIds =
Sets.newHashSet(TypeUtil.getProjectedIds(expectedSchema));
+ // retain one real leaf under each struct that projects only constants
like default values,
+ // so its definition level still shows whether the struct is present
+ TypeWithSchemaVisitor.visit(
+ expectedSchema.asStruct(), fileSchema, new
PresenceColumnSelector(fileSchema, selectedIds));
return (MessageType)
TypeWithSchemaVisitor.visit(
expectedSchema.asStruct(), fileSchema, new
PruneColumns(selectedIds));
}
+ /**
+ * Adds one leaf id under each projected struct whose fields are all
constants and would otherwise
+ * retain no file leaf. That leaf's definition level is what still shows
whether the struct is
+ * present.
+ */
+ private static class PresenceColumnSelector extends
TypeWithSchemaVisitor<Void> {
+ private final MessageType fileSchema;
+ private final Set<Integer> selectedIds;
+
+ private PresenceColumnSelector(MessageType fileSchema, Set<Integer>
selectedIds) {
+ this.fileSchema = fileSchema;
+ this.selectedIds = selectedIds;
+ }
+
+ @Override
+ public Void struct(Types.StructType expected, GroupType struct, List<Void>
fields) {
+ // nothing projected under this struct, so there is nothing to track
+ if (expected == null || expected.fields().isEmpty()) {
+ return null;
+ }
+
+ String[] path = currentPath();
+ // add a presence column only if no real leaf under the struct is
already read
+ boolean readsRealLeaf =
+ leafColumns(fileSchema, path).stream()
+ .anyMatch(leaf -> selectedIds.contains(leafId(leaf)));
+ if (!readsRealLeaf) {
+ ColumnDescriptor presence = selectPresenceColumn(fileSchema, path);
+ if (presence != null) {
+ selectedIds.add(leafId(presence));
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Void variant(Types.VariantType expected, GroupType variantGroup,
Void result) {
+ return null;
+ }
+ }
+
+ /** Shallowest leaf under path; its definition level shows whether the
struct is present. */
+ static ColumnDescriptor selectPresenceColumn(MessageType fileSchema,
String[] path) {
+ if (fileSchema.getMaxDefinitionLevel(path) <= 0) {
+ return null;
+ }
+
+ return leafColumns(fileSchema, path).stream()
+ .min(Comparator.comparingInt(ColumnDescriptor::getMaxRepetitionLevel))
+ .orElse(null);
+ }
+
+ /** Returns the leaf columns with ids under the given path. */
+ static List<ColumnDescriptor> leafColumns(MessageType fileSchema, String[]
path) {
+ List<ColumnDescriptor> columns = Lists.newArrayList();
+ for (ColumnDescriptor column : fileSchema.getColumns()) {
+ // a presence column is kept in the read set by id, so a leaf without
one cannot be used
Review Comment:
Agreed it's worth doing, leaving it as a follow-up as you suggested.
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java:
##########
@@ -129,12 +132,103 @@ public static Type fieldType(GroupType group, String
name) {
public static MessageType pruneColumns(MessageType fileSchema, Schema
expectedSchema) {
// column order must match the incoming type, so it doesn't matter that
the ids are unordered
- Set<Integer> selectedIds = TypeUtil.getProjectedIds(expectedSchema);
+ Set<Integer> selectedIds =
Sets.newHashSet(TypeUtil.getProjectedIds(expectedSchema));
+ // retain one real leaf under each struct that projects only constants
like default values,
+ // so its definition level still shows whether the struct is present
+ TypeWithSchemaVisitor.visit(
+ expectedSchema.asStruct(), fileSchema, new
PresenceColumnSelector(fileSchema, selectedIds));
return (MessageType)
TypeWithSchemaVisitor.visit(
expectedSchema.asStruct(), fileSchema, new
PruneColumns(selectedIds));
}
+ /**
+ * Adds one leaf id under each projected struct whose fields are all
constants and would otherwise
+ * retain no file leaf. That leaf's definition level is what still shows
whether the struct is
+ * present.
+ */
+ private static class PresenceColumnSelector extends
TypeWithSchemaVisitor<Void> {
+ private final MessageType fileSchema;
+ private final Set<Integer> selectedIds;
+
+ private PresenceColumnSelector(MessageType fileSchema, Set<Integer>
selectedIds) {
+ this.fileSchema = fileSchema;
+ this.selectedIds = selectedIds;
+ }
+
+ @Override
+ public Void struct(Types.StructType expected, GroupType struct, List<Void>
fields) {
+ // nothing projected under this struct, so there is nothing to track
+ if (expected == null || expected.fields().isEmpty()) {
+ return null;
+ }
+
+ String[] path = currentPath();
+ // add a presence column only if no real leaf under the struct is
already read
+ boolean readsRealLeaf =
+ leafColumns(fileSchema, path).stream()
+ .anyMatch(leaf -> selectedIds.contains(leafId(leaf)));
+ if (!readsRealLeaf) {
+ ColumnDescriptor presence = selectPresenceColumn(fileSchema, path);
+ if (presence != null) {
+ selectedIds.add(leafId(presence));
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Void variant(Types.VariantType expected, GroupType variantGroup,
Void result) {
+ return null;
+ }
+ }
+
+ /** Shallowest leaf under path; its definition level shows whether the
struct is present. */
+ static ColumnDescriptor selectPresenceColumn(MessageType fileSchema,
String[] path) {
+ if (fileSchema.getMaxDefinitionLevel(path) <= 0) {
+ return null;
+ }
+
+ return leafColumns(fileSchema, path).stream()
Review Comment:
Yeah, this is the same seam as the thread below (unify the presence decision
so both sides use the same leaf set) — replying there with the plan.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
