You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by liancheng <gi...@git.apache.org> on 2015/04/08 15:21:42 UTC
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
GitHub user liancheng opened a pull request:
https://github.com/apache/spark/pull/5422
[SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactors converters to convert Parquet records to row objects
This PR simplifies the `CatalystConverter` class hierarchy, and implements backwards-compatibility rules for them. The new class hierarchy basically resembles `AvroIndexedRecordConverter` in parquet-avro.
This is still in WIP status. I'm opening this PR to check whether it passes Jenkins.
TODO:
- [ ] Tests for legacy Parquet data format.
Major changes:
- Replaces `CatalystConverter` trait with a much simpler `ParentContainerUpdater`.
Now instead of extending the original `CatalystConverter` trait, every converter class accepts an updater whose's responsible to set the converted value to some parent container. For example, appending array elements to a parent array buffer, appending a key-value pairs to a parent mutable map, or setting a converted value to some specific field of a parent row.
This simplifies the design since converters don't need to care about details of their parent converters anymore.
- Replaces `CatalystRootConverter`, `CatalystGroupConverter` and `CatalystPrimitiveRowConverter` with `CatalystRowConverter`
Specifically, now all row objects are represented with `SpecificMutableRow` during conversion.
- Removes `CatalystArrayContainsNullConverter` and `CatalystNativeArrayConverter`, revamped `CatalystArrayConverter`
`CatalystNativeArrayConverter` should be designed with the intention of avoiding boxing costs. However, the way it uses Scala generics actually doesn't achieve this goal.
The new `CatalystArrayConverter` handles both nullable and non-nullable array elements.
- Implemented backwards-compatibility rules in `CatalystArrayConverter`
When Parquet records are being converted, schema of Parquet files should have already been verified. So we only need to care about the structure rather than field names in the Parquet schema. Since all map objects represented in legacy systems have the same structure as the standard one (see [backwards-compatibility rules for MAP] [1]), we only need to deal with LIST (namely array) here.
[1]: https://github.com/apache/incubator-parquet-format/blob/master/LogicalTypes.md#backward-compatibility-rules-1
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/liancheng/spark refactor-parquet-converters
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/5422.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5422
----
commit 1ed764bad9fa093770ad64a0ea46f51db0a9b208
Author: Cheng Lian <li...@databricks.com>
Date: 2015-03-07T16:21:50Z
Initial draft for Parquet converters refactoring
commit 43e2252966a74823fa514d93904c7daa92ed3aef
Author: Cheng Lian <li...@databricks.com>
Date: 2015-04-01T16:12:22Z
Removes old converters
commit 184eb278456ce24436b8a5b10020ff3016af14e6
Author: Cheng Lian <li...@databricks.com>
Date: 2015-04-01T16:36:45Z
Minor cleanups
commit 058ad8410485b1365eac077296d10127d2affaf1
Author: Cheng Lian <li...@databricks.com>
Date: 2015-04-02T10:28:56Z
More cleanups, and comments
commit 0de2e8b83fb4b616603c2013792db3351c82f324
Author: Cheng Lian <li...@databricks.com>
Date: 2015-04-06T11:58:57Z
Moves converter classes back to ParquetConverter.scala
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90993587
[Test build #29869 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29869/consoleFull) for PR 5422 at commit [`2529b76`](https://github.com/apache/spark/commit/2529b76141c068fe69a03e29ddc50ca496cd9dcd).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
* This patch does not change any dependencies.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90965416
[Test build #29868 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29868/consoleFull) for PR 5422 at commit [`02d78c4`](https://github.com/apache/spark/commit/02d78c47801998af1fc9c76cc400acfd57ccbe33).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-96801888
[Test build #31021 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/31021/consoleFull) for PR 5422 at commit [`2529b76`](https://github.com/apache/spark/commit/2529b76141c068fe69a03e29ddc50ca496cd9dcd).
* This patch **fails Spark unit tests**.
* This patch **does not merge cleanly**.
* This patch adds no public classes.
* This patch **adds the following new dependencies:**
* `RoaringBitmap-0.4.5.jar`
* `activation-1.1.jar`
* `akka-actor_2.10-2.3.4-spark.jar`
* `akka-remote_2.10-2.3.4-spark.jar`
* `akka-slf4j_2.10-2.3.4-spark.jar`
* `aopalliance-1.0.jar`
* `arpack_combined_all-0.1.jar`
* `avro-1.7.7.jar`
* `breeze-macros_2.10-0.11.2.jar`
* `breeze_2.10-0.11.2.jar`
* `chill-java-0.5.0.jar`
* `chill_2.10-0.5.0.jar`
* `commons-beanutils-1.7.0.jar`
* `commons-beanutils-core-1.8.0.jar`
* `commons-cli-1.2.jar`
* `commons-codec-1.10.jar`
* `commons-collections-3.2.1.jar`
* `commons-compress-1.4.1.jar`
* `commons-configuration-1.6.jar`
* `commons-digester-1.8.jar`
* `commons-httpclient-3.1.jar`
* `commons-io-2.1.jar`
* `commons-lang-2.5.jar`
* `commons-lang3-3.3.2.jar`
* `commons-math-2.1.jar`
* `commons-math3-3.1.1.jar`
* `commons-net-2.2.jar`
* `compress-lzf-1.0.0.jar`
* `config-1.2.1.jar`
* `core-1.1.2.jar`
* `curator-client-2.4.0.jar`
* `curator-framework-2.4.0.jar`
* `curator-recipes-2.4.0.jar`
* `gmbal-api-only-3.0.0-b023.jar`
* `grizzly-framework-2.1.2.jar`
* `grizzly-http-2.1.2.jar`
* `grizzly-http-server-2.1.2.jar`
* `grizzly-http-servlet-2.1.2.jar`
* `grizzly-rcm-2.1.2.jar`
* `groovy-all-2.3.7.jar`
* `guava-14.0.1.jar`
* `guice-3.0.jar`
* `hadoop-annotations-2.2.0.jar`
* `hadoop-auth-2.2.0.jar`
* `hadoop-client-2.2.0.jar`
* `hadoop-common-2.2.0.jar`
* `hadoop-hdfs-2.2.0.jar`
* `hadoop-mapreduce-client-app-2.2.0.jar`
* `hadoop-mapreduce-client-common-2.2.0.jar`
* `hadoop-mapreduce-client-core-2.2.0.jar`
* `hadoop-mapreduce-client-jobclient-2.2.0.jar`
* `hadoop-mapreduce-client-shuffle-2.2.0.jar`
* `hadoop-yarn-api-2.2.0.jar`
* `hadoop-yarn-client-2.2.0.jar`
* `hadoop-yarn-common-2.2.0.jar`
* `hadoop-yarn-server-common-2.2.0.jar`
* `ivy-2.4.0.jar`
* `jackson-annotations-2.4.0.jar`
* `jackson-core-2.4.4.jar`
* `jackson-core-asl-1.8.8.jar`
* `jackson-databind-2.4.4.jar`
* `jackson-jaxrs-1.8.8.jar`
* `jackson-mapper-asl-1.8.8.jar`
* `jackson-module-scala_2.10-2.4.4.jar`
* `jackson-xc-1.8.8.jar`
* `jansi-1.4.jar`
* `javax.inject-1.jar`
* `javax.servlet-3.0.0.v201112011016.jar`
* `javax.servlet-3.1.jar`
* `javax.servlet-api-3.0.1.jar`
* `jaxb-api-2.2.2.jar`
* `jaxb-impl-2.2.3-1.jar`
* `jcl-over-slf4j-1.7.10.jar`
* `jersey-client-1.9.jar`
* `jersey-core-1.9.jar`
* `jersey-grizzly2-1.9.jar`
* `jersey-guice-1.9.jar`
* `jersey-json-1.9.jar`
* `jersey-server-1.9.jar`
* `jersey-test-framework-core-1.9.jar`
* `jersey-test-framework-grizzly2-1.9.jar`
* `jets3t-0.7.1.jar`
* `jettison-1.1.jar`
* `jetty-util-6.1.26.jar`
* `jline-0.9.94.jar`
* `jline-2.10.4.jar`
* `jodd-core-3.6.3.jar`
* `json4s-ast_2.10-3.2.10.jar`
* `json4s-core_2.10-3.2.10.jar`
* `json4s-jackson_2.10-3.2.10.jar`
* `jsr305-1.3.9.jar`
* `jtransforms-2.4.0.jar`
* `jul-to-slf4j-1.7.10.jar`
* `kryo-2.21.jar`
* `log4j-1.2.17.jar`
* `lz4-1.2.0.jar`
* `management-api-3.0.0-b012.jar`
* `mesos-0.21.0-shaded-protobuf.jar`
* `metrics-core-3.1.0.jar`
* `metrics-graphite-3.1.0.jar`
* `metrics-json-3.1.0.jar`
* `metrics-jvm-3.1.0.jar`
* `minlog-1.2.jar`
* `netty-3.8.0.Final.jar`
* `netty-all-4.0.23.Final.jar`
* `objenesis-1.2.jar`
* `opencsv-2.3.jar`
* `oro-2.0.8.jar`
* `paranamer-2.6.jar`
* `parquet-column-1.6.0rc3.jar`
* `parquet-common-1.6.0rc3.jar`
* `parquet-encoding-1.6.0rc3.jar`
* `parquet-format-2.2.0-rc1.jar`
* `parquet-generator-1.6.0rc3.jar`
* `parquet-hadoop-1.6.0rc3.jar`
* `parquet-jackson-1.6.0rc3.jar`
* `protobuf-java-2.4.1.jar`
* `protobuf-java-2.5.0-spark.jar`
* `py4j-0.8.2.1.jar`
* `pyrolite-2.0.1.jar`
* `quasiquotes_2.10-2.0.1.jar`
* `reflectasm-1.07-shaded.jar`
* `scala-compiler-2.10.4.jar`
* `scala-library-2.10.4.jar`
* `scala-reflect-2.10.4.jar`
* `scalap-2.10.4.jar`
* `scalatest_2.10-2.2.1.jar`
* `slf4j-api-1.7.10.jar`
* `slf4j-log4j12-1.7.10.jar`
* `snappy-java-1.1.1.6.jar`
* `spark-bagel_2.10-1.4.0-SNAPSHOT.jar`
* `spark-catalyst_2.10-1.4.0-SNAPSHOT.jar`
* `spark-core_2.10-1.4.0-SNAPSHOT.jar`
* `spark-graphx_2.10-1.4.0-SNAPSHOT.jar`
* `spark-launcher_2.10-1.4.0-SNAPSHOT.jar`
* `spark-mllib_2.10-1.4.0-SNAPSHOT.jar`
* `spark-network-common_2.10-1.4.0-SNAPSHOT.jar`
* `spark-network-shuffle_2.10-1.4.0-SNAPSHOT.jar`
* `spark-repl_2.10-1.4.0-SNAPSHOT.jar`
* `spark-sql_2.10-1.4.0-SNAPSHOT.jar`
* `spark-streaming_2.10-1.4.0-SNAPSHOT.jar`
* `spire-macros_2.10-0.7.4.jar`
* `spire_2.10-0.7.4.jar`
* `stax-api-1.0.1.jar`
* `stream-2.7.0.jar`
* `tachyon-0.5.0.jar`
* `tachyon-client-0.5.0.jar`
* `uncommons-maths-1.2.2a.jar`
* `unused-1.0.0.jar`
* `xmlenc-0.52.jar`
* `xz-1.0.jar`
* `zookeeper-3.4.5.jar`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5422#discussion_r27969335
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetConverter.scala ---
@@ -18,480 +18,408 @@
package org.apache.spark.sql.parquet
import java.sql.Timestamp
-import java.util.{TimeZone, Calendar}
+import java.util.{Calendar, TimeZone}
-import scala.collection.mutable.{Buffer, ArrayBuffer, HashMap}
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
import jodd.datetime.JDateTime
import parquet.column.Dictionary
-import parquet.io.api.{PrimitiveConverter, GroupConverter, Binary, Converter}
-import parquet.schema.MessageType
+import parquet.io.api.{Binary, Converter, GroupConverter, PrimitiveConverter}
+import parquet.schema.{GroupType, PrimitiveType => ParquetPrimitiveType, Type}
import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.parquet.CatalystConverter.FieldType
-import org.apache.spark.sql.types._
import org.apache.spark.sql.parquet.timestamp.NanoTime
+import org.apache.spark.sql.types._
/**
- * Collection of converters of Parquet types (group and primitive types) that
- * model arrays and maps. The conversions are partly based on the AvroParquet
- * converters that are part of Parquet in order to be able to process these
- * types.
- *
- * There are several types of converters:
- * <ul>
- * <li>[[org.apache.spark.sql.parquet.CatalystPrimitiveConverter]] for primitive
- * (numeric, boolean and String) types</li>
- * <li>[[org.apache.spark.sql.parquet.CatalystNativeArrayConverter]] for arrays
- * of native JVM element types; note: currently null values are not supported!</li>
- * <li>[[org.apache.spark.sql.parquet.CatalystArrayConverter]] for arrays of
- * arbitrary element types (including nested element types); note: currently
- * null values are not supported!</li>
- * <li>[[org.apache.spark.sql.parquet.CatalystStructConverter]] for structs</li>
- * <li>[[org.apache.spark.sql.parquet.CatalystMapConverter]] for maps; note:
- * currently null values are not supported!</li>
- * <li>[[org.apache.spark.sql.parquet.CatalystPrimitiveRowConverter]] for rows
- * of only primitive element types</li>
- * <li>[[org.apache.spark.sql.parquet.CatalystGroupConverter]] for other nested
- * records, including the top-level row record</li>
- * </ul>
+ * A [[ParentContainerUpdater]] is used by a Parquet converter to set converted values to some
+ * corresponding parent container. For example, a converter for a `Struct` field may set converted
+ * values to a [[MutableRow]]; or a converter for array element may append converted values to an
+ * [[ArrayBuffer]].
*/
-
-private[sql] object CatalystConverter {
- // The type internally used for fields
- type FieldType = StructField
-
- // This is mostly Parquet convention (see, e.g., `ConversionPatterns`).
- // Note that "array" for the array elements is chosen by ParquetAvro.
- // Using a different value will result in Parquet silently dropping columns.
- val ARRAY_CONTAINS_NULL_BAG_SCHEMA_NAME = "bag"
- val ARRAY_ELEMENTS_SCHEMA_NAME = "array"
- // SPARK-4520: Thrift generated parquet files have different array element
- // schema names than avro. Thrift parquet uses array_schema_name + "_tuple"
- // as opposed to "array" used by default. For more information, check
- // TestThriftSchemaConverter.java in parquet.thrift.
- val THRIFT_ARRAY_ELEMENTS_SCHEMA_NAME_SUFFIX = "_tuple"
- val MAP_KEY_SCHEMA_NAME = "key"
- val MAP_VALUE_SCHEMA_NAME = "value"
- val MAP_SCHEMA_NAME = "map"
-
- // TODO: consider using Array[T] for arrays to avoid boxing of primitive types
- type ArrayScalaType[T] = Seq[T]
- type StructScalaType[T] = Row
- type MapScalaType[K, V] = Map[K, V]
-
- protected[parquet] def createConverter(
- field: FieldType,
- fieldIndex: Int,
- parent: CatalystConverter): Converter = {
- val fieldType: DataType = field.dataType
- fieldType match {
- case udt: UserDefinedType[_] => {
- createConverter(field.copy(dataType = udt.sqlType), fieldIndex, parent)
- }
- // For native JVM types we use a converter with native arrays
- case ArrayType(elementType: NativeType, false) => {
- new CatalystNativeArrayConverter(elementType, fieldIndex, parent)
- }
- // This is for other types of arrays, including those with nested fields
- case ArrayType(elementType: DataType, false) => {
- new CatalystArrayConverter(elementType, fieldIndex, parent)
- }
- case ArrayType(elementType: DataType, true) => {
- new CatalystArrayContainsNullConverter(elementType, fieldIndex, parent)
- }
- case StructType(fields: Array[StructField]) => {
- new CatalystStructConverter(fields, fieldIndex, parent)
- }
- case MapType(keyType: DataType, valueType: DataType, valueContainsNull: Boolean) => {
- new CatalystMapConverter(
- Array(
- new FieldType(MAP_KEY_SCHEMA_NAME, keyType, false),
- new FieldType(MAP_VALUE_SCHEMA_NAME, valueType, valueContainsNull)),
- fieldIndex,
- parent)
- }
- // Strings, Shorts and Bytes do not have a corresponding type in Parquet
- // so we need to treat them separately
- case StringType =>
- new CatalystPrimitiveStringConverter(parent, fieldIndex)
- case ShortType => {
- new CatalystPrimitiveConverter(parent, fieldIndex) {
- override def addInt(value: Int): Unit =
- parent.updateShort(fieldIndex, value.asInstanceOf[ShortType.JvmType])
- }
- }
- case ByteType => {
- new CatalystPrimitiveConverter(parent, fieldIndex) {
- override def addInt(value: Int): Unit =
- parent.updateByte(fieldIndex, value.asInstanceOf[ByteType.JvmType])
- }
- }
- case DateType => {
- new CatalystPrimitiveConverter(parent, fieldIndex) {
- override def addInt(value: Int): Unit =
- parent.updateDate(fieldIndex, value.asInstanceOf[DateType.JvmType])
- }
- }
- case d: DecimalType => {
- new CatalystPrimitiveConverter(parent, fieldIndex) {
- override def addBinary(value: Binary): Unit =
- parent.updateDecimal(fieldIndex, value, d)
- }
- }
- case TimestampType => {
- new CatalystPrimitiveConverter(parent, fieldIndex) {
- override def addBinary(value: Binary): Unit =
- parent.updateTimestamp(fieldIndex, value)
- }
- }
- // All other primitive types use the default converter
- case ctype: PrimitiveType => { // note: need the type tag here!
- new CatalystPrimitiveConverter(parent, fieldIndex)
- }
- case _ => throw new RuntimeException(
- s"unable to convert datatype ${field.dataType.toString} in CatalystConverter")
- }
- }
-
- protected[parquet] def createRootConverter(
- parquetSchema: MessageType,
- attributes: Seq[Attribute]): CatalystConverter = {
- // For non-nested types we use the optimized Row converter
- if (attributes.forall(a => ParquetTypesConverter.isPrimitiveType(a.dataType))) {
- new CatalystPrimitiveRowConverter(attributes.toArray)
- } else {
- new CatalystGroupConverter(attributes.toArray)
- }
- }
+private[parquet] trait ParentContainerUpdater {
+ def set(value: Any): Unit = ()
+ def setBoolean(value: Boolean): Unit = set(value)
+ def setByte(value: Byte): Unit = set(value)
+ def setShort(value: Short): Unit = set(value)
+ def setInt(value: Int): Unit = set(value)
+ def setLong(value: Long): Unit = set(value)
+ def setFloat(value: Float): Unit = set(value)
+ def setDouble(value: Double): Unit = set(value)
}
-private[parquet] abstract class CatalystConverter extends GroupConverter {
- /**
- * The number of fields this group has
- */
- protected[parquet] val size: Int
+/** A no-op updater used for root converter (who doesn't have a parent). */
+private[parquet] object NoopUpdater extends ParentContainerUpdater
- /**
- * The index of this converter in the parent
- */
- protected[parquet] val index: Int
+/**
+ * This Parquet converter converts Parquet records to Spark SQL rows.
+ *
+ * @param parquetType Parquet type of Parquet records
+ * @param sparkType A Spark SQL struct type that corresponds to the Parquet record type
+ * @param updater An updater which takes care of the converted row object
+ */
+private[parquet] class CatalystRowConverter(
+ parquetType: GroupType,
+ sparkType: StructType,
+ updater: ParentContainerUpdater)
+ extends GroupConverter {
/**
- * The parent converter
+ * Updater used together with [[CatalystRowConverter]].
+ *
+ * @constructor Constructs a [[RowUpdater]] which sets converted filed values to the `ordinal`-th
+ * cell in `row`.
*/
- protected[parquet] val parent: CatalystConverter
+ private final class RowUpdater(row: MutableRow, ordinal: Int) extends ParentContainerUpdater {
+ override def set(value: Any): Unit = row(ordinal) = value
+ override def setBoolean(value: Boolean): Unit = row.setBoolean(ordinal, value)
+ override def setByte(value: Byte): Unit = row.setByte(ordinal, value)
+ override def setShort(value: Short): Unit = row.setShort(ordinal, value)
+ override def setInt(value: Int): Unit = row.setInt(ordinal, value)
+ override def setLong(value: Long): Unit = row.setLong(ordinal, value)
+ override def setDouble(value: Double): Unit = row.setDouble(ordinal, value)
+ override def setFloat(value: Float): Unit = row.setFloat(ordinal, value)
+ }
/**
- * Called by child converters to update their value in its parent (this).
- * Note that if possible the more specific update methods below should be used
- * to avoid auto-boxing of native JVM types.
- *
- * @param fieldIndex
- * @param value
+ * Represents the converted row object once an entire Parquet record is converted.
*/
- protected[parquet] def updateField(fieldIndex: Int, value: Any): Unit
+ val currentRow = new SpecificMutableRow(sparkType.map(_.dataType))
- protected[parquet] def updateBoolean(fieldIndex: Int, value: Boolean): Unit =
- updateField(fieldIndex, value)
+ // Converters for each field.
+ private val fieldConverters: Array[Converter] = {
+ parquetType.getFields.zip(sparkType).zipWithIndex.map {
+ case ((parquetFieldType, sparkField), ordinal) =>
+ newConverter(parquetFieldType, sparkField.dataType, new RowUpdater(currentRow, ordinal))
+ }.toArray
+ }
- protected[parquet] def updateInt(fieldIndex: Int, value: Int): Unit =
- updateField(fieldIndex, value)
+ override def getConverter(fieldIndex: Int): Converter = fieldConverters(fieldIndex)
- protected[parquet] def updateDate(fieldIndex: Int, value: Int): Unit =
- updateField(fieldIndex, value)
+ override def end(): Unit = updater.set(currentRow)
- protected[parquet] def updateLong(fieldIndex: Int, value: Long): Unit =
- updateField(fieldIndex, value)
+ override def start(): Unit = {
+ var i = 0
+ while (i < currentRow.length) {
+ currentRow.setNullAt(i)
+ i += 1
+ }
+ }
- protected[parquet] def updateShort(fieldIndex: Int, value: Short): Unit =
- updateField(fieldIndex, value)
+ /**
+ * Creates a converter for the given Parquet type `parquetType` and Spark SQL data type
+ * `sparkType`. Converted values are handled by `updater`.
+ */
+ private def newConverter(
+ parquetType: Type,
+ sparkType: DataType,
+ updater: ParentContainerUpdater): Converter = {
- protected[parquet] def updateByte(fieldIndex: Int, value: Byte): Unit =
- updateField(fieldIndex, value)
+ sparkType match {
+ case BooleanType | IntegerType | LongType | FloatType | DoubleType | BinaryType =>
+ new CatalystPrimitiveConverter(updater)
- protected[parquet] def updateDouble(fieldIndex: Int, value: Double): Unit =
- updateField(fieldIndex, value)
+ case ByteType =>
+ new PrimitiveConverter {
+ override def addInt(value: Int): Unit =
+ updater.setByte(value.asInstanceOf[ByteType#JvmType])
+ }
- protected[parquet] def updateFloat(fieldIndex: Int, value: Float): Unit =
- updateField(fieldIndex, value)
+ case ShortType =>
+ new PrimitiveConverter {
+ override def addInt(value: Int): Unit =
+ updater.setShort(value.asInstanceOf[ShortType#JvmType])
+ }
- protected[parquet] def updateBinary(fieldIndex: Int, value: Binary): Unit =
- updateField(fieldIndex, value.getBytes)
+ case t: DecimalType =>
+ new CatalystDecimalConverter(t, updater)
- protected[parquet] def updateString(fieldIndex: Int, value: String): Unit =
- updateField(fieldIndex, value)
+ case StringType =>
+ new CatalystStringConverter(updater)
- protected[parquet] def updateTimestamp(fieldIndex: Int, value: Binary): Unit =
- updateField(fieldIndex, readTimestamp(value))
+ case TimestampType =>
+ new PrimitiveConverter {
+ override def addBinary(value: Binary): Unit = {
+ updater.set(CatalystTimestampConverter.convertToTimestamp(value))
+ }
+ }
- protected[parquet] def updateDecimal(fieldIndex: Int, value: Binary, ctype: DecimalType): Unit =
- updateField(fieldIndex, readDecimal(new Decimal(), value, ctype))
+ case DateType =>
+ new PrimitiveConverter {
+ override def addInt(value: Int): Unit = {
+ updater.setInt(value.asInstanceOf[DateType#JvmType])
+ }
+ }
- protected[parquet] def isRootConverter: Boolean = parent == null
+ case t: ArrayType =>
+ new CatalystArrayConverter(parquetType.asGroupType(), t, updater)
- protected[parquet] def clearBuffer(): Unit
+ case t: MapType =>
+ new CatalystMapConverter(parquetType.asGroupType(), t, updater)
- /**
- * Should only be called in the root (group) converter!
- *
- * @return
- */
- def getCurrentRecord: Row = throw new UnsupportedOperationException
+ case t: StructType =>
+ new CatalystRowConverter(parquetType.asGroupType(), t, updater)
- /**
- * Read a decimal value from a Parquet Binary into "dest". Only supports decimals that fit in
- * a long (i.e. precision <= 18)
- */
- protected[parquet] def readDecimal(dest: Decimal, value: Binary, ctype: DecimalType): Unit = {
- val precision = ctype.precisionInfo.get.precision
- val scale = ctype.precisionInfo.get.scale
- val bytes = value.getBytes
- require(bytes.length <= 16, "Decimal field too large to read")
- var unscaled = 0L
- var i = 0
- while (i < bytes.length) {
- unscaled = (unscaled << 8) | (bytes(i) & 0xFF)
- i += 1
+ case t: UserDefinedType[_] =>
+ val sparkTypeForUDT = t.sqlType
+ val parquetTypeForUDT = ParquetTypesConverter.fromDataType(sparkTypeForUDT, "<placeholder>")
+ newConverter(parquetTypeForUDT, sparkTypeForUDT, updater)
+
+ case _ =>
+ throw new RuntimeException(
+ s"Unable to create Parquet converter for data type ${sparkType.json}")
}
- // Make sure unscaled has the right sign, by sign-extending the first bit
- val numBits = 8 * bytes.length
- unscaled = (unscaled << (64 - numBits)) >> (64 - numBits)
- dest.set(unscaled, precision, scale)
}
/**
- * Read a Timestamp value from a Parquet Int96Value
+ * Parquet converter for Parquet primitive types. Note that not all Spark SQL primitive types
+ * are handled by this converter. Parquet primitive types are only a subset of those of Spark
+ * SQL. For example, BYTE, SHORT and INT in Spark SQL are all covered by INT32 in Parquet.
*/
- protected[parquet] def readTimestamp(value: Binary): Timestamp = {
- CatalystTimestampConverter.convertToTimestamp(value)
- }
-}
+ private final class CatalystPrimitiveConverter(updater: ParentContainerUpdater)
+ extends PrimitiveConverter {
-/**
- * A `parquet.io.api.GroupConverter` that is able to convert a Parquet record
- * to a [[org.apache.spark.sql.catalyst.expressions.Row]] object.
- *
- * @param schema The corresponding Catalyst schema in the form of a list of attributes.
- */
-private[parquet] class CatalystGroupConverter(
- protected[parquet] val schema: Array[FieldType],
- protected[parquet] val index: Int,
- protected[parquet] val parent: CatalystConverter,
- protected[parquet] var current: ArrayBuffer[Any],
- protected[parquet] var buffer: ArrayBuffer[Row])
- extends CatalystConverter {
-
- def this(schema: Array[FieldType], index: Int, parent: CatalystConverter) =
- this(
- schema,
- index,
- parent,
- current = null,
- buffer = new ArrayBuffer[Row](
- CatalystArrayConverter.INITIAL_ARRAY_SIZE))
+ override def addBoolean(value: Boolean): Unit = updater.setBoolean(value)
+ override def addInt(value: Int): Unit = updater.setInt(value)
+ override def addLong(value: Long): Unit = updater.setLong(value)
+ override def addFloat(value: Float): Unit = updater.setFloat(value)
+ override def addDouble(value: Double): Unit = updater.setDouble(value)
+ override def addBinary(value: Binary): Unit = updater.set(value.getBytes)
+ }
/**
- * This constructor is used for the root converter only!
+ * Parquet converter for strings. A dictionary is used to avoid minimize string decoding cost.
*/
- def this(attributes: Array[Attribute]) =
- this(attributes.map(a => new FieldType(a.name, a.dataType, a.nullable)), 0, null)
-
- protected [parquet] val converters: Array[Converter] =
- schema.zipWithIndex.map {
- case (field, idx) => CatalystConverter.createConverter(field, idx, this)
- }.toArray
-
- override val size = schema.size
+ private final class CatalystStringConverter(updater: ParentContainerUpdater)
+ extends PrimitiveConverter {
- override def getCurrentRecord: Row = {
- assert(isRootConverter, "getCurrentRecord should only be called in root group converter!")
- // TODO: use iterators if possible
- // Note: this will ever only be called in the root converter when the record has been
- // fully processed. Therefore it will be difficult to use mutable rows instead, since
- // any non-root converter never would be sure when it would be safe to re-use the buffer.
- new GenericRow(current.toArray)
- }
-
- override def getConverter(fieldIndex: Int): Converter = converters(fieldIndex)
-
- // for child converters to update upstream values
- override protected[parquet] def updateField(fieldIndex: Int, value: Any): Unit = {
- current.update(fieldIndex, value)
- }
+ private var expandedDictionary: Array[String] = null
- override protected[parquet] def clearBuffer(): Unit = buffer.clear()
+ override def hasDictionarySupport: Boolean = true
- override def start(): Unit = {
- current = ArrayBuffer.fill(size)(null)
- converters.foreach {
- converter => if (!converter.isPrimitive) {
- converter.asInstanceOf[CatalystConverter].clearBuffer
+ override def setDictionary(dictionary: Dictionary): Unit = {
+ this.expandedDictionary = Array.tabulate(dictionary.getMaxId + 1) {
+ dictionary.decodeToBinary(_).toStringUsingUTF8
}
}
- }
- override def end(): Unit = {
- if (!isRootConverter) {
- assert(current != null) // there should be no empty groups
- buffer.append(new GenericRow(current.toArray))
- parent.updateField(index, new GenericRow(buffer.toArray.asInstanceOf[Array[Any]]))
+ override def addValueFromDictionary(dictionaryId: Int): Unit = {
+ updater.set(expandedDictionary(dictionaryId))
}
- }
-}
-/**
- * A `parquet.io.api.GroupConverter` that is able to convert a Parquet record
- * to a [[org.apache.spark.sql.catalyst.expressions.Row]] object. Note that his
- * converter is optimized for rows of primitive types (non-nested records).
- */
-private[parquet] class CatalystPrimitiveRowConverter(
- protected[parquet] val schema: Array[FieldType],
- protected[parquet] var current: MutableRow)
- extends CatalystConverter {
-
- // This constructor is used for the root converter only
- def this(attributes: Array[Attribute]) =
- this(
- attributes.map(a => new FieldType(a.name, a.dataType, a.nullable)),
- new SpecificMutableRow(attributes.map(_.dataType)))
-
- protected [parquet] val converters: Array[Converter] =
- schema.zipWithIndex.map {
- case (field, idx) => CatalystConverter.createConverter(field, idx, this)
- }.toArray
-
- override val size = schema.size
+ override def addBinary(value: Binary): Unit = updater.set(value.toStringUsingUTF8)
+ }
- override val index = 0
+ /**
+ * Parquet converter for fixed-precision decimals.
+ *
+ * @todo Handle fixed-precision decimals stored as INT32 and INT64
+ */
+ private final class CatalystDecimalConverter(
+ decimalType: DecimalType,
+ updater: ParentContainerUpdater)
+ extends PrimitiveConverter {
- override val parent = null
+ override def addBinary(value: Binary): Unit = {
+ updater.set(toDecimal(value))
+ }
- // Should be only called in root group converter!
- override def getCurrentRecord: Row = current
+ private def toDecimal(value: Binary): Decimal = {
+ val precision = decimalType.precision
+ val scale = decimalType.scale
+ val bytes = value.getBytes
- override def getConverter(fieldIndex: Int): Converter = converters(fieldIndex)
+ require(bytes.length <= 16, "Decimal field too large to read")
- // for child converters to update upstream values
- override protected[parquet] def updateField(fieldIndex: Int, value: Any): Unit = {
- throw new UnsupportedOperationException // child converters should use the
- // specific update methods below
- }
+ var unscaled = 0L
+ var i = 0
- override protected[parquet] def clearBuffer(): Unit = {}
+ while (i < bytes.length) {
+ unscaled = (unscaled << 8) | (bytes(i) & 0xff)
+ i += 1
+ }
- override def start(): Unit = {
- var i = 0
- while (i < size) {
- current.setNullAt(i)
- i = i + 1
+ val bits = 8 * bytes.length
+ unscaled = (unscaled << (64 - bits)) >> (64 - bits)
+ Decimal(unscaled, precision, scale)
}
}
- override def end(): Unit = {}
-
- // Overridden here to avoid auto-boxing for primitive types
- override protected[parquet] def updateBoolean(fieldIndex: Int, value: Boolean): Unit =
- current.setBoolean(fieldIndex, value)
-
- override protected[parquet] def updateInt(fieldIndex: Int, value: Int): Unit =
- current.setInt(fieldIndex, value)
+ /**
+ * Parquet converter for arrays. Spark SQL arrays are represented as Parquet lists. Standard
+ * Parquet lists are represented as a 3-level group annotated by `LIST`:
+ * {{{
+ * <list-repetition> group <name> (LIST) { <-- parquetSchema points here
+ * repeated group list {
+ * <element-repetition> <element-type> element;
+ * }
+ * }
+ * }}}
+ * The `parquetSchema` constructor argument points to the outermost group.
+ *
+ * However, before this representation is standardized, some Parquet libraries/tools also use some
+ * non-standard formats to represent list-like structures. Backwards-compatibility rules for
+ * handling these cases are described in Parquet format spec.
+ *
+ * @see https://github.com/apache/incubator-parquet-format/blob/master/LogicalTypes.md#lists
+ */
+ private final class CatalystArrayConverter(
+ parquetSchema: GroupType,
+ sparkSchema: ArrayType,
+ updater: ParentContainerUpdater)
+ extends GroupConverter {
+
+ // TODO This is slow! Needs specialization.
+ private val currentArray = ArrayBuffer.empty[Any]
+
+ private val elementConverter: Converter = {
+ val repeatedType = parquetSchema.getType(0)
+ val elementType = sparkSchema.elementType
+
+ if (isElementType(repeatedType, elementType)) {
+ newConverter(repeatedType, elementType, new ParentContainerUpdater {
+ override def set(value: Any): Unit = currentArray += value
+ })
+ } else {
+ new ElementConverter(repeatedType.asGroupType().getType(0), elementType)
+ }
+ }
- override protected[parquet] def updateDate(fieldIndex: Int, value: Int): Unit =
- current.update(fieldIndex, value)
+ override def getConverter(fieldIndex: Int): Converter = elementConverter
+
+ override def end(): Unit = updater.set(currentArray)
+
+ override def start(): Unit = currentArray.clear()
+
+ // scalastyle:off
+ /**
+ * Returns whether the given type is the element type of a list or is a syntactic group with
+ * one field that is the element type. This is determined by checking whether the type can be
+ * a syntactic group and by checking whether a potential syntactic group matches the expected
+ * schema.
+ * {{{
+ * <list-repetition> group <name> (LIST) {
+ * repeated group list { <-- repeatedType points here
+ * <element-repetition> <element-type> element;
+ * }
+ * }
+ * }}}
+ * In short, here we handle Parquet list backwards-compatibility rules on the read path. This
+ * method is based on `AvroIndexedRecordConverter.isElementType`.
+ *
+ * @see https://github.com/apache/incubator-parquet-format/blob/master/LogicalTypes.md#backward-compatibility-rules
+ */
+ // scalastyle:on
+ private def isElementType(repeatedType: Type, elementType: DataType): Boolean = {
+ (repeatedType, elementType) match {
+ case (t: ParquetPrimitiveType, _) => true
+ case (t: GroupType, _) if t.getFieldCount > 1 => true
+ case (t: GroupType, StructType(Array(f))) if f.name == t.getFieldName(0) => true
+ case _ => false
+ }
+ }
- override protected[parquet] def updateLong(fieldIndex: Int, value: Long): Unit =
- current.setLong(fieldIndex, value)
+ /** Array element converter */
+ private final class ElementConverter(parquetType: Type, sparkType: DataType)
+ extends GroupConverter {
- override protected[parquet] def updateShort(fieldIndex: Int, value: Short): Unit =
- current.setShort(fieldIndex, value)
+ private var currentElement: Any = _
- override protected[parquet] def updateByte(fieldIndex: Int, value: Byte): Unit =
- current.setByte(fieldIndex, value)
+ private val converter = newConverter(parquetType, sparkType, new ParentContainerUpdater {
+ override def set(value: Any): Unit = currentElement = value
+ })
- override protected[parquet] def updateDouble(fieldIndex: Int, value: Double): Unit =
- current.setDouble(fieldIndex, value)
+ override def getConverter(fieldIndex: Int): Converter = converter
- override protected[parquet] def updateFloat(fieldIndex: Int, value: Float): Unit =
- current.setFloat(fieldIndex, value)
+ override def end(): Unit = currentArray += currentElement
- override protected[parquet] def updateBinary(fieldIndex: Int, value: Binary): Unit =
- current.update(fieldIndex, value.getBytes)
+ override def start(): Unit = currentElement = null
+ }
+ }
- override protected[parquet] def updateString(fieldIndex: Int, value: String): Unit =
- current.setString(fieldIndex, value)
+ /** Parquet converter for maps */
+ private final class CatalystMapConverter(
+ parquetType: GroupType,
+ sparkType: MapType,
+ updater: ParentContainerUpdater)
+ extends GroupConverter {
- override protected[parquet] def updateTimestamp(fieldIndex: Int, value: Binary): Unit =
- current.update(fieldIndex, readTimestamp(value))
+ private val currentMap = mutable.Map.empty[Any, Any]
- override protected[parquet] def updateDecimal(
- fieldIndex: Int, value: Binary, ctype: DecimalType): Unit = {
- var decimal = current(fieldIndex).asInstanceOf[Decimal]
- if (decimal == null) {
- decimal = new Decimal
- current(fieldIndex) = decimal
+ private val keyValueConverter = {
+ val repeatedType = parquetType.getType(0).asGroupType()
+ new KeyValueConverter(
+ repeatedType.getType(0),
+ repeatedType.getType(1),
+ sparkType.keyType,
+ sparkType.valueType)
}
- readDecimal(decimal, value, ctype)
- }
-}
-/**
- * A `parquet.io.api.PrimitiveConverter` that converts Parquet types to Catalyst types.
- *
- * @param parent The parent group converter.
- * @param fieldIndex The index inside the record.
- */
-private[parquet] class CatalystPrimitiveConverter(
- parent: CatalystConverter,
- fieldIndex: Int) extends PrimitiveConverter {
- override def addBinary(value: Binary): Unit =
- parent.updateBinary(fieldIndex, value)
+ override def getConverter(fieldIndex: Int): Converter = keyValueConverter
- override def addBoolean(value: Boolean): Unit =
- parent.updateBoolean(fieldIndex, value)
+ override def end(): Unit = updater.set(currentMap)
- override def addDouble(value: Double): Unit =
- parent.updateDouble(fieldIndex, value)
+ override def start(): Unit = currentMap.clear()
- override def addFloat(value: Float): Unit =
- parent.updateFloat(fieldIndex, value)
+ /** Parquet converter for key-value pairs within the map. */
+ private final class KeyValueConverter(
+ parquetKeyType: Type,
+ parquetValueType: Type,
+ sparkKeyType: DataType,
+ sparkValueType: DataType)
+ extends GroupConverter {
- override def addInt(value: Int): Unit =
- parent.updateInt(fieldIndex, value)
+ private var currentKey: Any = _
- override def addLong(value: Long): Unit =
- parent.updateLong(fieldIndex, value)
-}
+ private var currentValue: Any = _
-/**
- * A `parquet.io.api.PrimitiveConverter` that converts Parquet Binary to Catalyst String.
- * Supports dictionaries to reduce Binary to String conversion overhead.
- *
- * Follows pattern in Parquet of using dictionaries, where supported, for String conversion.
- *
- * @param parent The parent group converter.
- * @param fieldIndex The index inside the record.
- */
-private[parquet] class CatalystPrimitiveStringConverter(parent: CatalystConverter, fieldIndex: Int)
- extends CatalystPrimitiveConverter(parent, fieldIndex) {
+ private val converters = Array(
+ // Converter for keys
+ newConverter(parquetKeyType, sparkKeyType, new ParentContainerUpdater {
+ override def set(value: Any): Unit = currentKey = value
+ }),
- private[this] var dict: Array[String] = null
+ // Converter for values
+ newConverter(parquetValueType, sparkValueType, new ParentContainerUpdater {
+ override def set(value: Any): Unit = currentValue = value
+ }))
- override def hasDictionarySupport: Boolean = true
+ override def getConverter(fieldIndex: Int): Converter = converters(fieldIndex)
- override def setDictionary(dictionary: Dictionary):Unit =
- dict = Array.tabulate(dictionary.getMaxId + 1) {dictionary.decodeToBinary(_).toStringUsingUTF8}
+ override def end(): Unit = currentMap(currentKey) = currentValue
+ override def start(): Unit = {
+ currentKey = null
+ currentValue = null
+ }
+ }
+ }
+}
- override def addValueFromDictionary(dictionaryId: Int): Unit =
- parent.updateString(fieldIndex, dict(dictionaryId))
+private[sql] object CatalystConverter {
+ // The type internally used for fields
+ type FieldType = StructField
- override def addBinary(value: Binary): Unit =
- parent.updateString(fieldIndex, value.toStringUsingUTF8)
-}
+ // This is mostly Parquet convention (see, e.g., `ConversionPatterns`).
+ // Note that "array" for the array elements is chosen by ParquetAvro.
+ // Using a different value will result in Parquet silently dropping columns.
+ val ARRAY_CONTAINS_NULL_BAG_SCHEMA_NAME = "bag"
+ val ARRAY_ELEMENTS_SCHEMA_NAME = "array"
+ // SPARK-4520: Thrift generated parquet files have different array element
+ // schema names than avro. Thrift parquet uses array_schema_name + "_tuple"
+ // as opposed to "array" used by default. For more information, check
+ // TestThriftSchemaConverter.java in parquet.thrift.
+ val THRIFT_ARRAY_ELEMENTS_SCHEMA_NAME_SUFFIX = "_tuple"
+ val MAP_KEY_SCHEMA_NAME = "key"
+ val MAP_VALUE_SCHEMA_NAME = "value"
+ val MAP_SCHEMA_NAME = "map"
-private[parquet] object CatalystArrayConverter {
- val INITIAL_ARRAY_SIZE = 20
+ // TODO: consider using Array[T] for arrays to avoid boxing of primitive types
+ type ArrayScalaType[T] = Seq[T]
+ type StructScalaType[T] = Row
+ type MapScalaType[K, V] = Map[K, V]
--- End diff --
These values and types are from the old converter code. They are preserved here because they're also used elsewhere.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90919013
[Test build #29858 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29858/consoleFull) for PR 5422 at commit [`0de2e8b`](https://github.com/apache/spark/commit/0de2e8b83fb4b616603c2013792db3351c82f324).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
* This patch does not change any dependencies.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90924913
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/29859/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90980809
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/29868/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90919032
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/29858/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90980793
[Test build #29868 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29868/consoleFull) for PR 5422 at commit [`02d78c4`](https://github.com/apache/spark/commit/02d78c47801998af1fc9c76cc400acfd57ccbe33).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
* This patch does not change any dependencies.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90924894
[Test build #29859 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29859/consoleFull) for PR 5422 at commit [`9522949`](https://github.com/apache/spark/commit/95229493ec2d3a69762f4dd4fb707a87911980b3).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
* This patch does not change any dependencies.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90977850
[Test build #29869 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29869/consoleFull) for PR 5422 at commit [`2529b76`](https://github.com/apache/spark/commit/2529b76141c068fe69a03e29ddc50ca496cd9dcd).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90913336
[Test build #29858 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29858/consoleFull) for PR 5422 at commit [`0de2e8b`](https://github.com/apache/spark/commit/0de2e8b83fb4b616603c2013792db3351c82f324).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90993598
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/29869/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/5422#issuecomment-90918532
[Test build #29859 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/29859/consoleFull) for PR 5422 at commit [`9522949`](https://github.com/apache/spark/commit/95229493ec2d3a69762f4dd4fb707a87911980b3).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6775] [SPARK-6776] [SQL] [WIP] Refactor...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5422#discussion_r27968497
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificMutableRow.scala ---
@@ -202,6 +202,7 @@ final class SpecificMutableRow(val values: Array[MutableValue]) extends MutableR
case DoubleType => new MutableDouble
case BooleanType => new MutableBoolean
case LongType => new MutableLong
+ case DateType => new MutableInt
--- End diff --
Dates are represented as integers inside Catalyst. Since `SpecificMutableRow` is only used inside Catalyst, it should be safe to specialize `DateType` here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org