You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "rdblue (via GitHub)" <gi...@apache.org> on 2025/01/31 00:19:12 UTC
[PR] Parquet: Implement Variant readers [iceberg]
rdblue opened a new pull request, #12139:
URL: https://github.com/apache/iceberg/pull/12139
This adds Parquet readers for Variant data, including for shredded fields.
The readers are built using a new Variant visitor that is plugged into the existing Parquet type visitor by returning a Variant visitor class. When a Variant visitor is present, it is called to produce a result to pass into the Parquet `variant` visitor method. This PR adds a Variant visitor that produces Parquet readers that handle metadata, serialized values, and shredded values.
Tests are in `TestVariantReaders`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936493604
##########
parquet/src/main/java/org/apache/iceberg/parquet/TypeToMessageType.java:
##########
@@ -117,6 +120,14 @@ public GroupType map(MapType map, Type.Repetition repetition, int id, String nam
.named(AvroSchemaUtil.makeCompatibleName(name));
}
+ public Type variant(Type.Repetition repetition, int id, String originalName) {
Review Comment:
Implements converting an Iceberg variant schema into a Parquet schema, without shredding.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941507423
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
+
+ @SuppressWarnings("unchecked")
+ public static ParquetValueReader<Variant> variant(
+ ParquetValueReader<?> metadata, ParquetValueReader<?> value) {
+ return new VariantReader(
+ (ParquetValueReader<VariantMetadata>) metadata, (VariantValueReader) value);
+ }
+
+ public static ParquetValueReader<VariantMetadata> metadata(ColumnDescriptor desc) {
+ return new VariantMetadataReader(desc);
+ }
+
+ public static VariantValueReader serialized(ColumnDescriptor desc) {
+ return new SerializedVariantReader(desc);
+ }
+
+ public static VariantValueReader shredded(
+ int valueDL,
Review Comment:
why not use fully expanded name? DL isn't exactly clear if you don't already know what it's referring to. Doesn't look like it would affect formatting either.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936493915
##########
parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java:
##########
@@ -54,101 +57,108 @@ public static <T> T visit(
} else {
// if not a primitive, the typeId must be a group
GroupType group = type.asGroupType();
- OriginalType annotation = group.getOriginalType();
Review Comment:
The existing code was refactored into methods for each case. Handling for LIST and MAP did not change.
##########
parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java:
##########
@@ -54,101 +57,108 @@ public static <T> T visit(
} else {
// if not a primitive, the typeId must be a group
GroupType group = type.asGroupType();
- OriginalType annotation = group.getOriginalType();
Review Comment:
The existing code was refactored into methods for each case. Handling for LIST and MAP and struct fields did not change.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941578309
##########
parquet/src/main/java/org/apache/iceberg/parquet/TypeToMessageType.java:
##########
@@ -86,6 +86,9 @@ public Type field(NestedField field) {
if (field.type().isPrimitiveType()) {
return primitive(field.type().asPrimitiveType(), repetition, id, name);
+ } else if (field.type() instanceof org.apache.iceberg.types.Types.VariantType) {
Review Comment:
Same 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1960602946
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantVisitor.java:
##########
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation.ListLogicalTypeAnnotation;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+
+public abstract class ParquetVariantVisitor<R> {
+ static final String METADATA = "metadata";
+ static final String VALUE = "value";
+ static final String TYPED_VALUE = "typed_value";
+
+ /**
+ * Handles the root variant column group.
+ *
+ * <p>The value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group v (VARIANT) { <-- metadata result and combined value and typed_value result
+ * required binary metadata;
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R variant(GroupType variant, R metadataResult, R valueResult) {
+ return null;
+ }
+
+ /**
+ * Handles a serialized variant metadata column.
+ *
+ * <pre>
+ * group v (VARIANT) {
+ * required binary metadata; <-- this column
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R metadata(PrimitiveType metadata) {
+ return null;
+ }
+
+ /**
+ * Handles a serialized variant value column.
+ *
+ * <pre>
+ * group variant_value_pair {
+ * optional binary value; <-- this column
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R serialized(PrimitiveType value) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded primitive typed_value column.
+ *
+ * <pre>
+ * group variant_value_pair {
+ * optional binary value;
+ * optional int32 typed_value; <-- this column when it is any primitive
+ * }
+ * </pre>
+ */
+ public R primitive(PrimitiveType primitive) {
+ return null;
+ }
+
+ /**
+ * Handles a variant value result and typed_value result pair.
+ *
+ * <p>The value and typed_value pair may be nested in an object field, array element, or in the
+ * root group of a variant.
+ *
+ * <p>This method is also called when the typed_value field is missing.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and typed_value result
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R value(GroupType value, R valueResult, R typedResult) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded object value result and a list of field value results.
+ *
+ * <p>Each field's value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and typed_value field results
+ * optional binary value;
+ * optional group typed_value {
+ * required group a {
+ * optional binary value;
+ * optional binary typed_value (UTF8);
+ * }
+ * ...
+ * }
+ * }
+ * </pre>
+ */
+ public R object(GroupType object, R valueResult, List<R> fieldResults) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded array value result and an element value result.
+ *
+ * <p>The element's value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and element result
+ * optional binary value;
+ * optional group typed_value (LIST) {
+ * repeated group list {
+ * required group element {
+ * optional binary value;
+ * optional binary typed_value (UTF8);
+ * }
+ * }
+ * }
+ * }
+ * </pre>
+ */
+ public R array(GroupType array, R valueResult, R elementResult) {
+ return null;
+ }
+
+ /** Handler called before visiting any primitive or group type. */
+ public void beforeField(Type type) {}
+
+ /** Handler called after visiting any primitive or group type. */
+ public void afterField(Type type) {}
+
+ public static <R> R visit(GroupType type, ParquetVariantVisitor<R> visitor) {
+ Preconditions.checkArgument(
+ ParquetSchemaUtil.hasField(type, METADATA), "Invalid variant, missing metadata: %s", type);
+
+ Type metadataType = type.getType(METADATA);
+ Preconditions.checkArgument(
+ isBinary(metadataType), "Invalid variant metadata, expecting BINARY: %s", metadataType);
+
+ R metadataResult =
+ withBeforeAndAfter(
+ () -> visitor.metadata(metadataType.asPrimitiveType()), metadataType, visitor);
+ R valueResult = visitValue(type, visitor);
+
+ return visitor.variant(type, metadataResult, valueResult);
+ }
+
+ public static <R> R visitValue(GroupType valueGroup, ParquetVariantVisitor<R> visitor) {
Review Comment:
Fixed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue merged PR #12139:
URL: https://github.com/apache/iceberg/pull/12139
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943539725
##########
core/src/main/java/org/apache/iceberg/variants/Variants.java:
##########
@@ -209,59 +213,59 @@ public static VariantPrimitive<Void> ofNull() {
return new PrimitiveWrapper<>(PhysicalType.NULL, null);
}
- static VariantPrimitive<Boolean> of(boolean value) {
+ public static VariantPrimitive<Boolean> of(boolean value) {
Review Comment:
I think that we're going to want these in order to be able to construct variants anyway, so I'm inclined to leave them as public, at least in core.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941577496
##########
parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java:
##########
@@ -162,9 +168,11 @@ private Integer getId(Type type) {
return type.getId() == null ? null : type.getId().intValue();
}
- private boolean isStruct(Type field) {
+ private boolean isStruct(Type field, NestedField expected) {
if (field.isPrimitive()) {
return false;
+ } else if (expected.type() == org.apache.iceberg.types.Types.VariantType.get()) {
Review Comment:
Here's another case where I feel like the `Type:isVariantType` would make things more clear.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943800768
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
Review Comment:
Yes, I considered using the existing `ParquetValueReader` interface with different type parameters. The problem that this is solving doesn't fit well with reusing this though. The need for a new `read` method is to pass the root variant's `VariantMetadata` down to children because its dictionary is used for all of the object field names in the current value structure.
To pass `VariantMetadata` down, we could use `ParquetValueReader<VariantMetadata>` but then the `read` method would return `VariantMetadata`. To overcome that, we could use `Variant` to pass metadata into read and a value back out, but then each value is wrapped in an unnecessary `Variant` that needs to be removed (same with other cleverness, like `OneOf<VariantMetadata, VariantValue>`). The only workable option is to parameterize by `Object` and pass whatever we want.
If we were to parameterize by object there's a second problem, which is that using the `reuse` value for that purpose is actually violating the contract of the `read` method. So my rationale was that as long as we are changing the contract, we should just state what we are changing the contract to by adding a new interface that encapsulates it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936491220
##########
api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java:
##########
@@ -47,7 +47,7 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu
@Override
public Set<Integer> field(Types.NestedField field, Set<Integer> fieldResult) {
- if ((includeStructIds && field.type().isStructType()) || field.type().isPrimitiveType()) {
+ if ((includeStructIds && field.type().isStructType()) || field.type().isPrimitiveType() || field.type() instanceof Types.VariantType) {
Review Comment:
This was needed for reading Parquet files without dropping variant columns.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941495914
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
Review Comment:
I feel like this makes working with value reads a little awkward for variant since you'll need to have a concrete interface. Have you considered having an interface that implements both `VarientMetadata` and `VarientValue` (e.g. `VariantRecrod` or similar) to bridge this gap?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943535834
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
+
+ @SuppressWarnings("unchecked")
+ public static ParquetValueReader<Variant> variant(
+ ParquetValueReader<?> metadata, ParquetValueReader<?> value) {
+ return new VariantReader(
+ (ParquetValueReader<VariantMetadata>) metadata, (VariantValueReader) value);
+ }
+
+ public static ParquetValueReader<VariantMetadata> metadata(ColumnDescriptor desc) {
+ return new VariantMetadataReader(desc);
+ }
+
+ public static VariantValueReader serialized(ColumnDescriptor desc) {
+ return new SerializedVariantReader(desc);
+ }
+
+ public static VariantValueReader shredded(
+ int valueDL,
+ ParquetValueReader<?> valueReader,
+ int typedDL,
+ ParquetValueReader<?> typedReader) {
+ return new ShreddedVariantReader(
+ valueDL, (VariantValueReader) valueReader, typedDL, (VariantValueReader) typedReader);
+ }
+
+ public static VariantValueReader objects(
+ int valueDL,
+ ParquetValueReader<?> valueReader,
+ int fieldDL,
+ List<String> fieldNames,
+ List<VariantValueReader> fieldReaders) {
+ return new ShreddedObjectReader(
+ valueDL, (VariantValueReader) valueReader, fieldDL, fieldNames, fieldReaders);
+ }
+
+ public static VariantValueReader asVariant(PhysicalType type, ParquetValueReader<?> reader) {
+ return new ValueAsVariantReader<>(type, reader);
+ }
+
+ private abstract static class DelegatingValueReader<S, T> implements ParquetValueReader<T> {
+ private final ParquetValueReader<S> reader;
+
+ private DelegatingValueReader(ParquetValueReader<S> reader) {
+ this.reader = reader;
+ }
+
+ protected S readFromDelegate(S reuse) {
+ return reader.read(reuse);
+ }
+
+ @Override
+ public TripleIterator<?> column() {
+ return reader.column();
+ }
+
+ @Override
+ public List<TripleIterator<?>> columns() {
+ return reader.columns();
+ }
+
+ @Override
+ public void setPageSource(PageReadStore pageStore) {
+ reader.setPageSource(pageStore);
+ }
+ }
+
+ private static ByteBuffer readBinary(ColumnIterator<?> column) {
+ ByteBuffer data = column.nextBinary().toByteBuffer();
+ byte[] array = new byte[data.remaining()];
+ data.get(array, 0, data.remaining());
+ return ByteBuffer.wrap(array).order(ByteOrder.LITTLE_ENDIAN);
+ }
+
+ private static class VariantMetadataReader extends PrimitiveReader<VariantMetadata> {
+ private VariantMetadataReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantMetadata read(VariantMetadata reuse) {
+ return Variants.metadata(readBinary(column));
+ }
+ }
+
+ private static class SerializedVariantReader extends PrimitiveReader<VariantValue>
+ implements VariantValueReader {
+ private SerializedVariantReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ return Variants.value(metadata, readBinary(column));
+ }
+ }
+
+ private static class ValueAsVariantReader<V> extends DelegatingValueReader<V, VariantValue>
+ implements VariantValueReader {
+ private final PhysicalType type;
+
+ private ValueAsVariantReader(PhysicalType type, ParquetValueReader<V> reader) {
+ super(reader);
+ this.type = type;
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata ignored) {
+ return Variants.of(type, readFromDelegate(null));
+ }
+ }
+
+ /**
+ * A Variant reader that combines value and typed_value columns from Parquet.
+ *
+ * <p>This reader does not handle merging partially shredded objects. To handle shredded objects,
+ * use {@link ShreddedObjectReader}.
+ */
+ private static class ShreddedVariantReader implements VariantValueReader {
+ private final int valueDL;
+ private final VariantValueReader valueReader;
+ private final int typedDL;
+ private final VariantValueReader typedReader;
+ private final TripleIterator<?> column;
+ private final List<TripleIterator<?>> children;
+
+ private ShreddedVariantReader(
+ int valueDL, VariantValueReader valueReader, int typedDL, VariantValueReader typedReader) {
+ this.valueDL = valueDL;
+ this.valueReader = valueReader;
+ this.typedDL = typedDL;
+ this.typedReader = typedReader;
+ this.column = valueReader != null ? valueReader.column() : typedReader.column();
+ this.children = children(valueReader, typedReader);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ VariantValue value = ParquetVariantReaders.read(metadata, valueReader, valueDL);
+ VariantValue typed = ParquetVariantReaders.read(metadata, typedReader, typedDL);
+
+ if (typed != null) {
Review Comment:
Yes, I'll update this to implement the latest version of the spec.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943818949
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
+
+ @SuppressWarnings("unchecked")
+ public static ParquetValueReader<Variant> variant(
+ ParquetValueReader<?> metadata, ParquetValueReader<?> value) {
+ return new VariantReader(
+ (ParquetValueReader<VariantMetadata>) metadata, (VariantValueReader) value);
+ }
+
+ public static ParquetValueReader<VariantMetadata> metadata(ColumnDescriptor desc) {
+ return new VariantMetadataReader(desc);
+ }
+
+ public static VariantValueReader serialized(ColumnDescriptor desc) {
+ return new SerializedVariantReader(desc);
+ }
+
+ public static VariantValueReader shredded(
+ int valueDL,
+ ParquetValueReader<?> valueReader,
+ int typedDL,
+ ParquetValueReader<?> typedReader) {
+ return new ShreddedVariantReader(
+ valueDL, (VariantValueReader) valueReader, typedDL, (VariantValueReader) typedReader);
+ }
+
+ public static VariantValueReader objects(
+ int valueDL,
+ ParquetValueReader<?> valueReader,
+ int fieldDL,
+ List<String> fieldNames,
+ List<VariantValueReader> fieldReaders) {
+ return new ShreddedObjectReader(
+ valueDL, (VariantValueReader) valueReader, fieldDL, fieldNames, fieldReaders);
+ }
+
+ public static VariantValueReader asVariant(PhysicalType type, ParquetValueReader<?> reader) {
+ return new ValueAsVariantReader<>(type, reader);
+ }
+
+ private abstract static class DelegatingValueReader<S, T> implements ParquetValueReader<T> {
+ private final ParquetValueReader<S> reader;
+
+ private DelegatingValueReader(ParquetValueReader<S> reader) {
+ this.reader = reader;
+ }
+
+ protected S readFromDelegate(S reuse) {
+ return reader.read(reuse);
+ }
+
+ @Override
+ public TripleIterator<?> column() {
+ return reader.column();
+ }
+
+ @Override
+ public List<TripleIterator<?>> columns() {
+ return reader.columns();
+ }
+
+ @Override
+ public void setPageSource(PageReadStore pageStore) {
+ reader.setPageSource(pageStore);
+ }
+ }
+
+ private static ByteBuffer readBinary(ColumnIterator<?> column) {
+ ByteBuffer data = column.nextBinary().toByteBuffer();
+ byte[] array = new byte[data.remaining()];
+ data.get(array, 0, data.remaining());
+ return ByteBuffer.wrap(array).order(ByteOrder.LITTLE_ENDIAN);
+ }
+
+ private static class VariantMetadataReader extends PrimitiveReader<VariantMetadata> {
+ private VariantMetadataReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantMetadata read(VariantMetadata reuse) {
+ return Variants.metadata(readBinary(column));
+ }
+ }
+
+ private static class SerializedVariantReader extends PrimitiveReader<VariantValue>
+ implements VariantValueReader {
+ private SerializedVariantReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ return Variants.value(metadata, readBinary(column));
+ }
+ }
+
+ private static class ValueAsVariantReader<V> extends DelegatingValueReader<V, VariantValue>
+ implements VariantValueReader {
+ private final PhysicalType type;
+
+ private ValueAsVariantReader(PhysicalType type, ParquetValueReader<V> reader) {
+ super(reader);
+ this.type = type;
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata ignored) {
+ return Variants.of(type, readFromDelegate(null));
+ }
+ }
+
+ /**
+ * A Variant reader that combines value and typed_value columns from Parquet.
+ *
+ * <p>This reader does not handle merging partially shredded objects. To handle shredded objects,
+ * use {@link ShreddedObjectReader}.
+ */
+ private static class ShreddedVariantReader implements VariantValueReader {
+ private final int valueDL;
+ private final VariantValueReader valueReader;
+ private final int typedDL;
+ private final VariantValueReader typedReader;
+ private final TripleIterator<?> column;
+ private final List<TripleIterator<?>> children;
+
+ private ShreddedVariantReader(
+ int valueDL, VariantValueReader valueReader, int typedDL, VariantValueReader typedReader) {
+ this.valueDL = valueDL;
+ this.valueReader = valueReader;
+ this.typedDL = typedDL;
+ this.typedReader = typedReader;
+ this.column = valueReader != null ? valueReader.column() : typedReader.column();
+ this.children = children(valueReader, typedReader);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ VariantValue value = ParquetVariantReaders.read(metadata, valueReader, valueDL);
+ VariantValue typed = ParquetVariantReaders.read(metadata, typedReader, typedDL);
+
+ if (typed != null) {
Review Comment:
Fixed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941495914
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
Review Comment:
I feel like this makes working with value reads a little awkward for variant since you'll need to have a concrete interface. Have you considered having an interface that implements both `VarientMetadata` and `VarientEntry` (e.g. `VariantRecrod` or similar) to bridge this gap?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936493090
##########
parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java:
##########
@@ -162,9 +168,11 @@ private Integer getId(Type type) {
return type.getId() == null ? null : type.getId().intValue();
}
- private boolean isStruct(Type field) {
+ private boolean isStruct(Type field, NestedField expected) {
if (field.isPrimitive()) {
return false;
+ } else if (expected.type() == org.apache.iceberg.types.Types.VariantType.get()) {
Review Comment:
These changes were needed to project the sub-fields of a variant because it looks like a struct.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936491473
##########
core/src/main/java/org/apache/iceberg/variants/Variants.java:
##########
@@ -209,59 +213,59 @@ public static VariantPrimitive<Void> ofNull() {
return new PrimitiveWrapper<>(PhysicalType.NULL, null);
}
- static VariantPrimitive<Boolean> of(boolean value) {
+ public static VariantPrimitive<Boolean> of(boolean value) {
Review Comment:
Needed to expose these to create test values in the Parquet package.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1960605176
##########
core/src/test/java/org/apache/iceberg/variants/VariantTestUtil.java:
##########
@@ -27,10 +29,55 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
public class VariantTestUtil {
private VariantTestUtil() {}
+ public static void assertEqual(VariantMetadata expected, VariantMetadata actual) {
+ assertThat(actual).isNotNull();
+ assertThat(expected).isNotNull();
+ assertThat(actual.dictionarySize())
+ .as("Dictionary size should match")
+ .isEqualTo(expected.dictionarySize());
+
+ for (int i = 0; i < expected.dictionarySize(); i += 1) {
+ assertThat(actual.get(i)).isEqualTo(expected.get(i));
Review Comment:
The mapping from ID to string must match. We don't really care if the `sorted` bit is set or not, but the mapping must be identical or else the data is changed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941571976
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
Review Comment:
Shouldn't we have a concrete instance of `VariantValue` for `MISSING` rather than use `null`? I suspect it might be possible to handle everything this way, but if there's a mistake then equality checks will reflect the same for both missing and null, which could be difficult to track down.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943536772
##########
parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java:
##########
@@ -162,9 +168,11 @@ private Integer getId(Type type) {
return type.getId() == null ? null : type.getId().intValue();
}
- private boolean isStruct(Type field) {
+ private boolean isStruct(Type field, NestedField expected) {
if (field.isPrimitive()) {
return false;
+ } else if (expected.type() == org.apache.iceberg.types.Types.VariantType.get()) {
Review Comment:
Yes, I'll update this when the other PR from @aihuaxu is in.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "aihuaxu (via GitHub)" <gi...@apache.org>.
aihuaxu commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1958863116
##########
core/src/test/java/org/apache/iceberg/variants/VariantTestUtil.java:
##########
@@ -27,10 +29,55 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
public class VariantTestUtil {
private VariantTestUtil() {}
+ public static void assertEqual(VariantMetadata expected, VariantMetadata actual) {
+ assertThat(actual).isNotNull();
+ assertThat(expected).isNotNull();
+ assertThat(actual.dictionarySize())
+ .as("Dictionary size should match")
+ .isEqualTo(expected.dictionarySize());
+
+ for (int i = 0; i < expected.dictionarySize(); i += 1) {
+ assertThat(actual.get(i)).isEqualTo(expected.get(i));
+ }
+ }
+
+ public static void assertEqual(VariantValue expected, VariantValue actual) {
+ assertThat(actual).isNotNull();
+ assertThat(expected).isNotNull();
+ assertThat(actual.type()).as("Variant type should match").isEqualTo(expected.type());
+
+ if (expected.type() == PhysicalType.OBJECT) {
+ VariantObject expectedObject = expected.asObject();
+ VariantObject actualObject = actual.asObject();
+ assertThat(actualObject.numFields())
+ .as("Variant object num fields should match")
+ .isEqualTo(expectedObject.numFields());
+ for (String fieldName : expectedObject.fieldNames()) {
+ assertEqual(expectedObject.get(fieldName), actualObject.get(fieldName));
+ }
+
Review Comment:
nit: remove extra empty line. Not exactly sure if this is expected code style in Iceberg.
##########
core/src/test/java/org/apache/iceberg/variants/VariantTestUtil.java:
##########
@@ -27,10 +29,55 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
public class VariantTestUtil {
private VariantTestUtil() {}
+ public static void assertEqual(VariantMetadata expected, VariantMetadata actual) {
+ assertThat(actual).isNotNull();
+ assertThat(expected).isNotNull();
+ assertThat(actual.dictionarySize())
+ .as("Dictionary size should match")
+ .isEqualTo(expected.dictionarySize());
+
+ for (int i = 0; i < expected.dictionarySize(); i += 1) {
+ assertThat(actual.get(i)).isEqualTo(expected.get(i));
Review Comment:
For this check, if one is sorted and the other is unsorted, then we are considering them different, right? Just want to confirm.
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantVisitor.java:
##########
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation.ListLogicalTypeAnnotation;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+
+public abstract class ParquetVariantVisitor<R> {
+ static final String METADATA = "metadata";
+ static final String VALUE = "value";
+ static final String TYPED_VALUE = "typed_value";
+
+ /**
+ * Handles the root variant column group.
+ *
+ * <p>The value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group v (VARIANT) { <-- metadata result and combined value and typed_value result
+ * required binary metadata;
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R variant(GroupType variant, R metadataResult, R valueResult) {
+ return null;
+ }
+
+ /**
+ * Handles a serialized variant metadata column.
+ *
+ * <pre>
+ * group v (VARIANT) {
+ * required binary metadata; <-- this column
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R metadata(PrimitiveType metadata) {
+ return null;
+ }
+
+ /**
+ * Handles a serialized variant value column.
+ *
+ * <pre>
+ * group variant_value_pair {
+ * optional binary value; <-- this column
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R serialized(PrimitiveType value) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded primitive typed_value column.
+ *
+ * <pre>
+ * group variant_value_pair {
+ * optional binary value;
+ * optional int32 typed_value; <-- this column when it is any primitive
+ * }
+ * </pre>
+ */
+ public R primitive(PrimitiveType primitive) {
+ return null;
+ }
+
+ /**
+ * Handles a variant value result and typed_value result pair.
+ *
+ * <p>The value and typed_value pair may be nested in an object field, array element, or in the
+ * root group of a variant.
+ *
+ * <p>This method is also called when the typed_value field is missing.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and typed_value result
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R value(GroupType value, R valueResult, R typedResult) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded object value result and a list of field value results.
+ *
+ * <p>Each field's value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and typed_value field results
+ * optional binary value;
+ * optional group typed_value {
+ * required group a {
+ * optional binary value;
+ * optional binary typed_value (UTF8);
+ * }
+ * ...
+ * }
+ * }
+ * </pre>
+ */
+ public R object(GroupType object, R valueResult, List<R> fieldResults) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded array value result and an element value result.
+ *
+ * <p>The element's value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and element result
+ * optional binary value;
+ * optional group typed_value (LIST) {
+ * repeated group list {
+ * required group element {
+ * optional binary value;
+ * optional binary typed_value (UTF8);
+ * }
+ * }
+ * }
+ * }
+ * </pre>
+ */
+ public R array(GroupType array, R valueResult, R elementResult) {
+ return null;
+ }
+
+ /** Handler called before visiting any primitive or group type. */
+ public void beforeField(Type type) {}
+
+ /** Handler called after visiting any primitive or group type. */
+ public void afterField(Type type) {}
+
+ public static <R> R visit(GroupType type, ParquetVariantVisitor<R> visitor) {
+ Preconditions.checkArgument(
+ ParquetSchemaUtil.hasField(type, METADATA), "Invalid variant, missing metadata: %s", type);
+
+ Type metadataType = type.getType(METADATA);
+ Preconditions.checkArgument(
+ isBinary(metadataType), "Invalid variant metadata, expecting BINARY: %s", metadataType);
+
+ R metadataResult =
+ withBeforeAndAfter(
+ () -> visitor.metadata(metadataType.asPrimitiveType()), metadataType, visitor);
+ R valueResult = visitValue(type, visitor);
+
+ return visitor.variant(type, metadataResult, valueResult);
+ }
+
+ public static <R> R visitValue(GroupType valueGroup, ParquetVariantVisitor<R> visitor) {
Review Comment:
This can be private scope for now.
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestVariantReaders.java:
##########
@@ -0,0 +1,1107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.InternalReader;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Streams;
+import org.apache.iceberg.types.Types.IntegerType;
+import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.types.Types.VariantType;
+import org.apache.iceberg.variants.PhysicalType;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantPrimitive;
+import org.apache.iceberg.variants.VariantTestUtil;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.avro.AvroSchemaConverter;
+import org.apache.parquet.avro.AvroWriteSupport;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+import org.assertj.core.api.Assumptions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.FieldSource;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestVariantReaders {
+ private static final Schema SCHEMA =
+ new Schema(
+ NestedField.required(1, "id", IntegerType.get()),
+ NestedField.required(2, "var", VariantType.get()));
+
+ private static final LogicalTypeAnnotation STRING = LogicalTypeAnnotation.stringType();
+
+ private static final ByteBuffer TEST_METADATA_BUFFER =
+ VariantTestUtil.createMetadata(ImmutableList.of("a", "b", "c", "d", "e"), true);
+ private static final ByteBuffer TEST_OBJECT_BUFFER =
+ VariantTestUtil.createObject(
+ TEST_METADATA_BUFFER,
+ ImmutableMap.of(
+ "a", Variants.ofNull(),
+ "d", Variants.of(PhysicalType.STRING, "iceberg")));
+
+ private static final VariantMetadata EMPTY_METADATA =
+ Variants.metadata(VariantTestUtil.emptyMetadata());
+ private static final VariantMetadata TEST_METADATA = Variants.metadata(TEST_METADATA_BUFFER);
+ private static final VariantObject TEST_OBJECT =
+ (VariantObject) Variants.value(TEST_METADATA, TEST_OBJECT_BUFFER);
+
+ private static final VariantPrimitive<?>[] PRIMITIVES =
+ new VariantPrimitive[] {
+ Variants.ofNull(),
+ Variants.of(true),
+ Variants.of(false),
+ Variants.of((byte) 34),
+ Variants.of((byte) -34),
+ Variants.of((short) 1234),
+ Variants.of((short) -1234),
+ Variants.of(12345),
+ Variants.of(-12345),
+ Variants.of(9876543210L),
+ Variants.of(-9876543210L),
+ Variants.of(10.11F),
+ Variants.of(-10.11F),
+ Variants.of(14.3D),
+ Variants.of(-14.3D),
+ Variants.ofIsoDate("2024-11-07"),
+ Variants.ofIsoDate("1957-11-07"),
+ Variants.ofIsoTimestamptz("2024-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestamptz("1957-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestampntz("2024-11-07T12:33:54.123456"),
+ Variants.ofIsoTimestampntz("1957-11-07T12:33:54.123456"),
+ Variants.of(new BigDecimal("123456.7890")), // decimal4
+ Variants.of(new BigDecimal("-123456.7890")), // decimal4
+ Variants.of(new BigDecimal("1234567890.987654321")), // decimal8
+ Variants.of(new BigDecimal("-1234567890.987654321")), // decimal8
+ Variants.of(new BigDecimal("9876543210.123456789")), // decimal16
+ Variants.of(new BigDecimal("-9876543210.123456789")), // decimal16
+ Variants.of(ByteBuffer.wrap(new byte[] {0x0a, 0x0b, 0x0c, 0x0d})),
+ Variants.of("iceberg"),
+ };
+
+ private static Stream<Arguments> metadataAndValues() {
+ Stream<Arguments> primitives =
+ Stream.of(PRIMITIVES).map(variant -> Arguments.of(EMPTY_METADATA, variant));
+ Stream<Arguments> object = Stream.of(Arguments.of(TEST_METADATA, TEST_OBJECT));
+ return Streams.concat(primitives, object);
+ }
+
+ @ParameterizedTest
+ @MethodSource("metadataAndValues")
+ public void testUnshreddedVariants(VariantMetadata metadata, VariantValue expected)
+ throws IOException {
+ GroupType variantType = variant("var", 2);
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", serialize(metadata), "value", serialize(expected)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(metadata, actualVariant.metadata());
+ VariantTestUtil.assertEqual(expected, actualVariant.value());
+ }
+
+ @ParameterizedTest
+ @MethodSource("metadataAndValues")
+ public void testUnshreddedVariantsWithShreddedSchema(
+ VariantMetadata metadata, VariantValue expected) throws IOException {
+ // the variant's Parquet schema has a shredded field that is unused by all data values
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.BINARY, STRING));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", serialize(metadata), "value", serialize(expected)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(metadata, actualVariant.metadata());
+ VariantTestUtil.assertEqual(expected, actualVariant.value());
+ }
+
+ @ParameterizedTest
+ @FieldSource("PRIMITIVES")
+ public void testShreddedVariantPrimitives(VariantPrimitive<?> primitive) throws IOException {
+ Assumptions.assumeThat(primitive.type() != PhysicalType.NULL)
+ .as("Null is not a shredded type")
+ .isTrue();
+
+ GroupType variantType = variant("var", 2, shreddedType(primitive));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(
+ variantType,
+ Map.of(
+ "metadata",
+ VariantTestUtil.emptyMetadata(),
+ "typed_value",
+ toAvroValue(primitive)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(primitive, actualVariant.value());
+ }
+
+ @Test
+ public void testNullValueAndNullTypedValue() throws IOException {
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.INT32));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata()));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(Variants.ofNull(), actualVariant.value());
+ }
+
+ @Test
+ public void testMissingValueColumn() throws IOException {
+ GroupType variantType =
+ Types.buildGroup(Type.Repetition.REQUIRED)
+ .id(2)
+ .required(PrimitiveTypeName.BINARY)
+ .named("metadata")
+ .addField(shreddedPrimitive(PrimitiveTypeName.INT32))
+ .named("var");
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata(), "typed_value", 34));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(Variants.of(34), actualVariant.value());
+ }
+
+ @Test
+ public void testValueAndTypedValueConflict() throws IOException {
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.INT32));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(
+ variantType,
+ Map.of(
+ "metadata",
+ VariantTestUtil.emptyMetadata(),
+ "value",
+ serialize(Variants.of("str")),
+ "typed_value",
+ 34));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ assertThatThrownBy(() -> writeAndRead(parquetSchema, record))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid variant, conflicting value and typed_value");
+ }
+
+ @Test
+ public void testUnsignedInteger() {
+ GroupType variantType =
+ variant(
+ "var",
+ 2,
+ shreddedPrimitive(PrimitiveTypeName.INT32, LogicalTypeAnnotation.intType(32, false)));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata()));
Review Comment:
Addd typed_value
`record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata(), "typed_value", 34));` to have the shredded value?
##########
parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java:
##########
@@ -201,6 +211,17 @@ private static <T> List<T> visitFields(
return results;
}
+ private static <T> T visitVariant(
+ Types.VariantType variant, GroupType group, TypeWithSchemaVisitor<T> visitor) {
+ ParquetVariantVisitor<T> variantVisitor = visitor.variantVisitor();
+ if (variantVisitor != null) {
+ T variantResult = ParquetVariantVisitor.visit(group, variantVisitor);
Review Comment:
Rather than creating a static visit() function, can we use `variantVisitor.visit(group)`?
With that, we have many functions in ParquetVariantVisitor.java to be class instance method rather than static method. Seems that will make the code there simpler.
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestVariantReaders.java:
##########
@@ -0,0 +1,1107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.InternalReader;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Streams;
+import org.apache.iceberg.types.Types.IntegerType;
+import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.types.Types.VariantType;
+import org.apache.iceberg.variants.PhysicalType;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantPrimitive;
+import org.apache.iceberg.variants.VariantTestUtil;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.avro.AvroSchemaConverter;
+import org.apache.parquet.avro.AvroWriteSupport;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+import org.assertj.core.api.Assumptions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.FieldSource;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestVariantReaders {
+ private static final Schema SCHEMA =
+ new Schema(
+ NestedField.required(1, "id", IntegerType.get()),
+ NestedField.required(2, "var", VariantType.get()));
+
+ private static final LogicalTypeAnnotation STRING = LogicalTypeAnnotation.stringType();
+
+ private static final ByteBuffer TEST_METADATA_BUFFER =
+ VariantTestUtil.createMetadata(ImmutableList.of("a", "b", "c", "d", "e"), true);
+ private static final ByteBuffer TEST_OBJECT_BUFFER =
+ VariantTestUtil.createObject(
+ TEST_METADATA_BUFFER,
+ ImmutableMap.of(
+ "a", Variants.ofNull(),
+ "d", Variants.of(PhysicalType.STRING, "iceberg")));
+
+ private static final VariantMetadata EMPTY_METADATA =
+ Variants.metadata(VariantTestUtil.emptyMetadata());
+ private static final VariantMetadata TEST_METADATA = Variants.metadata(TEST_METADATA_BUFFER);
+ private static final VariantObject TEST_OBJECT =
+ (VariantObject) Variants.value(TEST_METADATA, TEST_OBJECT_BUFFER);
+
+ private static final VariantPrimitive<?>[] PRIMITIVES =
+ new VariantPrimitive[] {
+ Variants.ofNull(),
+ Variants.of(true),
+ Variants.of(false),
+ Variants.of((byte) 34),
+ Variants.of((byte) -34),
+ Variants.of((short) 1234),
+ Variants.of((short) -1234),
+ Variants.of(12345),
+ Variants.of(-12345),
+ Variants.of(9876543210L),
+ Variants.of(-9876543210L),
+ Variants.of(10.11F),
+ Variants.of(-10.11F),
+ Variants.of(14.3D),
+ Variants.of(-14.3D),
+ Variants.ofIsoDate("2024-11-07"),
+ Variants.ofIsoDate("1957-11-07"),
+ Variants.ofIsoTimestamptz("2024-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestamptz("1957-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestampntz("2024-11-07T12:33:54.123456"),
+ Variants.ofIsoTimestampntz("1957-11-07T12:33:54.123456"),
+ Variants.of(new BigDecimal("123456.7890")), // decimal4
+ Variants.of(new BigDecimal("-123456.7890")), // decimal4
+ Variants.of(new BigDecimal("1234567890.987654321")), // decimal8
+ Variants.of(new BigDecimal("-1234567890.987654321")), // decimal8
+ Variants.of(new BigDecimal("9876543210.123456789")), // decimal16
+ Variants.of(new BigDecimal("-9876543210.123456789")), // decimal16
+ Variants.of(ByteBuffer.wrap(new byte[] {0x0a, 0x0b, 0x0c, 0x0d})),
+ Variants.of("iceberg"),
+ };
+
+ private static Stream<Arguments> metadataAndValues() {
+ Stream<Arguments> primitives =
+ Stream.of(PRIMITIVES).map(variant -> Arguments.of(EMPTY_METADATA, variant));
+ Stream<Arguments> object = Stream.of(Arguments.of(TEST_METADATA, TEST_OBJECT));
+ return Streams.concat(primitives, object);
+ }
+
+ @ParameterizedTest
+ @MethodSource("metadataAndValues")
+ public void testUnshreddedVariants(VariantMetadata metadata, VariantValue expected)
+ throws IOException {
+ GroupType variantType = variant("var", 2);
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", serialize(metadata), "value", serialize(expected)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(metadata, actualVariant.metadata());
+ VariantTestUtil.assertEqual(expected, actualVariant.value());
+ }
+
+ @ParameterizedTest
+ @MethodSource("metadataAndValues")
+ public void testUnshreddedVariantsWithShreddedSchema(
+ VariantMetadata metadata, VariantValue expected) throws IOException {
+ // the variant's Parquet schema has a shredded field that is unused by all data values
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.BINARY, STRING));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", serialize(metadata), "value", serialize(expected)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(metadata, actualVariant.metadata());
+ VariantTestUtil.assertEqual(expected, actualVariant.value());
+ }
+
+ @ParameterizedTest
+ @FieldSource("PRIMITIVES")
+ public void testShreddedVariantPrimitives(VariantPrimitive<?> primitive) throws IOException {
+ Assumptions.assumeThat(primitive.type() != PhysicalType.NULL)
+ .as("Null is not a shredded type")
+ .isTrue();
+
+ GroupType variantType = variant("var", 2, shreddedType(primitive));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(
+ variantType,
+ Map.of(
+ "metadata",
+ VariantTestUtil.emptyMetadata(),
+ "typed_value",
+ toAvroValue(primitive)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(primitive, actualVariant.value());
+ }
+
+ @Test
+ public void testNullValueAndNullTypedValue() throws IOException {
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.INT32));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata()));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(Variants.ofNull(), actualVariant.value());
+ }
+
+ @Test
+ public void testMissingValueColumn() throws IOException {
+ GroupType variantType =
+ Types.buildGroup(Type.Repetition.REQUIRED)
+ .id(2)
+ .required(PrimitiveTypeName.BINARY)
+ .named("metadata")
+ .addField(shreddedPrimitive(PrimitiveTypeName.INT32))
+ .named("var");
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata(), "typed_value", 34));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(Variants.of(34), actualVariant.value());
+ }
+
+ @Test
+ public void testValueAndTypedValueConflict() throws IOException {
Review Comment:
IOException is not thrown 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1938261512
##########
core/src/main/java/org/apache/iceberg/variants/Variants.java:
##########
@@ -209,59 +213,59 @@ public static VariantPrimitive<Void> ofNull() {
return new PrimitiveWrapper<>(PhysicalType.NULL, null);
}
- static VariantPrimitive<Boolean> of(boolean value) {
+ public static VariantPrimitive<Boolean> of(boolean value) {
Review Comment:
I think it would be fine to expose it here, but it we could also add some builders in the test module if we want to not have as much exposure. So TestParquet depends on TestCore and core has public builders for Variants.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936494392
##########
parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java:
##########
@@ -217,11 +238,19 @@ public T map(Types.MapType iMap, GroupType map, T key, T value) {
return null;
}
+ public T variant(Types.VariantType iVariant, T result) {
+ throw new UnsupportedOperationException("Not implemented for variant");
Review Comment:
Thrown to force implementations to update in order to handle variants.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936491857
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java:
##########
@@ -75,6 +76,32 @@ private static Schema convertInternal(
converter.getAliases());
}
+ /**
+ * Returns true if the name identifies a field in the struct/group.
+ *
+ * @param group a GroupType
+ * @param name a String name
+ * @return true if the group contains a field with the given name
+ */
+ public static boolean hasField(GroupType group, String name) {
Review Comment:
New helper methods for working with Parquet fields.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941495914
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
Review Comment:
I feel like this makes working with value reads a little awkward for variant since you'll need to have a concrete interface. Have you considered having an interface that implements (or wraps) both `VarientMetadata` and `VarientEntry` (e.g. `VariantRecrod` or similar) to bridge this gap?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1960241060
##########
parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java:
##########
@@ -201,6 +211,17 @@ private static <T> List<T> visitFields(
return results;
}
+ private static <T> T visitVariant(
+ Types.VariantType variant, GroupType group, TypeWithSchemaVisitor<T> visitor) {
+ ParquetVariantVisitor<T> variantVisitor = visitor.variantVisitor();
+ if (variantVisitor != null) {
+ T variantResult = ParquetVariantVisitor.visit(group, variantVisitor);
Review Comment:
The project uses visitors to separate the logic of traversing a type from the logic of what to do with parts of the structure. While you could put the logic in `visit` into an instance method of the visitor (or other places) I think it is important to have that traversal logic in only one place across visitors with the same pattern (i.e. `TypeWithSchemaVisitor`). If we allow customization of the `visit` method, I think we would soon defeat the purpose of the visitors.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#issuecomment-2667157868
Thanks for reviewing, @aihuaxu, @danielcweeks, @RussellSpitzer, and @amogh-jahagirdar!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943546072
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
Review Comment:
I don't think so given how this is used. This constant is intended to make code more clear in places where the value may be `null` in Parquet or may be a non-null `VariantValue` that represents null (`00`).
That way we don't have code that looks like `if (value == null) { ... }` because it is confusing that the value could be non-null but represent null. Instead, I wanted the code to be either `if (value == MISSING) { ... }` or `if (value.type() == PhysicalType.NULL) { ... }`.
There also isn't a non-null Variant value that represents _missing_. A value is missing when it was not encoded in Parquet, as with fields that are in the shredding schema but not present in an object so the field's `value` and `typed_value` are both null.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941486767
##########
api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java:
##########
@@ -47,7 +47,9 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu
@Override
public Set<Integer> field(Types.NestedField field, Set<Integer> fieldResult) {
- if ((includeStructIds && field.type().isStructType()) || field.type().isPrimitiveType()) {
+ if ((includeStructIds && field.type().isStructType())
+ || field.type().isPrimitiveType()
+ || field.type() instanceof Types.VariantType) {
Review Comment:
Variant is really a new class of type (it's not primitive, really). I feel like we should move this to `Type::isVariantType()` and perform the check there. I'm just concerned that this type of checks will proliferate and be less readable.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941890941
##########
api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java:
##########
@@ -47,7 +47,9 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu
@Override
public Set<Integer> field(Types.NestedField field, Set<Integer> fieldResult) {
- if ((includeStructIds && field.type().isStructType()) || field.type().isPrimitiveType()) {
+ if ((includeStructIds && field.type().isStructType())
+ || field.type().isPrimitiveType()
+ || field.type() instanceof Types.VariantType) {
Review Comment:
Yes, this new method is part of #11831 (https://github.com/apache/iceberg/pull/11831/files#diff-07bf8c52bb598ed9e1f2ad684e0afa646673cf380a24f286ea320e0dba039bbaR101) from @aihuaxu. I just didn't want to add the same change here so I worked around it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1936492645
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantVisitor.java:
##########
@@ -0,0 +1,287 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one
+ * * or more contributor license agreements. See the NOTICE file
+ * * distributed with this work for additional information
+ * * regarding copyright ownership. The ASF licenses this file
+ * * to you under the Apache License, Version 2.0 (the
+ * * "License"); you may not use this file except in compliance
+ * * with the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing,
+ * * software distributed under the License is distributed on an
+ * * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * * KIND, either express or implied. See the License for the
+ * * specific language governing permissions and limitations
+ * * under the License.
+ *
+ */
+package org.apache.iceberg.parquet;
+
+import java.util.List;
+import org.apache.commons.compress.utils.Lists;
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation.ListLogicalTypeAnnotation;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+
+public abstract class ParquetVariantVisitor<R> {
+ static final String METADATA = "metadata";
+ static final String VALUE = "value";
+ static final String TYPED_VALUE = "typed_value";
+
+ /**
+ * Handles the root variant column group.
+ *
+ * <p>The value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group v (VARIANT) { <-- metadata result and combined value and typed_value result
+ * required binary metadata;
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R variant(GroupType variant, R metadataResult, R valueResult) {
+ return null;
+ }
+
+ /**
+ * Handles a serialized variant metadata column.
+ *
+ * <pre>
+ * group v (VARIANT) {
+ * required binary metadata; <-- this column
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R metadata(PrimitiveType metadata) {
+ return null;
+ }
+
+ /**
+ * Handles a serialized variant value column.
+ *
+ * <pre>
+ * group variant_value_pair {
+ * optional binary value; <-- this column
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R serialized(PrimitiveType value) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded primitive typed_value column.
+ *
+ * <pre>
+ * group variant_value_pair {
+ * optional binary value;
+ * optional int32 typed_value; <-- this column when it is any primitive
+ * }
+ * </pre>
+ */
+ public R primitive(PrimitiveType primitive) {
+ return null;
+ }
+
+ /**
+ * Handles a variant value result and typed_value result pair.
+ *
+ * <p>The value and typed_value pair may be nested in an object field, array element, or in the
+ * root group of a variant.
+ *
+ * <p>This method is also called when the typed_value field is missing.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and typed_value result
+ * optional binary value;
+ * optional ... typed_value;
+ * }
+ * </pre>
+ */
+ public R value(GroupType value, R valueResult, R typedResult) {
+ return null;
+ }
+
+ /**
+ * Handles a shredded object value result and a list of field value results.
+ *
+ * <p>Each field's value and typed_value results are combined by calling {@link #value}.
+ *
+ * <pre>
+ * group variant_value_pair { <-- value result and typed_value field results
+ * optional binary value;
+ * optional group typed_value {
+ * required group a {
+ * optional binary value;
+ * optional binary typed_value (UTF8);
+ * }
+ * ...
+ * }
+ * }
+ * </pre>
+ */
+ public R object(GroupType object, R valueResult, List<R> fieldResults) {
Review Comment:
The `value`, `object`, and `array` visitor methods are all used for a `value` / `typed_value` pair, depending on the structure of typed value (a shredded value, shredded array, or shredded object).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1941558670
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
+
+ @SuppressWarnings("unchecked")
+ public static ParquetValueReader<Variant> variant(
+ ParquetValueReader<?> metadata, ParquetValueReader<?> value) {
+ return new VariantReader(
+ (ParquetValueReader<VariantMetadata>) metadata, (VariantValueReader) value);
+ }
+
+ public static ParquetValueReader<VariantMetadata> metadata(ColumnDescriptor desc) {
+ return new VariantMetadataReader(desc);
+ }
+
+ public static VariantValueReader serialized(ColumnDescriptor desc) {
+ return new SerializedVariantReader(desc);
+ }
+
+ public static VariantValueReader shredded(
+ int valueDL,
+ ParquetValueReader<?> valueReader,
+ int typedDL,
+ ParquetValueReader<?> typedReader) {
+ return new ShreddedVariantReader(
+ valueDL, (VariantValueReader) valueReader, typedDL, (VariantValueReader) typedReader);
+ }
+
+ public static VariantValueReader objects(
+ int valueDL,
+ ParquetValueReader<?> valueReader,
+ int fieldDL,
+ List<String> fieldNames,
+ List<VariantValueReader> fieldReaders) {
+ return new ShreddedObjectReader(
+ valueDL, (VariantValueReader) valueReader, fieldDL, fieldNames, fieldReaders);
+ }
+
+ public static VariantValueReader asVariant(PhysicalType type, ParquetValueReader<?> reader) {
+ return new ValueAsVariantReader<>(type, reader);
+ }
+
+ private abstract static class DelegatingValueReader<S, T> implements ParquetValueReader<T> {
+ private final ParquetValueReader<S> reader;
+
+ private DelegatingValueReader(ParquetValueReader<S> reader) {
+ this.reader = reader;
+ }
+
+ protected S readFromDelegate(S reuse) {
+ return reader.read(reuse);
+ }
+
+ @Override
+ public TripleIterator<?> column() {
+ return reader.column();
+ }
+
+ @Override
+ public List<TripleIterator<?>> columns() {
+ return reader.columns();
+ }
+
+ @Override
+ public void setPageSource(PageReadStore pageStore) {
+ reader.setPageSource(pageStore);
+ }
+ }
+
+ private static ByteBuffer readBinary(ColumnIterator<?> column) {
+ ByteBuffer data = column.nextBinary().toByteBuffer();
+ byte[] array = new byte[data.remaining()];
+ data.get(array, 0, data.remaining());
+ return ByteBuffer.wrap(array).order(ByteOrder.LITTLE_ENDIAN);
+ }
+
+ private static class VariantMetadataReader extends PrimitiveReader<VariantMetadata> {
+ private VariantMetadataReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantMetadata read(VariantMetadata reuse) {
+ return Variants.metadata(readBinary(column));
+ }
+ }
+
+ private static class SerializedVariantReader extends PrimitiveReader<VariantValue>
+ implements VariantValueReader {
+ private SerializedVariantReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ return Variants.value(metadata, readBinary(column));
+ }
+ }
+
+ private static class ValueAsVariantReader<V> extends DelegatingValueReader<V, VariantValue>
+ implements VariantValueReader {
+ private final PhysicalType type;
+
+ private ValueAsVariantReader(PhysicalType type, ParquetValueReader<V> reader) {
+ super(reader);
+ this.type = type;
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata ignored) {
+ return Variants.of(type, readFromDelegate(null));
+ }
+ }
+
+ /**
+ * A Variant reader that combines value and typed_value columns from Parquet.
+ *
+ * <p>This reader does not handle merging partially shredded objects. To handle shredded objects,
+ * use {@link ShreddedObjectReader}.
+ */
+ private static class ShreddedVariantReader implements VariantValueReader {
+ private final int valueDL;
+ private final VariantValueReader valueReader;
+ private final int typedDL;
+ private final VariantValueReader typedReader;
+ private final TripleIterator<?> column;
+ private final List<TripleIterator<?>> children;
+
+ private ShreddedVariantReader(
+ int valueDL, VariantValueReader valueReader, int typedDL, VariantValueReader typedReader) {
+ this.valueDL = valueDL;
+ this.valueReader = valueReader;
+ this.typedDL = typedDL;
+ this.typedReader = typedReader;
+ this.column = valueReader != null ? valueReader.column() : typedReader.column();
+ this.children = children(valueReader, typedReader);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ VariantValue value = ParquetVariantReaders.read(metadata, valueReader, valueDL);
+ VariantValue typed = ParquetVariantReaders.read(metadata, typedReader, typedDL);
+
+ if (typed != null) {
Review Comment:
I don't recall where the spec landed on this, but shouldn't we fail if `typed` and `value` are both non-null?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943546072
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
Review Comment:
I don't think so given how this is used. This constant is intended to make code more clear in places where the value may be `null` in Parquet or may be a non-null `VariantValue` that represents null (`00`).
That way we don't have code that looks like `if (value == null) { ... }` because it is confusing that the value could be non-null but represent null. Instead, I wanted the code to be either `if (value == MISSING) { ... }` or `if (value.type() == PhysicalType.NULL) { ... }`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1943809738
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,409 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.iceberg.variants.Variants.PhysicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
+
+ @SuppressWarnings("unchecked")
+ public static ParquetValueReader<Variant> variant(
+ ParquetValueReader<?> metadata, ParquetValueReader<?> value) {
+ return new VariantReader(
+ (ParquetValueReader<VariantMetadata>) metadata, (VariantValueReader) value);
+ }
+
+ public static ParquetValueReader<VariantMetadata> metadata(ColumnDescriptor desc) {
+ return new VariantMetadataReader(desc);
+ }
+
+ public static VariantValueReader serialized(ColumnDescriptor desc) {
+ return new SerializedVariantReader(desc);
+ }
+
+ public static VariantValueReader shredded(
+ int valueDL,
Review Comment:
Updated.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1960242711
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestVariantReaders.java:
##########
@@ -0,0 +1,1107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.InternalReader;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Streams;
+import org.apache.iceberg.types.Types.IntegerType;
+import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.types.Types.VariantType;
+import org.apache.iceberg.variants.PhysicalType;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantPrimitive;
+import org.apache.iceberg.variants.VariantTestUtil;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.avro.AvroSchemaConverter;
+import org.apache.parquet.avro.AvroWriteSupport;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+import org.assertj.core.api.Assumptions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.FieldSource;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestVariantReaders {
+ private static final Schema SCHEMA =
+ new Schema(
+ NestedField.required(1, "id", IntegerType.get()),
+ NestedField.required(2, "var", VariantType.get()));
+
+ private static final LogicalTypeAnnotation STRING = LogicalTypeAnnotation.stringType();
+
+ private static final ByteBuffer TEST_METADATA_BUFFER =
+ VariantTestUtil.createMetadata(ImmutableList.of("a", "b", "c", "d", "e"), true);
+ private static final ByteBuffer TEST_OBJECT_BUFFER =
+ VariantTestUtil.createObject(
+ TEST_METADATA_BUFFER,
+ ImmutableMap.of(
+ "a", Variants.ofNull(),
+ "d", Variants.of(PhysicalType.STRING, "iceberg")));
+
+ private static final VariantMetadata EMPTY_METADATA =
+ Variants.metadata(VariantTestUtil.emptyMetadata());
+ private static final VariantMetadata TEST_METADATA = Variants.metadata(TEST_METADATA_BUFFER);
+ private static final VariantObject TEST_OBJECT =
+ (VariantObject) Variants.value(TEST_METADATA, TEST_OBJECT_BUFFER);
+
+ private static final VariantPrimitive<?>[] PRIMITIVES =
+ new VariantPrimitive[] {
+ Variants.ofNull(),
+ Variants.of(true),
+ Variants.of(false),
+ Variants.of((byte) 34),
+ Variants.of((byte) -34),
+ Variants.of((short) 1234),
+ Variants.of((short) -1234),
+ Variants.of(12345),
+ Variants.of(-12345),
+ Variants.of(9876543210L),
+ Variants.of(-9876543210L),
+ Variants.of(10.11F),
+ Variants.of(-10.11F),
+ Variants.of(14.3D),
+ Variants.of(-14.3D),
+ Variants.ofIsoDate("2024-11-07"),
+ Variants.ofIsoDate("1957-11-07"),
+ Variants.ofIsoTimestamptz("2024-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestamptz("1957-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestampntz("2024-11-07T12:33:54.123456"),
+ Variants.ofIsoTimestampntz("1957-11-07T12:33:54.123456"),
+ Variants.of(new BigDecimal("123456.7890")), // decimal4
+ Variants.of(new BigDecimal("-123456.7890")), // decimal4
+ Variants.of(new BigDecimal("1234567890.987654321")), // decimal8
+ Variants.of(new BigDecimal("-1234567890.987654321")), // decimal8
+ Variants.of(new BigDecimal("9876543210.123456789")), // decimal16
+ Variants.of(new BigDecimal("-9876543210.123456789")), // decimal16
+ Variants.of(ByteBuffer.wrap(new byte[] {0x0a, 0x0b, 0x0c, 0x0d})),
+ Variants.of("iceberg"),
+ };
+
+ private static Stream<Arguments> metadataAndValues() {
+ Stream<Arguments> primitives =
+ Stream.of(PRIMITIVES).map(variant -> Arguments.of(EMPTY_METADATA, variant));
+ Stream<Arguments> object = Stream.of(Arguments.of(TEST_METADATA, TEST_OBJECT));
+ return Streams.concat(primitives, object);
+ }
+
+ @ParameterizedTest
+ @MethodSource("metadataAndValues")
+ public void testUnshreddedVariants(VariantMetadata metadata, VariantValue expected)
+ throws IOException {
+ GroupType variantType = variant("var", 2);
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", serialize(metadata), "value", serialize(expected)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(metadata, actualVariant.metadata());
+ VariantTestUtil.assertEqual(expected, actualVariant.value());
+ }
+
+ @ParameterizedTest
+ @MethodSource("metadataAndValues")
+ public void testUnshreddedVariantsWithShreddedSchema(
+ VariantMetadata metadata, VariantValue expected) throws IOException {
+ // the variant's Parquet schema has a shredded field that is unused by all data values
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.BINARY, STRING));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", serialize(metadata), "value", serialize(expected)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(metadata, actualVariant.metadata());
+ VariantTestUtil.assertEqual(expected, actualVariant.value());
+ }
+
+ @ParameterizedTest
+ @FieldSource("PRIMITIVES")
+ public void testShreddedVariantPrimitives(VariantPrimitive<?> primitive) throws IOException {
+ Assumptions.assumeThat(primitive.type() != PhysicalType.NULL)
+ .as("Null is not a shredded type")
+ .isTrue();
+
+ GroupType variantType = variant("var", 2, shreddedType(primitive));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(
+ variantType,
+ Map.of(
+ "metadata",
+ VariantTestUtil.emptyMetadata(),
+ "typed_value",
+ toAvroValue(primitive)));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(primitive, actualVariant.value());
+ }
+
+ @Test
+ public void testNullValueAndNullTypedValue() throws IOException {
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.INT32));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata()));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(Variants.ofNull(), actualVariant.value());
+ }
+
+ @Test
+ public void testMissingValueColumn() throws IOException {
+ GroupType variantType =
+ Types.buildGroup(Type.Repetition.REQUIRED)
+ .id(2)
+ .required(PrimitiveTypeName.BINARY)
+ .named("metadata")
+ .addField(shreddedPrimitive(PrimitiveTypeName.INT32))
+ .named("var");
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata(), "typed_value", 34));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ Record actual = writeAndRead(parquetSchema, record);
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("var")).isInstanceOf(Variant.class);
+
+ Variant actualVariant = (Variant) actual.getField("var");
+ VariantTestUtil.assertEqual(EMPTY_METADATA, actualVariant.metadata());
+ VariantTestUtil.assertEqual(Variants.of(34), actualVariant.value());
+ }
+
+ @Test
+ public void testValueAndTypedValueConflict() throws IOException {
+ GroupType variantType = variant("var", 2, shreddedPrimitive(PrimitiveTypeName.INT32));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(
+ variantType,
+ Map.of(
+ "metadata",
+ VariantTestUtil.emptyMetadata(),
+ "value",
+ serialize(Variants.of("str")),
+ "typed_value",
+ 34));
+ GenericRecord record = record(parquetSchema, Map.of("id", 1, "var", variant));
+
+ assertThatThrownBy(() -> writeAndRead(parquetSchema, record))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid variant, conflicting value and typed_value");
+ }
+
+ @Test
+ public void testUnsignedInteger() {
+ GroupType variantType =
+ variant(
+ "var",
+ 2,
+ shreddedPrimitive(PrimitiveTypeName.INT32, LogicalTypeAnnotation.intType(32, false)));
+ MessageType parquetSchema = parquetSchema(variantType);
+
+ GenericRecord variant =
+ record(variantType, Map.of("metadata", VariantTestUtil.emptyMetadata()));
Review Comment:
We don't need to. This fails before attempting to read because the schema is invalid.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1960603787
##########
core/src/test/java/org/apache/iceberg/variants/VariantTestUtil.java:
##########
@@ -27,10 +29,55 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
public class VariantTestUtil {
private VariantTestUtil() {}
+ public static void assertEqual(VariantMetadata expected, VariantMetadata actual) {
+ assertThat(actual).isNotNull();
+ assertThat(expected).isNotNull();
+ assertThat(actual.dictionarySize())
+ .as("Dictionary size should match")
+ .isEqualTo(expected.dictionarySize());
+
+ for (int i = 0; i < expected.dictionarySize(); i += 1) {
+ assertThat(actual.get(i)).isEqualTo(expected.get(i));
+ }
+ }
+
+ public static void assertEqual(VariantValue expected, VariantValue actual) {
+ assertThat(actual).isNotNull();
+ assertThat(expected).isNotNull();
+ assertThat(actual.type()).as("Variant type should match").isEqualTo(expected.type());
+
+ if (expected.type() == PhysicalType.OBJECT) {
+ VariantObject expectedObject = expected.asObject();
+ VariantObject actualObject = actual.asObject();
+ assertThat(actualObject.numFields())
+ .as("Variant object num fields should match")
+ .isEqualTo(expectedObject.numFields());
+ for (String fieldName : expectedObject.fieldNames()) {
+ assertEqual(expectedObject.get(fieldName), actualObject.get(fieldName));
+ }
+
Review Comment:
This is fine since it helps readability.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Parquet: Implement Variant readers [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #12139:
URL: https://github.com/apache/iceberg/pull/12139#discussion_r1960633603
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantReaders.java:
##########
@@ -0,0 +1,424 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.parquet;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.stream.Streams;
+import org.apache.iceberg.parquet.ParquetValueReaders.PrimitiveReader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.variants.PhysicalType;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantObject;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+public class ParquetVariantReaders {
+ private ParquetVariantReaders() {}
+
+ public interface VariantValueReader extends ParquetValueReader<VariantValue> {
+ @Override
+ default VariantValue read(VariantValue reuse) {
+ throw new UnsupportedOperationException("Variants must be read using read(VariantMetadata)");
+ }
+
+ /** Reads a variant value */
+ VariantValue read(VariantMetadata metadata);
+ }
+
+ private static final VariantValue MISSING = null;
+
+ @SuppressWarnings("unchecked")
+ public static ParquetValueReader<Variant> variant(
+ ParquetValueReader<?> metadata, ParquetValueReader<?> value) {
+ return new VariantReader(
+ (ParquetValueReader<VariantMetadata>) metadata, (VariantValueReader) value);
+ }
+
+ public static ParquetValueReader<VariantMetadata> metadata(ColumnDescriptor desc) {
+ return new VariantMetadataReader(desc);
+ }
+
+ public static VariantValueReader serialized(ColumnDescriptor desc) {
+ return new SerializedVariantReader(desc);
+ }
+
+ public static VariantValueReader shredded(
+ int valueDefinitionLevel,
+ ParquetValueReader<?> valueReader,
+ int typedDefinitionLevel,
+ ParquetValueReader<?> typedReader) {
+ return new ShreddedVariantReader(
+ valueDefinitionLevel,
+ (VariantValueReader) valueReader,
+ typedDefinitionLevel,
+ (VariantValueReader) typedReader);
+ }
+
+ public static VariantValueReader objects(
+ int valueDefinitionLevel,
+ ParquetValueReader<?> valueReader,
+ int typedDefinitionLevel,
+ List<String> fieldNames,
+ List<VariantValueReader> fieldReaders) {
+ return new ShreddedObjectReader(
+ valueDefinitionLevel,
+ (VariantValueReader) valueReader,
+ typedDefinitionLevel,
+ fieldNames,
+ fieldReaders);
+ }
+
+ public static VariantValueReader asVariant(PhysicalType type, ParquetValueReader<?> reader) {
+ return new ValueAsVariantReader<>(type, reader);
+ }
+
+ private abstract static class DelegatingValueReader<S, T> implements ParquetValueReader<T> {
+ private final ParquetValueReader<S> reader;
+
+ private DelegatingValueReader(ParquetValueReader<S> reader) {
+ this.reader = reader;
+ }
+
+ protected S readFromDelegate(S reuse) {
+ return reader.read(reuse);
+ }
+
+ @Override
+ public TripleIterator<?> column() {
+ return reader.column();
+ }
+
+ @Override
+ public List<TripleIterator<?>> columns() {
+ return reader.columns();
+ }
+
+ @Override
+ public void setPageSource(PageReadStore pageStore) {
+ reader.setPageSource(pageStore);
+ }
+ }
+
+ private static ByteBuffer readBinary(ColumnIterator<?> column) {
+ ByteBuffer data = column.nextBinary().toByteBuffer();
+ byte[] array = new byte[data.remaining()];
+ data.get(array, 0, data.remaining());
+ return ByteBuffer.wrap(array).order(ByteOrder.LITTLE_ENDIAN);
+ }
+
+ private static class VariantMetadataReader extends PrimitiveReader<VariantMetadata> {
+ private VariantMetadataReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantMetadata read(VariantMetadata reuse) {
+ return Variants.metadata(readBinary(column));
+ }
+ }
+
+ private static class SerializedVariantReader extends PrimitiveReader<VariantValue>
+ implements VariantValueReader {
+ private SerializedVariantReader(ColumnDescriptor desc) {
+ super(desc);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ return Variants.value(metadata, readBinary(column));
+ }
+ }
+
+ private static class ValueAsVariantReader<V> extends DelegatingValueReader<V, VariantValue>
+ implements VariantValueReader {
+ private final PhysicalType type;
+
+ private ValueAsVariantReader(PhysicalType type, ParquetValueReader<V> reader) {
+ super(reader);
+ this.type = type;
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata ignored) {
+ return Variants.of(type, readFromDelegate(null));
+ }
+ }
+
+ /**
+ * A Variant reader that combines value and typed_value columns from Parquet.
+ *
+ * <p>This reader does not handle merging partially shredded objects. To handle shredded objects,
+ * use {@link ShreddedObjectReader}.
+ */
+ private static class ShreddedVariantReader implements VariantValueReader {
+ private final int valueDefinitionLevel;
+ private final VariantValueReader valueReader;
+ private final int typeDefinitionLevel;
+ private final VariantValueReader typedReader;
+ private final TripleIterator<?> column;
+ private final List<TripleIterator<?>> children;
+
+ private ShreddedVariantReader(
+ int valueDefinitionLevel,
+ VariantValueReader valueReader,
+ int typeDefinitionLevel,
+ VariantValueReader typedReader) {
+ this.valueDefinitionLevel = valueDefinitionLevel;
+ this.valueReader = valueReader;
+ this.typeDefinitionLevel = typeDefinitionLevel;
+ this.typedReader = typedReader;
+ this.column = valueReader != null ? valueReader.column() : typedReader.column();
+ this.children = children(valueReader, typedReader);
+ }
+
+ @Override
+ public VariantValue read(VariantMetadata metadata) {
+ VariantValue value = ParquetVariantReaders.read(metadata, valueReader, valueDefinitionLevel);
+ VariantValue typed = ParquetVariantReaders.read(metadata, typedReader, typeDefinitionLevel);
+
+ if (typed != null) {
+ Preconditions.checkArgument(
+ value == null,
+ "Invalid variant, conflicting value and typed_value: value=%s typed_value=%s",
+ value,
+ typed);
+ return typed;
+ }
+
+ return value;
+ }
+
+ @Override
+ public TripleIterator<?> column() {
+ return column;
+ }
+
+ @Override
+ public List<TripleIterator<?>> columns() {
+ return children;
+ }
+
+ @Override
+ public void setPageSource(PageReadStore pageStore) {
+ if (valueReader != null) {
+ valueReader.setPageSource(pageStore);
+ }
+
+ if (typedReader != null) {
+ typedReader.setPageSource(pageStore);
+ }
+ }
+ }
+
+ /**
+ * A Variant reader that combines value and partially shredded object columns.
+ *
+ * <p>This reader handles partially shredded objects. For shredded values, use {@link
+ * ShreddedVariantReader} instead.
+ */
+ private static class ShreddedObjectReader implements VariantValueReader {
+ private final int valueDefinitionLevel;
+ private final VariantValueReader valueReader;
+ private final int fieldsDefinitionLevel;
+ private final String[] fieldNames;
+ private final VariantValueReader[] fieldReaders;
+ private final TripleIterator<?> valueColumn;
+ private final TripleIterator<?> fieldColumn;
+ private final List<TripleIterator<?>> children;
+
+ private ShreddedObjectReader(
+ int valueDefinitionLevel,
+ VariantValueReader valueReader,
+ int fieldsDefinitionLevel,
+ List<String> fieldNames,
+ List<VariantValueReader> fieldReaders) {
+ this.valueDefinitionLevel = valueDefinitionLevel;
+ this.valueReader = valueReader;
+ this.fieldsDefinitionLevel = fieldsDefinitionLevel;
+ this.fieldNames = fieldNames.toArray(String[]::new);
+ this.fieldReaders = fieldReaders.toArray(VariantValueReader[]::new);
+ this.fieldColumn = this.fieldReaders[0].column();
+ this.valueColumn = valueReader != null ? valueReader.column() : fieldColumn;
+ this.children = children(Iterables.concat(Arrays.asList(valueReader), fieldReaders));
Review Comment:
nit: `Collections.singletonList` is preferable to Arrays.asList()
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org