You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by marmbrus <gi...@git.apache.org> on 2016/03/15 03:44:51 UTC
[GitHub] spark pull request: [WIP][SPARK-13883][SQL] Parquet Implementation...
GitHub user marmbrus opened a pull request:
https://github.com/apache/spark/pull/11709
[WIP][SPARK-13883][SQL] Parquet Implementation of FileFormat.buildReader
TODO...
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/marmbrus/spark parquetReader
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11709.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 #11709
----
commit aad7bd13495a4e90439da73982a9c8483994086f
Author: Michael Armbrust <mi...@databricks.com>
Date: 2016-03-15T02:41:53Z
WIP
----
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197585281
**[Test build #53367 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53367/consoleFull)** for PR 11709 at commit [`8ccdd77`](https://github.com/apache/spark/commit/8ccdd77fb1bdb63a589a214e6151d81b54eeb524).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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-13883][SQL] Parquet Implementation of F...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r57330560
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -269,6 +276,137 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ assert(file.partitionValues.numFields == partitionSchema.size)
+
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!enableVectorizedParquetReader) sys.error("Vectorized reader turned off.")
+ val vectorizedReader = new VectorizedParquetRecordReader()
+ vectorizedReader.initialize(split, hadoopAttemptContext)
+ logDebug(s"Appending $partitionSchema ${file.partitionValues}")
+ vectorizedReader.initBatch(partitionSchema, file.partitionValues)
--- End diff --
I was wondering why we need to pass partition schema and values to `buildReader` since partitioning should have already been handled during planning phase. Then I found they are only used here. Seems that this is pretty much a Parquet vectorized reader specific use case.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197583103
**[Test build #53367 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53367/consoleFull)** for PR 11709 at commit [`8ccdd77`](https://github.com/apache/spark/commit/8ccdd77fb1bdb63a589a214e6151d81b54eeb524).
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197585293
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53367/
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198542504
**[Test build #53565 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53565/consoleFull)** for PR 11709 at commit [`b0cd621`](https://github.com/apache/spark/commit/b0cd621e87970aa831bc877590ef011c86669a7c).
---
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-13883][SQL] Parquet Implementation of F...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56550417
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,141 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+
+ // UnsafeRowParquetRecordReader appends the columns internally to avoid another copy.
+ if (parquetReader.isInstanceOf[UnsafeRowParquetRecordReader] &&
+ enableVectorizedParquetReader) {
+ iter.asInstanceOf[Iterator[InternalRow]]
+ } else {
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
--- End diff --
Is this true that the partitioned value always come before others?
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197134624
**[Test build #53262 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53262/consoleFull)** for PR 11709 at commit [`122f572`](https://github.com/apache/spark/commit/122f572a9bfb7ebdda66fae0fd70d92c9d4d6524).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56428091
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,140 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+
+ // UnsafeRowParquetRecordReader appends the columns internally to avoid another copy.
+ if (parquetReader.isInstanceOf[UnsafeRowParquetRecordReader]) {
+ iter.asInstanceOf[Iterator[InternalRow]]
+ } else {
+ // This is a horrible erasure hack... if we type the iterator above, then it actually check
+ // the type in next() and we get a class cast exception. If we make that function return
+ // Object, then we can defer the cast until later!
+ iter.asInstanceOf[Iterator[InternalRow]]
+ .map(d => appendPartitionColumns(joinedRow(d, file.partitionValues)))
+ .map(r => r.copy())
--- End diff --
May could put a TODO here to fix other bugs.
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198590824
Merged build finished. 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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-199516605
**[Test build #53719 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53719/consoleFull)** for PR 11709 at commit [`2e21c7a`](https://github.com/apache/spark/commit/2e21c7add2bd6c7d5488ba53bfa75da3b5050132).
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197589892
Merged build finished. 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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196987147
**[Test build #53212 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53212/consoleFull)** for PR 11709 at commit [`d3a39e7`](https://github.com/apache/spark/commit/d3a39e755ab8e5fa958f35dddffaff972ec05967).
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197592132
**[Test build #53372 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53372/consoleFull)** for PR 11709 at commit [`ed60f3d`](https://github.com/apache/spark/commit/ed60f3d1c19bd33286d413472e8bc00a571a3064).
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198562203
**[Test build #53571 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53571/consoleFull)** for PR 11709 at commit [`40037e6`](https://github.com/apache/spark/commit/40037e6a49721332b3fad2a70c6c8dfae65f68e4).
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197069799
Merged build finished. 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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197134877
Merged build finished. 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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198553825
**[Test build #53558 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53558/consoleFull)** for PR 11709 at commit [`c7ca5fe`](https://github.com/apache/spark/commit/c7ca5fe5004b42cc61482b77ae8af1e04b2bee35).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197589893
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53369/
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56108838
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -57,7 +57,8 @@ import org.apache.spark.sql.types._
private[sql] object FileSourceStrategy extends Strategy with Logging {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case PhysicalOperation(projects, filters, l@LogicalRelation(files: HadoopFsRelation, _, _))
- if files.fileFormat.toString == "TestFileFormat" =>
+ if files.fileFormat.toString == "TestFileFormat" ||
+ files.fileFormat.toString == "ParquetFormat" =>
--- End diff --
Why not use `isInstanceOf`? `toString` looks fragile to me...
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196626022
**[Test build #53150 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53150/consoleFull)** for PR 11709 at commit [`aad7bd1`](https://github.com/apache/spark/commit/aad7bd13495a4e90439da73982a9c8483994086f).
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197614696
**[Test build #53372 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53372/consoleFull)** for PR 11709 at commit [`ed60f3d`](https://github.com/apache/spark/commit/ed60f3d1c19bd33286d413472e8bc00a571a3064).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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-13883][SQL] Parquet Implementation of F...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r57362838
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -269,6 +276,137 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ assert(file.partitionValues.numFields == partitionSchema.size)
+
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!enableVectorizedParquetReader) sys.error("Vectorized reader turned off.")
+ val vectorizedReader = new VectorizedParquetRecordReader()
+ vectorizedReader.initialize(split, hadoopAttemptContext)
+ logDebug(s"Appending $partitionSchema ${file.partitionValues}")
+ vectorizedReader.initBatch(partitionSchema, file.partitionValues)
--- End diff --
Yeah, this perhaps belongs at a higher layer (probably in FileScanRDD), but that would require us to vectorize everything or take a giant performance hit. This line is telling the vectorized reader to append the partition columns as static columns. This allows us to avoid an extra copy to append them for the optimized path.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56108937
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -57,7 +57,8 @@ import org.apache.spark.sql.types._
private[sql] object FileSourceStrategy extends Strategy with Logging {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case PhysicalOperation(projects, filters, l@LogicalRelation(files: HadoopFsRelation, _, _))
- if files.fileFormat.toString == "TestFileFormat" =>
+ if files.fileFormat.toString == "TestFileFormat" ||
+ files.fileFormat.toString == "ParquetFormat" =>
--- End diff --
ah got it, thanks!
---
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-13883][SQL] Parquet Implementation of F...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/11709
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197058845
Merged build finished. 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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197054035
**[Test build #53237 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53237/consoleFull)** for PR 11709 at commit [`9b720cd`](https://github.com/apache/spark/commit/9b720cdd3dac3b4d6dd6d86201e9fd65566f1767).
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198569712
**[Test build #53575 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53575/consoleFull)** for PR 11709 at commit [`e5725f8`](https://github.com/apache/spark/commit/e5725f84e46b167d7c4b044c7c84e631cbc061bc).
---
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-13883][SQL] Parquet Implementation of F...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56924139
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,141 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+
+ // UnsafeRowParquetRecordReader appends the columns internally to avoid another copy.
+ if (parquetReader.isInstanceOf[UnsafeRowParquetRecordReader] &&
+ enableVectorizedParquetReader) {
+ iter.asInstanceOf[Iterator[InternalRow]]
+ } else {
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
--- End diff --
Yeah, this is enforced by the planner.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197043761
**[Test build #53229 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53229/consoleFull)** for PR 11709 at commit [`c9229a9`](https://github.com/apache/spark/commit/c9229a9fa3a133811ba4b4adfc378cddfba2710c).
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-199550198
Merged build finished. Test PASSed.
---
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-13883][SQL] Parquet Implementation of F...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56597128
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -807,6 +809,11 @@ public final int appendStruct(boolean isNull) {
public final boolean isArray() { return resultArray != null; }
/**
+ * Marks this column as being constant.
+ */
+ public final void setIsConstant() { isConstant = true; }
--- End diff --
I think that's a good suggestion. Let's do it as a follow up though.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197585289
Merged build finished. 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-13883][SQL] Parquet Implementation of F...
Posted by tedyu <gi...@git.apache.org>.
Github user tedyu commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r57014281
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java ---
@@ -227,9 +276,19 @@ private void initializeInternal() throws IOException {
if (primitiveType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.INT96) {
throw new IOException("Int96 not supported.");
}
- ColumnDescriptor fd = fileSchema.getColumnDescription(requestedSchema.getPaths().get(i));
- if (!fd.equals(requestedSchema.getColumns().get(i))) {
- throw new IOException("Schema evolution not supported.");
+ String[] colPath = requestedSchema.getPaths().get(i);
+ if (fileSchema.containsPath(colPath)) {
+ ColumnDescriptor fd = fileSchema.getColumnDescription(colPath);
+ if (!fd.equals(requestedSchema.getColumns().get(i))) {
+ throw new IOException("Schema evolution not supported.");
--- End diff --
It would be helpful to include fd in the exception message
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-199550199
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53719/
Test PASSed.
---
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-13883][SQL] Parquet Implementation of F...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56551050
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -57,7 +57,8 @@ import org.apache.spark.sql.types._
private[sql] object FileSourceStrategy extends Strategy with Logging {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case PhysicalOperation(projects, filters, l@LogicalRelation(files: HadoopFsRelation, _, _))
- if files.fileFormat.toString == "TestFileFormat" =>
+ if files.fileFormat.toString == "TestFileFormat" ||
+ files.fileFormat.isInstanceOf[parquet.DefaultSource] =>
--- End diff --
Should we have a feature flag for this? Or user will saw regression if we didn't support locality when release 2.0.
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198563037
**[Test build #53571 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53571/consoleFull)** for PR 11709 at commit [`40037e6`](https://github.com/apache/spark/commit/40037e6a49721332b3fad2a70c6c8dfae65f68e4).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56427625
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,140 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+
+ // UnsafeRowParquetRecordReader appends the columns internally to avoid another copy.
+ if (parquetReader.isInstanceOf[UnsafeRowParquetRecordReader]) {
+ iter.asInstanceOf[Iterator[InternalRow]]
+ } else {
+ // This is a horrible erasure hack... if we type the iterator above, then it actually check
+ // the type in next() and we get a class cast exception. If we make that function return
+ // Object, then we can defer the cast until later!
+ iter.asInstanceOf[Iterator[InternalRow]]
+ .map(d => appendPartitionColumns(joinedRow(d, file.partitionValues)))
+ .map(r => r.copy())
--- End diff --
Is this copy necessary?
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197058750
**[Test build #53229 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53229/consoleFull)** for PR 11709 at commit [`c9229a9`](https://github.com/apache/spark/commit/c9229a9fa3a133811ba4b4adfc378cddfba2710c).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class RecordReaderIterator[T](rowReader: RecordReader[_, T]) extends Iterator[T] `
* ` // the type in next() and we get a class cast exception. If we make that function return`
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196987533
**[Test build #53212 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53212/consoleFull)** for PR 11709 at commit [`d3a39e7`](https://github.com/apache/spark/commit/d3a39e755ab8e5fa958f35dddffaff972ec05967).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class RecordReaderIterator[T](rowReader: RecordReader[_, T]) extends Iterator[T] `
* ` // the type in next() and we get a class cast exception. If we make that function return`
---
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-13883][SQL] Parquet Implementation of F...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-199577463
LGTM
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198519583
**[Test build #53558 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53558/consoleFull)** for PR 11709 at commit [`c7ca5fe`](https://github.com/apache/spark/commit/c7ca5fe5004b42cc61482b77ae8af1e04b2bee35).
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198590825
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53575/
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196626176
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53150/
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-13883][SQL] Parquet Implementation of F...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r57363361
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -269,6 +276,137 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ assert(file.partitionValues.numFields == partitionSchema.size)
+
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!enableVectorizedParquetReader) sys.error("Vectorized reader turned off.")
+ val vectorizedReader = new VectorizedParquetRecordReader()
+ vectorizedReader.initialize(split, hadoopAttemptContext)
+ logDebug(s"Appending $partitionSchema ${file.partitionValues}")
+ vectorizedReader.initBatch(partitionSchema, file.partitionValues)
--- End diff --
Basically, I'd just take the non-vectorized version below, put it in a utility function and use it everywhere. If we vectorize all the sources, that will be the only part we have to remove and then this can be done in FileScanRDD.
I think that you do not want to do the actually partition appending in the planner like we were before, because you can't have Spark Partitions (splits) that read from different partitions very easily. This is what was making the bucking logic so convoluted in the old code path. This makes bucketing and collapsing of small files into a single partition much simpler.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196626174
Merged build finished. 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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197134881
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53262/
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56427859
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,140 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+
+ // UnsafeRowParquetRecordReader appends the columns internally to avoid another copy.
+ if (parquetReader.isInstanceOf[UnsafeRowParquetRecordReader]) {
+ iter.asInstanceOf[Iterator[InternalRow]]
+ } else {
+ // This is a horrible erasure hack... if we type the iterator above, then it actually check
+ // the type in next() and we get a class cast exception. If we make that function return
+ // Object, then we can defer the cast until later!
+ iter.asInstanceOf[Iterator[InternalRow]]
+ .map(d => appendPartitionColumns(joinedRow(d, file.partitionValues)))
+ .map(r => r.copy())
--- End diff --
Yeah, I was surprised by this, but a lot of tests fail without this.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197069692
**[Test build #53237 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53237/consoleFull)** for PR 11709 at commit [`9b720cd`](https://github.com/apache/spark/commit/9b720cdd3dac3b4d6dd6d86201e9fd65566f1767).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* ` // the type in next() and we get a class cast exception. If we make that function return`
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198554168
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53558/
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-13883][SQL] Parquet Implementation of F...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r57422058
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -269,6 +276,137 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ assert(file.partitionValues.numFields == partitionSchema.size)
+
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!enableVectorizedParquetReader) sys.error("Vectorized reader turned off.")
+ val vectorizedReader = new VectorizedParquetRecordReader()
+ vectorizedReader.initialize(split, hadoopAttemptContext)
+ logDebug(s"Appending $partitionSchema ${file.partitionValues}")
+ vectorizedReader.initBatch(partitionSchema, file.partitionValues)
--- End diff --
Sorry for the confusion. When I mentioned "planning phase" what I really meant was that ideally the data source implementation shouldn't care about partitioning at all. But I mixed up partition discovery and partition value appending. I agree with your comments. Thanks for the explanations.
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198590763
**[Test build #53575 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53575/consoleFull)** for PR 11709 at commit [`e5725f8`](https://github.com/apache/spark/commit/e5725f84e46b167d7c4b044c7c84e631cbc061bc).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198563043
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53571/
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197615058
Merged build finished. 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-13883][SQL] Parquet Implementation of F...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56549945
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -807,6 +809,11 @@ public final int appendStruct(boolean isNull) {
public final boolean isArray() { return resultArray != null; }
/**
+ * Marks this column as being constant.
+ */
+ public final void setIsConstant() { isConstant = true; }
--- End diff --
Should we have a special ColumnVector for constants (that always return the same value )?
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197121955
**[Test build #53262 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53262/consoleFull)** for PR 11709 at commit [`122f572`](https://github.com/apache/spark/commit/122f572a9bfb7ebdda66fae0fd70d92c9d4d6524).
---
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-13883][SQL] Parquet Implementation of F...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-199578526
Thanks, merging to master!
---
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-13883][SQL] Parquet Implementation of F...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r57330179
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -269,6 +276,137 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ assert(file.partitionValues.numFields == partitionSchema.size)
+
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!enableVectorizedParquetReader) sys.error("Vectorized reader turned off.")
+ val vectorizedReader = new VectorizedParquetRecordReader()
+ vectorizedReader.initialize(split, hadoopAttemptContext)
+ logDebug(s"Appending $partitionSchema ${file.partitionValues}")
+ vectorizedReader.initBatch(partitionSchema, file.partitionValues)
--- End diff --
@marmbrus Not quite sure about the intention of this line. Are we "reserving" column batches for partition columns here so that partition values can be filled later after data columns are fetched?
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198569180
**[Test build #53565 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53565/consoleFull)** for PR 11709 at commit [`b0cd621`](https://github.com/apache/spark/commit/b0cd621e87970aa831bc877590ef011c86669a7c).
* This patch **fails Spark unit tests**.
* This patch **does not merge cleanly**.
* This patch adds no public classes.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196987547
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53212/
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56427934
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,140 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
--- End diff --
We do, this only happens when we fall back on the old parquet-mr reader. Otherwise columns are appended on like 370. I can more this into the if to make that more clear.
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198554166
Merged build finished. 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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197589886
**[Test build #53369 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53369/consoleFull)** for PR 11709 at commit [`19e455c`](https://github.com/apache/spark/commit/19e455cc69c4f3ae69c2f6fc784c18cce8092c60).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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-13883][SQL] Parquet Implementation of F...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-199549804
**[Test build #53719 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53719/consoleFull)** for PR 11709 at commit [`2e21c7a`](https://github.com/apache/spark/commit/2e21c7add2bd6c7d5488ba53bfa75da3b5050132).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196987542
Merged build finished. 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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56108571
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +273,127 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
--- End diff --
@liancheng any idea why this isn't working?
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56427603
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,140 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
--- End diff --
Can we append the partition values as batches?
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198569293
Build finished. 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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198563042
Merged build finished. 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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56108606
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -57,7 +57,8 @@ import org.apache.spark.sql.types._
private[sql] object FileSourceStrategy extends Strategy with Logging {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case PhysicalOperation(projects, filters, l@LogicalRelation(files: HadoopFsRelation, _, _))
- if files.fileFormat.toString == "TestFileFormat" =>
+ if files.fileFormat.toString == "TestFileFormat" ||
+ files.fileFormat.toString == "ParquetFormat" =>
--- End diff --
@cloud-fan like this
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-196626171
**[Test build #53150 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53150/consoleFull)** for PR 11709 at commit [`aad7bd1`](https://github.com/apache/spark/commit/aad7bd13495a4e90439da73982a9c8483994086f).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class RecordReaderIterator[T](rowReader: RecordReader[_, T]) extends Iterator[T] `
* ` // the type in next() and we get a class cast exception. If we make that function return`
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11709#discussion_r56428259
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala ---
@@ -268,6 +275,140 @@ private[sql] class DefaultSource extends FileFormat with DataSourceRegister with
file.getName == ParquetFileWriter.PARQUET_METADATA_FILE
}
+ /**
+ * Returns a function that can be used to read a single file in as an Iterator of InternalRow.
+ *
+ * @param partitionSchema The schema of the partition column row that will be present in each
+ * PartitionedFile. These columns should be prepended to the rows that
+ * are produced by the iterator.
+ * @param dataSchema The schema of the data that should be output for each row. This may be a
+ * subset of the columns that are present in the file if column pruning has
+ * occurred.
+ * @param filters A set of filters than can optionally be used to reduce the number of rows output
+ * @param options A set of string -> string configuration options.
+ * @return
+ */
+ override def buildReader(
+ sqlContext: SQLContext,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String]): PartitionedFile => Iterator[InternalRow] = {
+ val parquetConf = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ parquetConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, classOf[CatalystReadSupport].getName)
+ parquetConf.set(
+ CatalystReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+ parquetConf.set(
+ CatalystWriteSupport.SPARK_ROW_SCHEMA,
+ CatalystSchemaConverter.checkFieldNames(dataSchema).json)
+
+ // We want to clear this temporary metadata from saving into Parquet file.
+ // This metadata is only useful for detecting optional columns when pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, parquetConf)
+
+ // Sets flags for `CatalystSchemaConverter`
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_BINARY_AS_STRING))
+ parquetConf.setBoolean(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlContext.conf.getConf(SQLConf.PARQUET_INT96_AS_TIMESTAMP))
+
+ // Try to push down filters when filter push-down is enabled.
+ val pushed = if (sqlContext.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key).toBoolean) {
+ filters
+ // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+ // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+ // is used here.
+ .flatMap(ParquetFilters.createFilter(dataSchema, _))
+ .reduceOption(FilterApi.and)
+ } else {
+ None
+ }
+
+ val broadcastedConf =
+ sqlContext.sparkContext.broadcast(new SerializableConfiguration(parquetConf))
+
+ // TODO: if you move this into the closure it reverts to the default values.
+ val useUnsafeReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED)
+
+ // If true, enable using the custom RecordReader for parquet. This only works for
+ // a subset of the types (no complex types).
+ val enableUnsafeRowParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED.key).toBoolean
+ val enableVectorizedParquetReader: Boolean =
+ sqlContext.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key).toBoolean
+ val enableWholestageCodegen: Boolean =
+ sqlContext.getConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key).toBoolean
+
+ (file: PartitionedFile) => {
+ val fileSplit =
+ new FileSplit(new Path(new URI(file.filePath)), file.start, file.length, Array.empty)
+
+ val split =
+ new org.apache.parquet.hadoop.ParquetInputSplit(
+ fileSplit.getPath,
+ fileSplit.getStart,
+ fileSplit.getStart + fileSplit.getLength,
+ fileSplit.getLength,
+ fileSplit.getLocations,
+ null)
+
+ val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+ val hadoopAttemptContext = new TaskAttemptContextImpl(broadcastedConf.value.value, attemptId)
+
+ val parquetReader = try {
+ if (!useUnsafeReader) sys.error("Unsafe reader turned off.")
+ val unsafeReader = new UnsafeRowParquetRecordReader()
+ unsafeReader.initialize(split, hadoopAttemptContext)
+
+ if (enableVectorizedParquetReader) {
+ unsafeReader.initBatch(partitionSchema, file.partitionValues)
+ // Whole stage codegen (PhysicalRDD) is able to deal with batches directly
+ // TODO: fix column appending
+ if (enableWholestageCodegen) {
+ unsafeReader.enableReturningBatches()
+ }
+ }
+ unsafeReader
+ } catch {
+ case NonFatal(e) =>
+ logError(s"Falling back to parquet-mr: $e", e)
+ val reader = pushed match {
+ case Some(filter) =>
+ new ParquetRecordReader[InternalRow](
+ new CatalystReadSupport,
+ FilterCompat.get(filter, null))
+ case _ =>
+ new ParquetRecordReader[InternalRow](new CatalystReadSupport)
+ }
+ reader.initialize(split, hadoopAttemptContext)
+ reader
+ }
+
+ val iter = new RecordReaderIterator(parquetReader)
+ val fullSchema = dataSchema.toAttributes ++ partitionSchema.toAttributes
+ val joinedRow = new JoinedRow()
+ val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+
+ // UnsafeRowParquetRecordReader appends the columns internally to avoid another copy.
+ if (parquetReader.isInstanceOf[UnsafeRowParquetRecordReader]) {
--- End diff --
UnsafeRowParquetRecordReader could still produce UnsafeRow (without partition values), if enableVectorizedParquetReader is false
---
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-198569295
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53565/
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197069801
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53237/
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197587267
**[Test build #53369 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53369/consoleFull)** for PR 11709 at commit [`19e455c`](https://github.com/apache/spark/commit/19e455cc69c4f3ae69c2f6fc784c18cce8092c60).
---
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: [WIP][SPARK-13883][SQL] Parquet Implementation...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197058846
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53229/
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-13883][SQL] Parquet Implementation of F...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11709#issuecomment-197615064
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53372/
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