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 0000000000000..41a43fa35d396 Binary files /dev/null and b/sql/core/src/test/resources/nested-array-struct.parquet differ 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 0000000000000..520922f73ebb7 Binary files /dev/null and b/sql/core/src/test/resources/old-repeated-int.parquet differ diff --git a/sql/core/src/test/resources/old-repeated-message.parquet b/sql/core/src/test/resources/old-repeated-message.parquet new file mode 100755 index 0000000000000..548db99162777 Binary files /dev/null and b/sql/core/src/test/resources/old-repeated-message.parquet differ 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 0000000000000..213f1a90291b3 Binary files /dev/null and b/sql/core/src/test/resources/old-repeated.parquet differ 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 0000000000000..8a7eea601d016 Binary files /dev/null and b/sql/core/src/test/resources/proto-repeated-string.parquet differ 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 0000000000000..c29eee35c350e Binary files /dev/null and b/sql/core/src/test/resources/proto-repeated-struct.parquet differ diff --git a/sql/core/src/test/resources/proto-struct-with-array-many.parquet b/sql/core/src/test/resources/proto-struct-with-array-many.parquet new file mode 100755 index 0000000000000..ff9809675fc04 Binary files /dev/null and b/sql/core/src/test/resources/proto-struct-with-array-many.parquet differ diff --git a/sql/core/src/test/resources/proto-struct-with-array.parquet b/sql/core/src/test/resources/proto-struct-with-array.parquet new file mode 100755 index 0000000000000..325a8370ad20e Binary files /dev/null and b/sql/core/src/test/resources/proto-struct-with-array.parquet differ diff --git a/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetQuerySuite.scala index 5c65a8ec57f00..b6f6b617d0ef1 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetQuerySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetQuerySuite.scala @@ -22,7 +22,7 @@ import java.io.File import org.apache.hadoop.fs.Path import org.apache.spark.sql.types._ -import org.apache.spark.sql.{QueryTest, Row, SQLConf} +import org.apache.spark.sql.{DataFrame, QueryTest, Row, SQLConf} import org.apache.spark.util.Utils /** @@ -198,7 +198,8 @@ class ParquetQuerySuite extends QueryTest with ParquetTest { val df = sqlContext.createDataFrame(rowRDD, schema) df.write.parquet(basePath) - val decimal = sqlContext.read.parquet(basePath).first().getDecimal(0) + val parquet: DataFrame = sqlContext.read.parquet(basePath) + val decimal = parquet.first().getDecimal(0) assert(Decimal("67123.45") === Decimal(decimal)) } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetSchemaSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetSchemaSuite.scala index 4a0b3b60f419d..412507527579c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetSchemaSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetSchemaSuite.scala @@ -57,11 +57,19 @@ abstract class ParquetSchemaTest extends SparkFunSuite with ParquetTest { binaryAsString: Boolean = true, int96AsTimestamp: Boolean = true, followParquetFormatSpec: Boolean = false, - isThriftDerived: Boolean = false): Unit = { - val converter = new CatalystSchemaConverter( - assumeBinaryIsString = binaryAsString, - assumeInt96IsTimestamp = int96AsTimestamp, - followParquetFormatSpec = followParquetFormatSpec) + isThriftDerived: Boolean = false, + isProtobufDerived: Boolean = false): Unit = { + val converter = + isProtobufDerived match { + case false => 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"))) + } +}