You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/05/04 18:46:31 UTC
[GitHub] [incubator-iceberg] rdblue opened a new pull request #999: Spark: Pass correct types to get data from InternalRow
rdblue opened a new pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999
This fixes a problem with Spark 3.0 CTAS queries that use tinyint or smallint types. When Iceberg converts a Dataset schema, it promotes both smaller integers to `int`. Normally, Spark will insert casts in the analyzer so that the values are ints, but during a CTAS query, the table is created and the values may be passed as short or byte in the rows passed to Iceberg.
The problem happens when Iceberg accesses values from `InternalRow`. Before this commit, Iceberg would use the table's type to fetch a value, causing unsafe rows to return a corrupted byte or short value because 4 bytes had been read instead of 1 or 2.
The fix is to keep track of the Dataset schema and use it when accessing fields. This required building visitors for Avro and Parquet that traverse a Spark schema with a file schema.
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdsr commented on a change in pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdsr commented on a change in pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#discussion_r420966133
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/ParquetWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Type.Repetition;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * Visitor for traversing a Parquet type with a companion Spark type.
+ *
+ * @param <T> the Java class returned by the visitor
+ */
+public class ParquetWithSparkSchemaVisitor<T> {
+ private final Deque<String> fieldNames = Lists.newLinkedList();
Review comment:
Sounds good to me!
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdblue commented on a change in pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#discussion_r420952625
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/AvroWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.avro.Schema;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.avro.LogicalMap;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.StringType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+public abstract class AvroWithSparkSchemaVisitor<T> {
+ public static <T> T visit(StructType struct, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ return visitRecord(struct, schema, visitor);
+ }
+
+ public static <T> T visit(DataType type, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ switch (schema.getType()) {
+ case RECORD:
+ Preconditions.checkArgument(type instanceof StructType, "Invalid struct: %s is not a struct", type);
+ return visitRecord((StructType) type, schema, visitor);
+
+ case UNION:
Review comment:
This is validated in `SparkAvroWriter`, but I'll add a check in `visitUnion` as well.
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdblue commented on pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#issuecomment-624777938
Thanks, @rdsr! I've fixed the things you pointed out.
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdsr commented on a change in pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdsr commented on a change in pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#discussion_r420852247
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/AvroWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.avro.Schema;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.avro.LogicalMap;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.StringType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+public abstract class AvroWithSparkSchemaVisitor<T> {
+ public static <T> T visit(StructType struct, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ return visitRecord(struct, schema, visitor);
+ }
+
+ public static <T> T visit(DataType type, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ switch (schema.getType()) {
+ case RECORD:
+ Preconditions.checkArgument(type instanceof StructType, "Invalid struct: %s is not a struct", type);
+ return visitRecord((StructType) type, schema, visitor);
+
+ case UNION:
Review comment:
I presume this will only be nullable unions
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/AvroWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.avro.Schema;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.avro.LogicalMap;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.StringType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+public abstract class AvroWithSparkSchemaVisitor<T> {
+ public static <T> T visit(StructType struct, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ return visitRecord(struct, schema, visitor);
+ }
+
+ public static <T> T visit(DataType type, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ switch (schema.getType()) {
+ case RECORD:
+ Preconditions.checkArgument(type instanceof StructType, "Invalid struct: %s is not a struct", type);
+ return visitRecord((StructType) type, schema, visitor);
+
+ case UNION:
+ return visitUnion(type, schema, visitor);
+
+ case ARRAY:
+ return visitArray(type, schema, visitor);
+
+ case MAP:
+ Preconditions.checkArgument(type instanceof MapType, "Invalid map: %s is not a map", type);
+ MapType map = (MapType) type;
+ Preconditions.checkArgument(map.keyType() instanceof StringType,
+ "Invalid map: %s is not a string", map.keyType());
+ return visitor.map(map, schema, visit(map.valueType(), schema.getValueType(), visitor));
+
+ default:
+ return visitor.primitive(type, schema);
+ }
+ }
+
+ private static <T> T visitRecord(StructType struct, Schema record, AvroWithSparkSchemaVisitor<T> visitor) {
+ // check to make sure this hasn't been visited before
+ String name = record.getFullName();
+ Preconditions.checkState(!visitor.recordLevels.contains(name),
+ "Cannot process recursive Avro record %s", name);
+ StructField[] sFields = struct.fields();
+ List<Schema.Field> fields = record.getFields();
+ Preconditions.checkArgument(sFields.length == fields.size(),
+ "Structs do not match: %s != %s", struct, record);
+
+ visitor.recordLevels.push(name);
+
+ List<String> names = Lists.newArrayListWithExpectedSize(fields.size());
+ List<T> results = Lists.newArrayListWithExpectedSize(fields.size());
+ for (int i = 0; i < sFields.length; i += 1) {
+ StructField sField = sFields[i];
+ Schema.Field field = fields.get(i);
+ Preconditions.checkArgument(AvroSchemaUtil.makeCompatibleName(sField.name()).equals(field.name()),
+ "Structs do not match: field %s != %s", sField.name(), field.name());
+ results.add(visit(sField.dataType(), field.schema(), visitor));
+ }
+
+ visitor.recordLevels.pop();
+
+ return visitor.record(struct, record, names, results);
+ }
+
+ private static <T> T visitUnion(DataType type, Schema union, AvroWithSparkSchemaVisitor<T> visitor) {
+ List<Schema> types = union.getTypes();
+ List<T> options = Lists.newArrayListWithExpectedSize(types.size());
+ for (Schema branch : types) {
Review comment:
Since we are traversing this with Spark schema, which does not support unions. Should we make sure that we are only processing nullable unions?
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/ParquetWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Type.Repetition;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * Visitor for traversing a Parquet type with a companion Spark type.
+ *
+ * @param <T> the Java class returned by the visitor
+ */
+public class ParquetWithSparkSchemaVisitor<T> {
+ private final Deque<String> fieldNames = Lists.newLinkedList();
+
+ public static <T> T visit(DataType sType, Type type, ParquetWithSparkSchemaVisitor<T> visitor) {
+ Preconditions.checkArgument(sType != null, "Invalid DataType: null");
+ if (type instanceof MessageType) {
+ Preconditions.checkArgument(sType instanceof StructType, "Invalid struct: %s is not a struct", sType);
+ StructType struct = (StructType) sType;
+ return visitor.message(struct, (MessageType) type, visitFields(struct, type.asGroupType(), visitor));
+
+ } else if (type.isPrimitive()) {
+ return visitor.primitive(sType, type.asPrimitiveType());
+
+ } else {
+ // if not a primitive, the typeId must be a group
+ GroupType group = type.asGroupType();
+ OriginalType annotation = group.getOriginalType();
+ if (annotation != null) {
+ switch (annotation) {
+ case LIST:
+ Preconditions.checkArgument(!group.isRepetition(Repetition.REPEATED),
+ "Invalid list: top-level group is repeated: %s", group);
+ Preconditions.checkArgument(group.getFieldCount() == 1,
+ "Invalid list: does not contain single repeated field: %s", group);
+
+ GroupType repeatedElement = group.getFields().get(0).asGroupType();
+ Preconditions.checkArgument(repeatedElement.isRepetition(Repetition.REPEATED),
+ "Invalid list: inner group is not repeated");
+ Preconditions.checkArgument(repeatedElement.getFieldCount() <= 1,
+ "Invalid list: repeated group is not a single field: %s", group);
+
+ Preconditions.checkArgument(sType instanceof ArrayType, "Invalid list: %s is not an array", sType);
+ ArrayType array = (ArrayType) sType;
+ StructField element = new StructField(
+ "element", array.elementType(), array.containsNull(), Metadata.empty());
+
+ visitor.fieldNames.push(repeatedElement.getName());
+ try {
+ T elementResult = null;
+ if (repeatedElement.getFieldCount() > 0) {
+ elementResult = visitField(element, repeatedElement.getType(0), visitor);
+ }
+
+ return visitor.list(array, group, elementResult);
+
+ } finally {
+ visitor.fieldNames.pop();
+ }
+
+ case MAP:
+ Preconditions.checkArgument(!group.isRepetition(Repetition.REPEATED),
+ "Invalid map: top-level group is repeated: %s", group);
+ Preconditions.checkArgument(group.getFieldCount() == 1,
+ "Invalid map: does not contain single repeated field: %s", group);
+
+ GroupType repeatedKeyValue = group.getType(0).asGroupType();
+ Preconditions.checkArgument(repeatedKeyValue.isRepetition(Repetition.REPEATED),
+ "Invalid map: inner group is not repeated");
+ Preconditions.checkArgument(repeatedKeyValue.getFieldCount() <= 2,
+ "Invalid map: repeated group does not have 2 fields");
+
+ Preconditions.checkArgument(sType instanceof MapType, "Invalid map: %s is not a map", sType);
+ MapType map = (MapType) sType;
+ StructField keyField = new StructField("key", map.keyType(), false, Metadata.empty());
+ StructField valueField = new StructField(
+ "value", map.valueType(), map.valueContainsNull(), Metadata.empty());
+
+ visitor.fieldNames.push(repeatedKeyValue.getName());
+ try {
+ T keyResult = null;
+ T valueResult = null;
+ switch (repeatedKeyValue.getFieldCount()) {
+ case 2:
+ // if there are 2 fields, both key and value are projected
+ keyResult = visitField(keyField, repeatedKeyValue.getType(0), visitor);
+ valueResult = visitField(valueField, repeatedKeyValue.getType(1), visitor);
+ break;
+ case 1:
+ // if there is just one, use the name to determine what it is
+ Type keyOrValue = repeatedKeyValue.getType(0);
+ if (keyOrValue.getName().equalsIgnoreCase("key")) {
+ keyResult = visitField(keyField, keyOrValue, visitor);
+ // value result remains null
+ } else {
+ valueResult = visitField(valueField, keyOrValue, visitor);
+ // key result remains null
+ }
+ break;
+ default:
+ // both results will remain null
+ }
+
+ return visitor.map(map, group, keyResult, valueResult);
+
+ } finally {
+ visitor.fieldNames.pop();
+ }
+
+ default:
+ }
+ }
+
+ Preconditions.checkArgument(sType instanceof StructType, "Invalid struct: %s is not a struct", sType);
+ StructType struct = (StructType) sType;
+ return visitor.struct(struct, group, visitFields(struct, group, visitor));
+ }
+ }
+
+ private static <T> T visitField(StructField sField, Type field, ParquetWithSparkSchemaVisitor<T> visitor) {
+ visitor.fieldNames.push(field.getName());
+ try {
+ return visit(sField.dataType(), field, visitor);
+ } finally {
+ visitor.fieldNames.pop();
+ }
+ }
+
+ private static <T> List<T> visitFields(StructType struct, GroupType group,
+ ParquetWithSparkSchemaVisitor<T> visitor) {
+ StructField[] sFields = struct.fields();
+ Preconditions.checkArgument(sFields.length == group.getFieldCount(),
+ "Structs do not match: %s and %s", struct, group);
+ List<T> results = Lists.newArrayListWithExpectedSize(group.getFieldCount());
+ for (int i = 0; i < sFields.length; i += 1) {
+ Type field = group.getFields().get(i);
+ StructField sField = sFields[i];
+ Preconditions.checkArgument(field.getName().equals(AvroSchemaUtil.makeCompatibleName(sField.name())),
+ "Structs do not match: field %s != %s", field.getName(), sField.name());
+ results.add(visitField(sField, field, visitor));
+ }
+
+ return results;
+ }
+
+ public T message(StructType sStruct, MessageType message, List<T> fields) {
+ return null;
+ }
+
+ public T struct(StructType sStruct, GroupType struct, List<T> fields) {
+ return null;
+ }
+
+ public T list(ArrayType sArray, GroupType array, T element) {
+ return null;
+ }
+
+ public T map(MapType sMap, GroupType map, T key, T value) {
+ return null;
+ }
+
+ public T primitive(DataType sPrimitive,
+ PrimitiveType primitive) {
Review comment:
nit: this can be on same line
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/ParquetWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Type.Repetition;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * Visitor for traversing a Parquet type with a companion Spark type.
+ *
+ * @param <T> the Java class returned by the visitor
+ */
+public class ParquetWithSparkSchemaVisitor<T> {
+ private final Deque<String> fieldNames = Lists.newLinkedList();
Review comment:
should we expose this functionality through the `beforeField` and `afterField` apis?
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdblue commented on a change in pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#discussion_r420955722
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/ParquetWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Type.Repetition;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * Visitor for traversing a Parquet type with a companion Spark type.
+ *
+ * @param <T> the Java class returned by the visitor
+ */
+public class ParquetWithSparkSchemaVisitor<T> {
+ private final Deque<String> fieldNames = Lists.newLinkedList();
Review comment:
Good question. I originally thought about doing this when refactoring Parquet visitors in #1001, but nearly ever Parquet visitor uses the `currentPath` or `path` functions, so it made more sense after #950 to keep the implementation in the base visitor.
Here, we could add similar `beforeField` and `afterField` callbacks, but that introduces 10 more methods that wouldn't actually be used. I'm inclined to add them when and if we need them. Does that sound reasonable?
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdblue commented on a change in pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#discussion_r420956121
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/AvroWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.avro.Schema;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.avro.LogicalMap;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.StringType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+public abstract class AvroWithSparkSchemaVisitor<T> {
+ public static <T> T visit(StructType struct, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ return visitRecord(struct, schema, visitor);
+ }
+
+ public static <T> T visit(DataType type, Schema schema, AvroWithSparkSchemaVisitor<T> visitor) {
+ switch (schema.getType()) {
+ case RECORD:
+ Preconditions.checkArgument(type instanceof StructType, "Invalid struct: %s is not a struct", type);
+ return visitRecord((StructType) type, schema, visitor);
+
+ case UNION:
+ return visitUnion(type, schema, visitor);
+
+ case ARRAY:
+ return visitArray(type, schema, visitor);
+
+ case MAP:
+ Preconditions.checkArgument(type instanceof MapType, "Invalid map: %s is not a map", type);
+ MapType map = (MapType) type;
+ Preconditions.checkArgument(map.keyType() instanceof StringType,
+ "Invalid map: %s is not a string", map.keyType());
+ return visitor.map(map, schema, visit(map.valueType(), schema.getValueType(), visitor));
+
+ default:
+ return visitor.primitive(type, schema);
+ }
+ }
+
+ private static <T> T visitRecord(StructType struct, Schema record, AvroWithSparkSchemaVisitor<T> visitor) {
+ // check to make sure this hasn't been visited before
+ String name = record.getFullName();
+ Preconditions.checkState(!visitor.recordLevels.contains(name),
+ "Cannot process recursive Avro record %s", name);
+ StructField[] sFields = struct.fields();
+ List<Schema.Field> fields = record.getFields();
+ Preconditions.checkArgument(sFields.length == fields.size(),
+ "Structs do not match: %s != %s", struct, record);
+
+ visitor.recordLevels.push(name);
+
+ List<String> names = Lists.newArrayListWithExpectedSize(fields.size());
+ List<T> results = Lists.newArrayListWithExpectedSize(fields.size());
+ for (int i = 0; i < sFields.length; i += 1) {
+ StructField sField = sFields[i];
+ Schema.Field field = fields.get(i);
+ Preconditions.checkArgument(AvroSchemaUtil.makeCompatibleName(sField.name()).equals(field.name()),
+ "Structs do not match: field %s != %s", sField.name(), field.name());
+ results.add(visit(sField.dataType(), field.schema(), visitor));
+ }
+
+ visitor.recordLevels.pop();
+
+ return visitor.record(struct, record, names, results);
+ }
+
+ private static <T> T visitUnion(DataType type, Schema union, AvroWithSparkSchemaVisitor<T> visitor) {
+ List<Schema> types = union.getTypes();
+ List<T> options = Lists.newArrayListWithExpectedSize(types.size());
+ for (Schema branch : types) {
Review comment:
Added a validation.
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdblue commented on a change in pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#discussion_r420956839
##########
File path: spark/src/main/java/org/apache/iceberg/spark/data/ParquetWithSparkSchemaVisitor.java
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.spark.data;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.util.Deque;
+import java.util.List;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Type.Repetition;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * Visitor for traversing a Parquet type with a companion Spark type.
+ *
+ * @param <T> the Java class returned by the visitor
+ */
+public class ParquetWithSparkSchemaVisitor<T> {
+ private final Deque<String> fieldNames = Lists.newLinkedList();
+
+ public static <T> T visit(DataType sType, Type type, ParquetWithSparkSchemaVisitor<T> visitor) {
+ Preconditions.checkArgument(sType != null, "Invalid DataType: null");
+ if (type instanceof MessageType) {
+ Preconditions.checkArgument(sType instanceof StructType, "Invalid struct: %s is not a struct", sType);
+ StructType struct = (StructType) sType;
+ return visitor.message(struct, (MessageType) type, visitFields(struct, type.asGroupType(), visitor));
+
+ } else if (type.isPrimitive()) {
+ return visitor.primitive(sType, type.asPrimitiveType());
+
+ } else {
+ // if not a primitive, the typeId must be a group
+ GroupType group = type.asGroupType();
+ OriginalType annotation = group.getOriginalType();
+ if (annotation != null) {
+ switch (annotation) {
+ case LIST:
+ Preconditions.checkArgument(!group.isRepetition(Repetition.REPEATED),
+ "Invalid list: top-level group is repeated: %s", group);
+ Preconditions.checkArgument(group.getFieldCount() == 1,
+ "Invalid list: does not contain single repeated field: %s", group);
+
+ GroupType repeatedElement = group.getFields().get(0).asGroupType();
+ Preconditions.checkArgument(repeatedElement.isRepetition(Repetition.REPEATED),
+ "Invalid list: inner group is not repeated");
+ Preconditions.checkArgument(repeatedElement.getFieldCount() <= 1,
+ "Invalid list: repeated group is not a single field: %s", group);
+
+ Preconditions.checkArgument(sType instanceof ArrayType, "Invalid list: %s is not an array", sType);
+ ArrayType array = (ArrayType) sType;
+ StructField element = new StructField(
+ "element", array.elementType(), array.containsNull(), Metadata.empty());
+
+ visitor.fieldNames.push(repeatedElement.getName());
+ try {
+ T elementResult = null;
+ if (repeatedElement.getFieldCount() > 0) {
+ elementResult = visitField(element, repeatedElement.getType(0), visitor);
+ }
+
+ return visitor.list(array, group, elementResult);
+
+ } finally {
+ visitor.fieldNames.pop();
+ }
+
+ case MAP:
+ Preconditions.checkArgument(!group.isRepetition(Repetition.REPEATED),
+ "Invalid map: top-level group is repeated: %s", group);
+ Preconditions.checkArgument(group.getFieldCount() == 1,
+ "Invalid map: does not contain single repeated field: %s", group);
+
+ GroupType repeatedKeyValue = group.getType(0).asGroupType();
+ Preconditions.checkArgument(repeatedKeyValue.isRepetition(Repetition.REPEATED),
+ "Invalid map: inner group is not repeated");
+ Preconditions.checkArgument(repeatedKeyValue.getFieldCount() <= 2,
+ "Invalid map: repeated group does not have 2 fields");
+
+ Preconditions.checkArgument(sType instanceof MapType, "Invalid map: %s is not a map", sType);
+ MapType map = (MapType) sType;
+ StructField keyField = new StructField("key", map.keyType(), false, Metadata.empty());
+ StructField valueField = new StructField(
+ "value", map.valueType(), map.valueContainsNull(), Metadata.empty());
+
+ visitor.fieldNames.push(repeatedKeyValue.getName());
+ try {
+ T keyResult = null;
+ T valueResult = null;
+ switch (repeatedKeyValue.getFieldCount()) {
+ case 2:
+ // if there are 2 fields, both key and value are projected
+ keyResult = visitField(keyField, repeatedKeyValue.getType(0), visitor);
+ valueResult = visitField(valueField, repeatedKeyValue.getType(1), visitor);
+ break;
+ case 1:
+ // if there is just one, use the name to determine what it is
+ Type keyOrValue = repeatedKeyValue.getType(0);
+ if (keyOrValue.getName().equalsIgnoreCase("key")) {
+ keyResult = visitField(keyField, keyOrValue, visitor);
+ // value result remains null
+ } else {
+ valueResult = visitField(valueField, keyOrValue, visitor);
+ // key result remains null
+ }
+ break;
+ default:
+ // both results will remain null
+ }
+
+ return visitor.map(map, group, keyResult, valueResult);
+
+ } finally {
+ visitor.fieldNames.pop();
+ }
+
+ default:
+ }
+ }
+
+ Preconditions.checkArgument(sType instanceof StructType, "Invalid struct: %s is not a struct", sType);
+ StructType struct = (StructType) sType;
+ return visitor.struct(struct, group, visitFields(struct, group, visitor));
+ }
+ }
+
+ private static <T> T visitField(StructField sField, Type field, ParquetWithSparkSchemaVisitor<T> visitor) {
+ visitor.fieldNames.push(field.getName());
+ try {
+ return visit(sField.dataType(), field, visitor);
+ } finally {
+ visitor.fieldNames.pop();
+ }
+ }
+
+ private static <T> List<T> visitFields(StructType struct, GroupType group,
+ ParquetWithSparkSchemaVisitor<T> visitor) {
+ StructField[] sFields = struct.fields();
+ Preconditions.checkArgument(sFields.length == group.getFieldCount(),
+ "Structs do not match: %s and %s", struct, group);
+ List<T> results = Lists.newArrayListWithExpectedSize(group.getFieldCount());
+ for (int i = 0; i < sFields.length; i += 1) {
+ Type field = group.getFields().get(i);
+ StructField sField = sFields[i];
+ Preconditions.checkArgument(field.getName().equals(AvroSchemaUtil.makeCompatibleName(sField.name())),
+ "Structs do not match: field %s != %s", field.getName(), sField.name());
+ results.add(visitField(sField, field, visitor));
+ }
+
+ return results;
+ }
+
+ public T message(StructType sStruct, MessageType message, List<T> fields) {
+ return null;
+ }
+
+ public T struct(StructType sStruct, GroupType struct, List<T> fields) {
+ return null;
+ }
+
+ public T list(ArrayType sArray, GroupType array, T element) {
+ return null;
+ }
+
+ public T map(MapType sMap, GroupType map, T key, T value) {
+ return null;
+ }
+
+ public T primitive(DataType sPrimitive,
+ PrimitiveType primitive) {
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.
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
[GitHub] [incubator-iceberg] rdsr commented on pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdsr commented on pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#issuecomment-624720675
LGTM. Minor comments
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdblue commented on pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#issuecomment-624143229
Thanks, @rdsr! Good for you to review since we will need to do the same thing for the ORC writers.
----------------------------------------------------------------
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.
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
[GitHub] [incubator-iceberg] rdsr commented on pull request #999: Spark: Pass correct types to get data from InternalRow
Posted by GitBox <gi...@apache.org>.
rdsr commented on pull request #999:
URL: https://github.com/apache/incubator-iceberg/pull/999#issuecomment-623874689
Thanks @rdblue . I'll look into it tomorrow!
----------------------------------------------------------------
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.
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