From fb23f28875871f82148a25838db723ddd67f9d05 Mon Sep 17 00:00:00 2001 From: Damian Guy Date: Sun, 9 Aug 2015 22:10:51 +0100 Subject: [PATCH] [SparkSPARK-9340] - make SparkSQL work with nested types in parquet-protobuf --- .../sql/parquet/CatalystReadSupport.scala | 7 +- .../sql/parquet/CatalystRowConverter.scala | 114 +++- .../sql/parquet/CatalystSchemaConverter.scala | 2 +- .../spark/sql/parquet/ParquetRelation.scala | 20 +- .../ProtobufCatalystSchemaConverter.scala | 560 ++++++++++++++++++ .../spark/sql/parquet/SchemaConverter.scala | 26 + .../resources/nested-array-struct.parquet | Bin 0 -> 775 bytes .../test/resources/old-repeated-int.parquet | Bin 0 -> 389 bytes .../resources/old-repeated-message.parquet | Bin 0 -> 600 bytes .../src/test/resources/old-repeated.parquet | Bin 0 -> 432 bytes .../resources/proto-repeated-string.parquet | Bin 0 -> 411 bytes .../resources/proto-repeated-struct.parquet | Bin 0 -> 608 bytes .../proto-struct-with-array-many.parquet | Bin 0 -> 802 bytes .../resources/proto-struct-with-array.parquet | Bin 0 -> 1576 bytes .../spark/sql/parquet/ParquetQuerySuite.scala | 5 +- .../sql/parquet/ParquetSchemaSuite.scala | 65 +- .../ProtoParquetTypesConverterTest.scala | 91 +++ 17 files changed, 874 insertions(+), 16 deletions(-) create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/parquet/ProtobufCatalystSchemaConverter.scala create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/parquet/SchemaConverter.scala create mode 100755 sql/core/src/test/resources/nested-array-struct.parquet create mode 100755 sql/core/src/test/resources/old-repeated-int.parquet create mode 100755 sql/core/src/test/resources/old-repeated-message.parquet create mode 100644 sql/core/src/test/resources/old-repeated.parquet create mode 100755 sql/core/src/test/resources/proto-repeated-string.parquet create mode 100755 sql/core/src/test/resources/proto-repeated-struct.parquet create mode 100755 sql/core/src/test/resources/proto-struct-with-array-many.parquet create mode 100755 sql/core/src/test/resources/proto-struct-with-array.parquet create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/parquet/ProtoParquetTypesConverterTest.scala diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystReadSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystReadSupport.scala index 975fec101d9c2..d6c7e7707ea9a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystReadSupport.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystReadSupport.scala @@ -39,7 +39,12 @@ private[parquet] class CatalystReadSupport extends ReadSupport[InternalRow] with readContext: ReadContext): RecordMaterializer[InternalRow] = { log.debug(s"Preparing for read Parquet file with message type: $fileSchema") - val toCatalyst = new CatalystSchemaConverter(conf) + val toCatalyst = if (keyValueMetaData.containsKey("parquet.proto.class")) { + new ProtobufCatalystSchemaConverter(conf) + } else { + new CatalystSchemaConverter(conf) + } + val parquetRequestedSchema = readContext.getRequestedSchema val catalystRequestedSchema = diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystRowConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystRowConverter.scala index 6938b071065cd..0648bfa9ebf0c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystRowConverter.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystRowConverter.scala @@ -20,12 +20,12 @@ package org.apache.spark.sql.parquet import java.math.{BigDecimal, BigInteger} import java.nio.ByteOrder + import scala.collection.JavaConversions._ -import scala.collection.mutable import scala.collection.mutable.ArrayBuffer import org.apache.parquet.column.Dictionary -import org.apache.parquet.io.api.{Binary, Converter, GroupConverter, PrimitiveConverter} +import org.apache.parquet.io.api._ import org.apache.parquet.schema.Type.Repetition import org.apache.parquet.schema.{GroupType, PrimitiveType, Type} @@ -103,16 +103,22 @@ private[parquet] class CatalystRowConverter( }.toArray } + private val needsArrayReset = + fieldConverters.filter(converter => converter.isInstanceOf[NeedsResetArray]) override def getConverter(fieldIndex: Int): Converter = fieldConverters(fieldIndex) override def end(): Unit = updater.set(currentRow) + override def start(): Unit = { var i = 0 while (i < currentRow.numFields) { currentRow.setNullAt(i) i += 1 } + needsArrayReset.foreach { + converter => converter.asInstanceOf[NeedsResetArray].resetArray() + } } /** @@ -171,8 +177,20 @@ private[parquet] class CatalystRowConverter( } } - case t: ArrayType => - new CatalystArrayConverter(parquetType.asGroupType(), t, updater) + case t: ArrayType => { + parquetType.isRepetition(Type.Repetition.REPEATED) match { + case true => { + t match { + case ArrayType(elementType: StructType, _) => + new CatalystRepeatedStructConverter(parquetType.asGroupType(), elementType, updater) + case ArrayType(elementType: DataType, _) => + new CatalystRepeatedPrimitiveConverter(parquetType, elementType, updater) + } + } + case false => + new CatalystArrayConverter(parquetType.asGroupType(), t, updater) + } + } case t: MapType => new CatalystMapConverter(parquetType.asGroupType(), t, updater) @@ -378,6 +396,66 @@ private[parquet] class CatalystRowConverter( } } + /** + * Support Protobuf native repeated. parquet-protobuf does a 1 - 1 conversion, i.e., + * repeated int32 myInt; + * @param parquetType + * @param catalystType + * @param updater + */ + private final class CatalystRepeatedPrimitiveConverter( + parquetType: Type, + myCatalystType: DataType, + updater: ParentContainerUpdater) + extends PrimitiveConverter with NeedsResetArray { + + private var elements: Int = 0 + private var buffer: ArrayBuffer[Any] = ArrayBuffer.empty[Any] + + private val stringConverter = new CatalystStringConverter(new ParentContainerUpdater { + override def set(value: Any): Unit = addValue(value) + }) + + override def hasDictionarySupport: Boolean = true + + override def addValueFromDictionary(dictionaryId: Int): Unit = + stringConverter.addValueFromDictionary(dictionaryId) + + override def setDictionary(dictionary: Dictionary): Unit = + stringConverter.setDictionary(dictionary) + + private def addValue(value: Any): Unit = { + buffer+= value + elements +=1 + updater.set(new GenericArrayData(buffer.slice(0, elements).toArray)) + } + + override def addBinary(value: Binary): Unit = + myCatalystType match { + case StringType => stringConverter.addBinary(value) + case _ => addValue(value) + } + + + override def addDouble(value: Double): Unit = addValue(value) + + override def addInt(value: Int): Unit = { + addValue(value) + } + + override def addBoolean(value: Boolean): Unit = addValue(value) + + override def addFloat(value: Float): Unit = addValue(value) + + override def addLong(value: Long): Unit = addValue(value) + + override def resetArray(): Unit = { + buffer.clear() + elements = 0 + } + } + + /** Parquet converter for maps */ private final class CatalystMapConverter( parquetType: GroupType, @@ -446,4 +524,32 @@ private[parquet] class CatalystRowConverter( } } } + + trait NeedsResetArray { + def resetArray(): Unit + } + + private class CatalystRepeatedStructConverter( + groupType: GroupType, + structType: StructType, + updater: ParentContainerUpdater) + extends CatalystRowConverter(groupType, structType, updater) + with NeedsResetArray { + + private var rowBuffer: ArrayBuffer[Any] = ArrayBuffer.empty + private var elements = 0 + + override def end(): Unit = { + rowBuffer += currentRow.copy() + elements += 1 + updater.set(new GenericArrayData(rowBuffer.slice(0, elements).toArray)) + } + + def resetArray(): Unit = { + rowBuffer.clear() + elements = 0 + } + + } + } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystSchemaConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystSchemaConverter.scala index d43ca95b4eea0..8bc3ca6ac2f46 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystSchemaConverter.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystSchemaConverter.scala @@ -56,7 +56,7 @@ import org.apache.spark.sql.{AnalysisException, SQLConf} private[parquet] class CatalystSchemaConverter( private val assumeBinaryIsString: Boolean, private val assumeInt96IsTimestamp: Boolean, - private val followParquetFormatSpec: Boolean) { + private val followParquetFormatSpec: Boolean) extends SchemaConverter{ // Only used when constructing converter for converting Spark SQL schema to Parquet schema, in // which case `assumeInt96IsTimestamp` and `assumeBinaryIsString` are irrelevant. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala index 29c388c22ef93..115b993f0bd29 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala @@ -705,8 +705,22 @@ private[sql] object ParquetRelation extends Logging { assumeInt96IsTimestamp = assumeInt96IsTimestamp, followParquetFormatSpec = followParquetFormatSpec) - footers.map { footer => - ParquetRelation.readSchemaFromFooter(footer, converter) + val protobufConverter = + new ProtobufCatalystSchemaConverter( + assumeBinaryIsString = assumeBinaryIsString, + assumeInt96IsTimestamp = assumeInt96IsTimestamp, + followParquetFormatSpec = followParquetFormatSpec) + + footers.map { footer => { + val myConverter = + if (footer.getParquetMetadata.getFileMetaData. + getKeyValueMetaData.containsKey("parquet.proto.class")) { + protobufConverter + } else { + converter + } + ParquetRelation.readSchemaFromFooter(footer, myConverter) + } }.reduceOption(_ merge _).iterator }.collect() @@ -719,7 +733,7 @@ private[sql] object ParquetRelation extends Logging { * a [[StructType]] converted from the [[MessageType]] stored in this footer. */ def readSchemaFromFooter( - footer: Footer, converter: CatalystSchemaConverter): StructType = { + footer: Footer, converter: SchemaConverter): StructType = { val fileMetaData = footer.getParquetMetadata.getFileMetaData fileMetaData .getKeyValueMetaData diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/ProtobufCatalystSchemaConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/ProtobufCatalystSchemaConverter.scala new file mode 100644 index 0000000000000..c6384012167c6 --- /dev/null +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/ProtobufCatalystSchemaConverter.scala @@ -0,0 +1,560 @@ +/* + * 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.spark.sql.parquet + +import scala.collection.JavaConversions._ + +import org.apache.hadoop.conf.Configuration +import org.apache.parquet.schema.OriginalType._ +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName._ +import org.apache.parquet.schema.Type.Repetition._ +import org.apache.parquet.schema._ + +import org.apache.spark.sql.types._ +import org.apache.spark.sql.{AnalysisException, SQLConf} + +/** + * This converter class is used to convert Parquet [[MessageType]] to Spark SQL [[StructType]] and + * vice versa. + * + * Parquet format backwards-compatibility rules are respected when converting Parquet + * [[MessageType]] schemas. + * + * @see https://github.com/apache/parquet-format/blob/master/LogicalTypes.md + * + * @constructor + * @param assumeBinaryIsString Whether unannotated BINARY fields should be assumed to be Spark SQL + * [[StringType]] fields when converting Parquet a [[MessageType]] to Spark SQL + * [[StructType]]. + * @param assumeInt96IsTimestamp Whether unannotated INT96 fields should be assumed to be Spark SQL + * [[TimestampType]] fields when converting Parquet a [[MessageType]] to Spark SQL + * [[StructType]]. Note that Spark SQL [[TimestampType]] is similar to Hive timestamp, which + * has optional nanosecond precision, but different from `TIME_MILLS` and `TIMESTAMP_MILLIS` + * described in Parquet format spec. + * @param followParquetFormatSpec Whether to generate standard DECIMAL, LIST, and MAP structure when + * converting Spark SQL [[StructType]] to Parquet [[MessageType]]. For Spark 1.4.x and + * prior versions, Spark SQL only supports decimals with a max precision of 18 digits, and + * uses non-standard LIST and MAP structure. Note that the current Parquet format spec is + * backwards-compatible with these settings. If this argument is set to `false`, we fallback + * to old style non-standard behaviors. + */ +private[parquet] class ProtobufCatalystSchemaConverter( + private val assumeBinaryIsString: Boolean, + private val assumeInt96IsTimestamp: Boolean, + private val followParquetFormatSpec: Boolean) extends SchemaConverter { + + + + // Only used when constructing converter for converting Spark SQL schema to Parquet schema, in + // which case `assumeInt96IsTimestamp` and `assumeBinaryIsString` are irrelevant. + def this() = this( + assumeBinaryIsString = SQLConf.PARQUET_BINARY_AS_STRING.defaultValue.get, + assumeInt96IsTimestamp = SQLConf.PARQUET_INT96_AS_TIMESTAMP.defaultValue.get, + followParquetFormatSpec = SQLConf.PARQUET_FOLLOW_PARQUET_FORMAT_SPEC.defaultValue.get) + + def this(conf: SQLConf) = this( + assumeBinaryIsString = conf.isParquetBinaryAsString, + assumeInt96IsTimestamp = conf.isParquetINT96AsTimestamp, + followParquetFormatSpec = conf.followParquetFormatSpec) + + def this(conf: Configuration) = this( + assumeBinaryIsString = + conf.getBoolean( + SQLConf.PARQUET_BINARY_AS_STRING.key, + SQLConf.PARQUET_BINARY_AS_STRING.defaultValue.get), + assumeInt96IsTimestamp = + conf.getBoolean( + SQLConf.PARQUET_INT96_AS_TIMESTAMP.key, + SQLConf.PARQUET_INT96_AS_TIMESTAMP.defaultValue.get), + followParquetFormatSpec = + conf.getBoolean( + SQLConf.PARQUET_FOLLOW_PARQUET_FORMAT_SPEC.key, + SQLConf.PARQUET_FOLLOW_PARQUET_FORMAT_SPEC.defaultValue.get)) + + /** + * Converts Parquet [[MessageType]] `parquetSchema` to a Spark SQL [[StructType]]. + */ + def convert(parquetSchema: MessageType): StructType = convert(parquetSchema.asGroupType()) + + private def convert(parquetSchema: GroupType): StructType = { + val fields = parquetSchema.getFields.map { field => + field.getRepetition match { + case OPTIONAL => + StructField(field.getName, convertField(field), nullable = true) + + case REQUIRED => + StructField(field.getName, convertField(field), nullable = false) + + case REPEATED => + StructField(field.getName, convertField(field), nullable = false) + } + } + + StructType(fields) + } + + /** + * Converts a Parquet [[Type]] to a Spark SQL [[DataType]]. + */ + def convertField(parquetType: Type): DataType = parquetType match { + case t: PrimitiveType => convertPrimitiveField(t) + case t: GroupType => convertGroupField(t.asGroupType()) + } + + private def convertPrimitiveField(field: PrimitiveType): DataType = { + val typeName = field.getPrimitiveTypeName + val originalType = field.getOriginalType + val repetition = field.getRepetition + + def typeString = + if (originalType == null) s"$typeName" else s"$typeName ($originalType)" + + def typeNotImplemented() = + throw new AnalysisException(s"Parquet type not yet supported: $typeString") + + def illegalType() = + throw new AnalysisException(s"Illegal Parquet type: $typeString") + + // When maxPrecision = -1, we skip precision range check, and always respect the precision + // specified in field.getDecimalMetadata. This is useful when interpreting decimal types stored + // as binaries with variable lengths. + def makeDecimalType(maxPrecision: Int = -1): DecimalType = { + val precision = field.getDecimalMetadata.getPrecision + val scale = field.getDecimalMetadata.getScale + + CatalystSchemaConverter.analysisRequire( + maxPrecision == -1 || 1 <= precision && precision <= maxPrecision, + s"Invalid decimal precision: $typeName cannot store $precision digits (max $maxPrecision)") + + DecimalType(precision, scale) + } + + def toPrimitiveType: AtomicType with Product with Serializable = { + typeName match { + case BOOLEAN => BooleanType + + case FLOAT => FloatType + + case DOUBLE => DoubleType + + case INT32 => + originalType match { + case INT_8 => ByteType + case INT_16 => ShortType + case INT_32 | null => IntegerType + case DATE => DateType + case DECIMAL => makeDecimalType(maxPrecisionForBytes(4)) + case TIME_MILLIS => typeNotImplemented() + case _ => illegalType() + } + + case INT64 => + originalType match { + case INT_64 | null => LongType + case DECIMAL => makeDecimalType(maxPrecisionForBytes(8)) + case TIMESTAMP_MILLIS => typeNotImplemented() + case _ => illegalType() + } + + case INT96 => + CatalystSchemaConverter.analysisRequire( + assumeInt96IsTimestamp, + "INT96 is not supported unless it's interpreted as timestamp. " + + s"Please try to set ${SQLConf.PARQUET_INT96_AS_TIMESTAMP.key} to true.") + TimestampType + + case BINARY => + originalType match { + case UTF8 | ENUM => StringType + case null if assumeBinaryIsString => StringType + case null => BinaryType + case DECIMAL => makeDecimalType() + case _ => illegalType() + } + + case FIXED_LEN_BYTE_ARRAY => + originalType match { + case DECIMAL => makeDecimalType(maxPrecisionForBytes(field.getTypeLength)) + case INTERVAL => typeNotImplemented() + case _ => illegalType() + } + + case _ => illegalType() + } + } + if(repetition.equals(Type.Repetition.REPEATED)) { + ArrayType(toPrimitiveType, false) + } else { + toPrimitiveType + } + + } + + private def convertGroupField(field: GroupType): DataType = { + if (field.getRepetition.equals(Type.Repetition.REPEATED)) { + ArrayType(convert(field), false) + } else { + Option(field.getOriginalType).fold(convert(field): DataType) { + // A Parquet list is represented as a 3-level structure: + // + // group (LIST) { + // repeated group list { + // element; + // } + // } + // + // However, according to the most recent Parquet format spec (not released yet up until + // writing), some 2-level structures are also recognized for backwards-compatibility. Thus, + // we need to check whether the 2nd level or the 3rd level refers to list element type. + // + // See: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#lists + case LIST => + CatalystSchemaConverter.analysisRequire( + field.getFieldCount == 1, s"Invalid list type $field") + + val repeatedType = field.getType(0) + CatalystSchemaConverter.analysisRequire( + repeatedType.isRepetition(REPEATED), s"Invalid list type $field") + + if (isElementType(repeatedType, field.getName)) { + ArrayType(convertField(repeatedType), containsNull = false) + } else { + val elementType = repeatedType.asGroupType().getType(0) + val optional = elementType.isRepetition(OPTIONAL) + ArrayType(convertField(elementType), containsNull = optional) + } + + // scalastyle:off + // `MAP_KEY_VALUE` is for backwards-compatibility + // See: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#backward-compatibility-rules-1 + // scalastyle:on + case MAP | MAP_KEY_VALUE => + CatalystSchemaConverter.analysisRequire( + field.getFieldCount == 1 && !field.getType(0).isPrimitive, + s"Invalid map type: $field") + + val keyValueType = field.getType(0).asGroupType() + CatalystSchemaConverter.analysisRequire( + keyValueType.isRepetition(REPEATED) && keyValueType.getFieldCount == 2, + s"Invalid map type: $field") + + val keyType = keyValueType.getType(0) + CatalystSchemaConverter.analysisRequire( + keyType.isPrimitive, + s"Map key type is expected to be a primitive type, but found: $keyType") + + val valueType = keyValueType.getType(1) + val valueOptional = valueType.isRepetition(OPTIONAL) + MapType( + convertField(keyType), + convertField(valueType), + valueContainsNull = valueOptional) + + case _ => + throw new AnalysisException(s"Unrecognized Parquet type: $field") + } + } + } + + // scalastyle:off + // Here we implement Parquet LIST backwards-compatibility rules. + // See: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#backward-compatibility-rules + // scalastyle:on + private def isElementType(repeatedType: Type, parentName: String): Boolean = { + { + // For legacy 2-level list types with primitive element type, e.g.: + // + // // List (nullable list, non-null elements) + // optional group my_list (LIST) { + // repeated int32 element; + // } + // + repeatedType.isPrimitive + } || { + // For legacy 2-level list types whose element type is a group type with 2 or more fields, + // e.g.: + // + // // List> (nullable list, non-null elements) + // optional group my_list (LIST) { + // repeated group element { + // required binary str (UTF8); + // required int32 num; + // }; + // } + // + repeatedType.asGroupType().getFieldCount > 1 + } || { + // For legacy 2-level list types generated by parquet-avro (Parquet version < 1.6.0), e.g.: + // + // // List> (nullable list, non-null elements) + // optional group my_list (LIST) { + // repeated group array { + // required binary str (UTF8); + // }; + // } + // + repeatedType.getName == "array" + } || { + // For Parquet data generated by parquet-thrift, e.g.: + // + // // List> (nullable list, non-null elements) + // optional group my_list (LIST) { + // repeated group my_list_tuple { + // required binary str (UTF8); + // }; + // } + // + repeatedType.getName == s"${parentName}_tuple" + } + } + + /** + * Converts a Spark SQL [[StructType]] to a Parquet [[MessageType]]. + */ + def convert(catalystSchema: StructType): MessageType = { + Types.buildMessage().addFields(catalystSchema.map(convertField): _*).named("root") + } + + /** + * Converts a Spark SQL [[StructField]] to a Parquet [[Type]]. + */ + def convertField(field: StructField): Type = { + convertField(field, if (field.nullable) OPTIONAL else REQUIRED) + } + + private def convertField(field: StructField, repetition: Type.Repetition): Type = { + CatalystSchemaConverter.checkFieldName(field.name) + + field.dataType match { + // =================== + // Simple atomic types + // =================== + + case BooleanType => + Types.primitive(BOOLEAN, repetition).named(field.name) + + case ByteType => + Types.primitive(INT32, repetition).as(INT_8).named(field.name) + + case ShortType => + Types.primitive(INT32, repetition).as(INT_16).named(field.name) + + case IntegerType => + Types.primitive(INT32, repetition).named(field.name) + + case LongType => + Types.primitive(INT64, repetition).named(field.name) + + case FloatType => + Types.primitive(FLOAT, repetition).named(field.name) + + case DoubleType => + Types.primitive(DOUBLE, repetition).named(field.name) + + case StringType => + Types.primitive(BINARY, repetition).as(UTF8).named(field.name) + + case DateType => + Types.primitive(INT32, repetition).as(DATE).named(field.name) + + // NOTE: Spark SQL TimestampType is NOT a well defined type in Parquet format spec. + // + // As stated in PARQUET-323, Parquet `INT96` was originally introduced to represent nanosecond + // timestamp in Impala for some historical reasons, it's not recommended to be used for any + // other types and will probably be deprecated in future Parquet format spec. That's the + // reason why Parquet format spec only defines `TIMESTAMP_MILLIS` and `TIMESTAMP_MICROS` which + // are both logical types annotating `INT64`. + // + // Originally, Spark SQL uses the same nanosecond timestamp type as Impala and Hive. Starting + // from Spark 1.5.0, we resort to a timestamp type with 100 ns precision so that we can store + // a timestamp into a `Long`. This design decision is subject to change though, for example, + // we may resort to microsecond precision in the future. + // + // For Parquet, we plan to write all `TimestampType` value as `TIMESTAMP_MICROS`, but it's + // currently not implemented yet because parquet-mr 1.7.0 (the version we're currently using) + // hasn't implemented `TIMESTAMP_MICROS` yet. + // + // TODO Implements `TIMESTAMP_MICROS` once parquet-mr has that. + case TimestampType => + Types.primitive(INT96, repetition).named(field.name) + + case BinaryType => + Types.primitive(BINARY, repetition).named(field.name) + + // ===================================== + // Decimals (for Spark version <= 1.4.x) + // ===================================== + + // Spark 1.4.x and prior versions only support decimals with a maximum precision of 18 and + // always store decimals in fixed-length byte arrays. To keep compatibility with these older + // versions, here we convert decimals with all precisions to `FIXED_LEN_BYTE_ARRAY` annotated + // by `DECIMAL`. + case DecimalType.Fixed(precision, scale) if !followParquetFormatSpec => + Types + .primitive(FIXED_LEN_BYTE_ARRAY, repetition) + .as(DECIMAL) + .precision(precision) + .scale(scale) + .length(CatalystSchemaConverter.minBytesForPrecision(precision)) + .named(field.name) + + // ===================================== + // Decimals (follow Parquet format spec) + // ===================================== + + // Uses INT32 for 1 <= precision <= 9 + case DecimalType.Fixed(precision, scale) + if precision <= maxPrecisionForBytes(4) && followParquetFormatSpec => + Types + .primitive(INT32, repetition) + .as(DECIMAL) + .precision(precision) + .scale(scale) + .named(field.name) + + // Uses INT64 for 1 <= precision <= 18 + case DecimalType.Fixed(precision, scale) + if precision <= maxPrecisionForBytes(8) && followParquetFormatSpec => + Types + .primitive(INT64, repetition) + .as(DECIMAL) + .precision(precision) + .scale(scale) + .named(field.name) + + // Uses FIXED_LEN_BYTE_ARRAY for all other precisions + case DecimalType.Fixed(precision, scale) if followParquetFormatSpec => + Types + .primitive(FIXED_LEN_BYTE_ARRAY, repetition) + .as(DECIMAL) + .precision(precision) + .scale(scale) + .length(CatalystSchemaConverter.minBytesForPrecision(precision)) + .named(field.name) + + // =================================================== + // ArrayType and MapType (for Spark versions <= 1.4.x) + // =================================================== + + // Spark 1.4.x and prior versions convert ArrayType with nullable elements into a 3-level + // LIST structure. This behavior mimics parquet-hive (1.6.0rc3). Note that this case is + // covered by the backwards-compatibility rules implemented in `isElementType()`. + case ArrayType(elementType, nullable @ true) if !followParquetFormatSpec => + // group (LIST) { + // optional group bag { + // repeated element; + // } + // } + ConversionPatterns.listType( + repetition, + field.name, + Types + .buildGroup(REPEATED) + // "array_element" is the name chosen by parquet-hive (1.7.0 and prior version) + .addField(convertField(StructField("array_element", elementType, nullable))) + .named(CatalystConverter.ARRAY_CONTAINS_NULL_BAG_SCHEMA_NAME)) + + // Spark 1.4.x and prior versions convert ArrayType with non-nullable elements into a 2-level + // LIST structure. This behavior mimics parquet-avro (1.6.0rc3). Note that this case is + // covered by the backwards-compatibility rules implemented in `isElementType()`. + case ArrayType(elementType, nullable @ false) if !followParquetFormatSpec => + // group (LIST) { + // repeated element; + // } + ConversionPatterns.listType( + repetition, + field.name, + // "array" is the name chosen by parquet-avro (1.7.0 and prior version) + convertField(StructField("array", elementType, nullable), REPEATED)) + + // Spark 1.4.x and prior versions convert MapType into a 3-level group annotated by + // MAP_KEY_VALUE. This is covered by `convertGroupField(field: GroupType): DataType`. + case MapType(keyType, valueType, valueContainsNull) if !followParquetFormatSpec => + // group (MAP) { + // repeated group map (MAP_KEY_VALUE) { + // required key; + // value; + // } + // } + ConversionPatterns.mapType( + repetition, + field.name, + convertField(StructField("key", keyType, nullable = false)), + convertField(StructField("value", valueType, valueContainsNull))) + + // ================================================== + // ArrayType and MapType (follow Parquet format spec) + // ================================================== + + case ArrayType(elementType, containsNull) if followParquetFormatSpec => + // group (LIST) { + // repeated group list { + // element; + // } + // } + Types + .buildGroup(repetition).as(LIST) + .addField( + Types.repeatedGroup() + .addField(convertField(StructField("element", elementType, containsNull))) + .named("list")) + .named(field.name) + + case MapType(keyType, valueType, valueContainsNull) => + // group (MAP) { + // repeated group key_value { + // required key; + // value; + // } + // } + Types + .buildGroup(repetition).as(MAP) + .addField( + Types + .repeatedGroup() + .addField(convertField(StructField("key", keyType, nullable = false))) + .addField(convertField(StructField("value", valueType, valueContainsNull))) + .named("key_value")) + .named(field.name) + + // =========== + // Other types + // =========== + + case StructType(fields) => + fields.foldLeft(Types.buildGroup(repetition)) { (builder, field) => + builder.addField(convertField(field)) + }.named(field.name) + + case udt: UserDefinedType[_] => + convertField(field.copy(dataType = udt.sqlType)) + + case _ => + throw new AnalysisException(s"Unsupported data type $field.dataType") + } + } + + // Max precision of a decimal value stored in `numBytes` bytes + private def maxPrecisionForBytes(numBytes: Int): Int = { + Math.round( // convert double to long + Math.floor(Math.log10( // number of base-10 digits + Math.pow(2, 8 * numBytes - 1) - 1))) // max value stored in numBytes + .asInstanceOf[Int] + } +} + diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/SchemaConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/SchemaConverter.scala new file mode 100644 index 0000000000000..544ba7c8c559a --- /dev/null +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/SchemaConverter.scala @@ -0,0 +1,26 @@ +/* + * 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.spark.sql.parquet + +import org.apache.parquet.schema.{Type, MessageType} +import org.apache.spark.sql.types.{StructField, StructType} + +trait SchemaConverter { + def convert(parquetSchema: MessageType): StructType + def convert(catalystSchema: StructType): MessageType + def convertField(field: StructField): Type +} diff --git a/sql/core/src/test/resources/nested-array-struct.parquet b/sql/core/src/test/resources/nested-array-struct.parquet new file mode 100755 index 0000000000000000000000000000000000000000..41a43fa35d39685e56ba4849a16cba4bb1aa86ae GIT binary patch literal 775 zcmaKr-%G+!6vvO#=8u;i;*JSE3{j~lAyWtuV%0Q3*U(Y)B(q&>u(@?NBZ;2+Px=dc z?6I>s%N6u+cQ4=bIp1@3cBjdsBLbvCDhGte15a`Q8~~)V;d2WY3V@LYX{-@GMj#!6 z`;fvdgDblto20oxMhs#hdKzt*4*3w}iuUE6PW?b*Zs1NAv%1WfvAnT@2NhLn_L#fy zHFWX={q7*H z93@^0*O&w#e53@lJ`hFEV2=wL)V**Pb(8vc%<=-4iJz&t;n22J{%<(t!px$!DZLaV zDaOCYR1UR;Go`F89pTwFrqpgr1NlrDOs+J&f2GO;)PtpmW%OH3neOEccXHoWyO`6Bl5({+ea14&qL7DtETw`(h_426$BxCYApN S1!5siKXe$p;U(Ab7x)6M)yB5~ literal 0 HcmV?d00001 diff --git a/sql/core/src/test/resources/old-repeated-int.parquet b/sql/core/src/test/resources/old-repeated-int.parquet new file mode 100755 index 0000000000000000000000000000000000000000..520922f73ebb75950c4e65dec6689af817cb7a33 GIT binary patch literal 389 zcmZWmO-sW-5FMkeB_3tN1_Ca@_7p=~Z@EPbSg5juTs)Ocvz0>9#NEw7ivQhdDcI1< z%;U}1booNUUY~PehCwzvumZho_zD!@TVtKoQHivd-PzgS{O4oWwfk)ZsDnB!ByvMUB7gt@Rk50{GMw}6DWn-w!#F0CGZwP` zh81XVct9peJU!6=O6yx`ZywS;EGVOwOOIsCr3p)d)y(vgv`4bce)5D7UiKlH0l6Ga0+m-EijTt zM!X)Rd#pSj~EkCao0il-K=62)db~64ZC7r8d1AwIhzFkoG;e&AbpZW#pJ&{5H literal 0 HcmV?d00001 diff --git a/sql/core/src/test/resources/old-repeated.parquet b/sql/core/src/test/resources/old-repeated.parquet new file mode 100644 index 0000000000000000000000000000000000000000..213f1a90291b30a8a3161b51c38f008f3ae9f6e5 GIT binary patch literal 432 zcmZWm!D@p*5ZxNF!5+)X3PMGioUA16Ei?y9g$B|h;-#ms>Ld--Xm{5`DgF13A<&42 znSH!BJNtMWhsm50I-@h68VC$(I7}ZALYRJm-NGUo*2p;a%Z@xEJgH{;FE=Sj6^mNc zS-TAqXn-pyRtNP8Qt};84d*60yAuBru{7JUo$1)Y6%&KlJ(Uv6u!JS1zOP)cw zaM$5ewB9699EEB0jJ*18aDDn7N1N4K`fzXlnuJ~V?c^nwk}Yeo3wXox4+#3Y!pMU2 V+-`?%2{TWZ?kYh(G4~k%>JK8=aDe~- literal 0 HcmV?d00001 diff --git a/sql/core/src/test/resources/proto-repeated-string.parquet b/sql/core/src/test/resources/proto-repeated-string.parquet new file mode 100755 index 0000000000000000000000000000000000000000..8a7eea601d0164b1177ab90efb7116c6e4c900da GIT binary patch literal 411 zcmY*W!D_-l5S^$E62wc{umKMtR4+{f-b!vM4Q)Y6&|G?w#H^aKansF;gi?C#*Yq1Z zv6e;{_C4Mk<_)t^FrN}2UmBK6hDddy19SkO`+9soFOY8;=b|A8A$itAvJoQdBBnKK zKy>)#|`4$W^3YtqL)WN5pTmWh1ZGv$>{f|s#sCG%1VN%LJ&FyD4siH@<(8PDu@ z!?sWEU$)ao`yyr1x2MQ?k}~ewv*0eAE$3kr261?gx~fYY8oxy0auLs;o*#@41L)=X h7Au}q6}>(e6`sLs-{PvZ8BpWYeN#xd)c_*=mLHLcZu|fM literal 0 HcmV?d00001 diff --git a/sql/core/src/test/resources/proto-repeated-struct.parquet b/sql/core/src/test/resources/proto-repeated-struct.parquet new file mode 100755 index 0000000000000000000000000000000000000000..c29eee35c350e84efa2818febb2528309a8ac3ea GIT binary patch literal 608 zcma))-D<)x6vvP811a8(bSZdI%9Js*tjca=2puciK%r=F1_P}cr%>B2jf^q&4ts<> z%r1PaC9^8^%A4e$li&GFTzg<)z+K#J;DQh(TmnDm&bFDCfsEak0$H6=|yp$CW-$_F@l={DK5j1GEpn8)DX!>A+15G`Fpg} zMZREE-l#~cYPa=r6<4%c3AD?tzx2bP7Sypiu9pGoi(^0p+XD*$Y;s4$HpQOVL*dnD3MsZWzL<}ml5ZY`6p``6p3uzOR6cOQizQ3hI@;TGm z8hy+Rm>Hl!&aiC8oEI4i7`u<#dC`r5%q<5r!9eDg1IFy*c2RU=AalzBO)!wT<$y7e zS0C?=T^uJ)6ePiDIW^oM?BO`}o-pLWHOS&h@*H8x zD56?ZuNu`Fl-0Tj)YHv=x(@J>C=j)+#mj%Z_4kgdp}D_<3b zc(o7;z363$6CCr9}+-E j<-Z;KUL2!lIhl~NDb+dIHUN;6ire!D{E~S&5gg5S?rsql%Icnq5|qgD{N<#x?oCY2!n{ZAEKv64iDOJsH{F!~)4uB-v0( z@BJM;_owufA5^-l4@Z{ekVAVh>zOxi-n`LDMyq>_0q^0x8b74l?^SjJrp%q&06h8-y4iMdSJ@MDH4c~HjX3j($=&sN1 zW|q&!OYxG3d&~^8@dlzhDa$1b0`rz(6tkBD*J153G=T1;gzF$B0g1WSKnPOy6M$~KQ|W5 z5A#DO!$$Zca>TK`;67WBib&?m76{4rqTmNgwH)UCc)*uPhjcg;fc!=Tfl{N?GyS_6 z3+tZPeSOS=k#BjS>(f75Q`2EhwX*hMsK_@Kv&ZT;SydBky3mDR6_J}cL*_TtV}7>H zA+wumr}b9v46coS`}(TY;qmaR$9wg^82X@n)jvIvzps*~J`|Fl new CatalystSchemaConverter( + assumeBinaryIsString = binaryAsString, + assumeInt96IsTimestamp = int96AsTimestamp, + followParquetFormatSpec = followParquetFormatSpec) + case true => + new ProtobufCatalystSchemaConverter(assumeBinaryIsString = binaryAsString, + assumeInt96IsTimestamp = int96AsTimestamp, + followParquetFormatSpec = followParquetFormatSpec) + } test(s"sql <= parquet: $testName") { val actual = converter.convert(MessageTypeParser.parseMessageType(parquetSchema)) @@ -457,6 +465,7 @@ class ParquetSchemaSuite extends ParquetSchemaTest { // Tests for converting Parquet LIST to Catalyst ArrayType // ======================================================= + testParquetToCatalyst( "Backwards-compatibility: LIST with nullable element type - 1 - standard", StructType(Seq( @@ -585,6 +594,52 @@ class ParquetSchemaSuite extends ParquetSchemaTest { |} """.stripMargin) + testParquetToCatalyst( + "parquet-protobuf primitive list compatibility", + StructType(Seq( + StructField( + "f1", + ArrayType(IntegerType, containsNull = false), + nullable = false))), + """message root { + | repeated int32 f1; + |} + """.stripMargin,isProtobufDerived = true) + + testParquetToCatalyst( + "parquet-protobuf struct list compatibility", + StructType(Seq( + StructField( + "inner", + ArrayType( + StructType(Seq( + StructField("one", StringType, nullable=true), + StructField("two", StringType, nullable=true), + StructField("three", StringType, nullable=true))), containsNull = false), + nullable = false))), + """message root { + | repeated group inner { + | optional binary one (UTF8); + | optional binary two (UTF8); + | optional binary three (UTF8); + | } + |} + """.stripMargin,isProtobufDerived = true) + + testParquetToCatalyst( + "parquet-protobuf string list compatibility", + StructType(Seq( + StructField( + "f1", + ArrayType(StringType, containsNull = false), + nullable = false))), + """message root { + | repeated binary f1 (UTF8); + |} + """.stripMargin,isProtobufDerived = true) + + + // ======================================================= // Tests for converting Catalyst ArrayType to Parquet LIST // ======================================================= diff --git a/sql/core/src/test/scala/org/apache/spark/sql/parquet/ProtoParquetTypesConverterTest.scala b/sql/core/src/test/scala/org/apache/spark/sql/parquet/ProtoParquetTypesConverterTest.scala new file mode 100644 index 0000000000000..52f0b9ea61f7c --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/parquet/ProtoParquetTypesConverterTest.scala @@ -0,0 +1,91 @@ +package org.apache.spark.sql.parquet + +import java.net.URL + +import org.apache.parquet.schema.{GroupType, PrimitiveType, MessageType} +import org.apache.spark.SparkFunSuite +import org.apache.spark.sql.{Row, QueryTest, DataFrame, SQLContext} +import org.apache.spark.sql.catalyst.expressions.{GenericRow, Attribute} +import org.apache.spark.sql.test.TestSQLContext +import org.scalatest.FunSuite + + +import scala.collection.mutable +import scala.collection.mutable.ArrayBuffer + +class ProtoParquetTypesConverterTest extends QueryTest with ParquetTest { + override val sqlContext: SQLContext = TestSQLContext + + def fetchRows(file: String, table: String ): Array[Row] = { + val resource: URL = getClass.getResource(file) + val pf: DataFrame = sqlContext.read.parquet(resource.toURI.toString) + pf.registerTempTable(table) + sqlContext.sql("select * from " + table).collect() + } + + test("should work with repeated primitive") { + val rows: Array[Row] = fetchRows("/old-repeated-int.parquet", "repeated_int") + assert(rows(0) == Row(Seq(1,2,3))) + } + + test("should work with repeated complex") { + val rows: Array[Row] = fetchRows("/old-repeated-message.parquet", "repeated_struct") + val array: mutable.WrappedArray[GenericRow] = rows(0)(0).asInstanceOf[mutable.WrappedArray[GenericRow]] + assert(array.length === 3) + assert(array(0)=== Row("First inner",null,null)) + assert(array(1) === Row(null,"Second inner",null)) + assert(array(2) === Row(null, null,"Third inner")) + } + + + test("should work with repeated complex with more than one item in array") { + val rows: Array[Row] = fetchRows("/proto-repeated-struct.parquet", "my_complex_table") + assert(rows.length === 1) + val array: mutable.WrappedArray[GenericRow] = rows(0)(0).asInstanceOf[mutable.WrappedArray[GenericRow]] + assert(array.length === 2) + assert(array(0) === Row("0 - 1", "0 - 2", "0 - 3")) + assert(array(1) === Row("1 - 1", "1 - 2", "1 - 3")) + } + + test("should work with repeated complex with many rows") { + val rows: Array[Row] = fetchRows("/proto-struct-with-array-many.parquet", "many_complex_rows") + assert(rows.length === 3) + val row0: mutable.WrappedArray[GenericRow] = rows(0)(0).asInstanceOf[mutable.WrappedArray[GenericRow]] + val row1: mutable.WrappedArray[GenericRow] = rows(1)(0).asInstanceOf[mutable.WrappedArray[GenericRow]] + val row2: mutable.WrappedArray[GenericRow] = rows(2)(0).asInstanceOf[mutable.WrappedArray[GenericRow]] + assert(row0(0) === Row("0 - 0 - 1", "0 - 0 - 2","0 - 0 - 3")) + assert(row0(1)=== Row("0 - 1 - 1", "0 - 1 - 2", "0 - 1 - 3")) + assert(row1(0) === Row("1 - 0 - 1", "1 - 0 - 2","1 - 0 - 3")) + assert(row1(1) === Row("1 - 1 - 1", "1 - 1 - 2", "1 - 1 - 3")) + assert(row2(0) === Row("2 - 0 - 1", "2 - 0 - 2","2 - 0 - 3")) + assert(row2(1) === Row("2 - 1 - 1", "2 - 1 - 2", "2 - 1 - 3")) + } + + test("should work with complex type containing array") { + val rows: Array[Row] = fetchRows("/proto-struct-with-array.parquet", "struct_containing_array") + assert(rows.length === 1) + val theRow: GenericRow = rows(0).asInstanceOf[GenericRow] + val expected = Row(10,9,null,null,Row(9),Seq(Row(9),Row(10))) + assert(theRow === expected) + } + + test("should work with multiple levels of nesting") { + val rows: Array[Row] = fetchRows("/nested-array-struct.parquet", "multiple_nesting") + assert(rows.length === 3) + val row0: GenericRow = rows(0).asInstanceOf[GenericRow] + val row1: GenericRow = rows(1).asInstanceOf[GenericRow] + val row2: GenericRow = rows(2).asInstanceOf[GenericRow] + assert(row0 === Row(2, Seq(Row(1,Seq(Row(3)))))) + assert(row1 === Row(5, Seq(Row(4,Seq(Row(6)))))) + assert(row2 === Row(8, Seq(Row(7,Seq(Row(9)))))) + + } + + test("should convert array of strings") { + val rows: Array[Row] = fetchRows("/proto-repeated-string.parquet", "strings") + assert(rows.length === 3) + assert(rows(0)=== Row(Seq("hello", "world"))) + assert(rows(1)=== Row(Seq("good", "bye"))) + assert(rows(2)=== Row(Seq("one","two", "three"))) + } +}