You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by brkyvz <gi...@git.apache.org> on 2016/11/20 22:35:12 UTC
[GitHub] spark pull request #15951: [SPARK-18510] Fix data corruption from inferred p...
GitHub user brkyvz opened a pull request:
https://github.com/apache/spark/pull/15951
[SPARK-18510] Fix data corruption from inferred partition column dataTypes
## What changes were proposed in this pull request?
### The Issue
If I specify my schema when doing
```scala
spark.read
.schema(someSchemaWherePartitionColumnsAreStrings)
```
but if the partition inference can infer it as IntegerType or I assume LongType or DoubleType (basically fixed size types), then once UnsafeRows are generated, your data will be corrupted.
### Proposed solution
The partition handling code path is kind of a mess. In my fix I'm probably adding to the mess, but at least trying to standardize the code path.
The real issue is that a user that uses the `spark.read` code path can never clearly specify what the partition columns are. If you try to specify the fields in `schema`, we practically ignore what the user provides, and fall back to our inferred data types. What happens in the end is data corruption.
My solution tries to fix this by always trying to infer partition columns the first time you specify the table. Once we find what the partition columns are, we try to find them in the user specified schema and use the dataType provided there, or fall back to the smallest common data type.
We will ALWAYS append partition columns to the user's schema, even if they didn't ask for it. We will only use the data type they provided if they specified it. While this is confusing, this has been the behavior since Spark 1.6, and I didn't want to change this behavior in the QA period of Spark 2.1. We may revisit this decision later.
A side effect of this PR is that we won't need https://github.com/apache/spark/pull/15942 if this PR goes in.
## How was this patch tested?
Regression tests
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/brkyvz/spark partition-corruption
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/15951.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 #15951
----
commit ef2e1c26cbe20c3f7b2fd6a4d528b2b29d842203
Author: Burak Yavuz <br...@gmail.com>
Date: 2016-11-20T22:19:33Z
Fix data corruption from inferred partition column dataTypes
commit 9080c4ed50e9ddf65d3e045e0604398618256fb4
Author: Burak Yavuz <br...@gmail.com>
Date: 2016-11-20T22:26:08Z
make naming and comments a bit more descriptive
----
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tejasapatil <gi...@git.apache.org>.
Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89015376
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
--- End diff --
This feels ugly. A client can accidentally end up using the data schema. Ideally, these could be separate methods for data schema and partition schema.
In case thats not possible, how about returning `None` instead of returning `new StructType()` ?
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/69010/
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88991938
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamReaderWriterSuite.scala ---
@@ -532,4 +532,50 @@ class DataStreamReaderWriterSuite extends StreamTest with BeforeAndAfter {
assert(e.getMessage.contains("does not support recovering"))
assert(e.getMessage.contains("checkpoint location"))
}
+
+ test("SPARK-18510: Data corruption from user specified partition column schemas") {
--- End diff --
I would say rename the test to something that explains what this test actually tests. For example, "use user-specified schema for partitioning columns in file sources"
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68975 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68975/consoleFull)** for PR 15951 at commit [`97003e2`](https://github.com/apache/spark/commit/97003e24ea377ae0d844e9f90f020b2111339661).
* 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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89030464
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return new StructType() -> partitionSchema
--- End diff --
done
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89007528
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,96 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return partitionSchema -> partitionSchema.map(_.name)
+ }
+ if (catalogTable.isDefined && userSpecifiedSchema.isDefined) {
+ // backwards compatibility before SPARK-18510. Return the schema of catalog tables as is
--- End diff --
its returning the user specified schema, not the catalog table schema.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/15951
Also cc @ericl @cloud-fan who just recently change the related codes.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68964 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68964/consoleFull)** for PR 15951 at commit [`ff0a2c7`](https://github.com/apache/spark/commit/ff0a2c72aeb629fc8067caac9133536ed6e20853).
* 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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89018476
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,90 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
--- End diff --
This aint right. You cant return something incorrect. Rather return null for the first schema.
also, the docs is confusing in this way. please add `@param` and `@return` after the params to clarify what gets returned in both cases.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89030487
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
--- End diff --
I included `format`. Not sure how to access the name
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68947 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68947/consoleFull)** for PR 15951 at commit [`f0a2754`](https://github.com/apache/spark/commit/f0a2754b244212af3e65601fad6b1e5c07aad3a2).
* This patch **fails SparkR 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88835591
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -272,14 +309,20 @@ case class DataSource(
HadoopFsRelation(
fileCatalog,
- partitionSchema = fileCatalog.partitionSpec().partitionColumns,
+ partitionSchema = fileCatalog.partitionSchema,
dataSchema = dataSchema,
bucketSpec = None,
format,
caseInsensitiveOptions)(sparkSession)
// This is a non-streaming file based datasource.
case (format: FileFormat, _) =>
+ val (schema, inferredPartitionColumns) = inferFileFormatSchema(format)
--- End diff --
`inferFileFormatSchema` is expensive, right?
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89400157
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -281,33 +361,25 @@ case class DataSource(
// This is a non-streaming file based datasource.
case (format: FileFormat, _) =>
val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.flatMap { path =>
val hdfsPath = new Path(path)
- val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val fs = hdfsPath.getFileSystem(hadoopConf)
--- End diff --
> any need for this change? hadoopConf is not reused any where else
this is to avoid creating a new hadoopConf for each 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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89248592
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
- val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val fs = hdfsPath.getFileSystem(hadoopConf)
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to materialize the lazy val below
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ val inferredOpt = inferredPartitions.find(_.name == partitionColumn)
+ if (inferredOpt.isDefined) {
+ logDebug(
+ s"""Type of partition column: $partitionColumn not found in specified schema
+ |for $format.
+ |User Specified Schema
+ |=====================
+ |${userSpecifiedSchema.orNull}
+ |
+ |Falling back to inferred dataType if it exists.
+ """.stripMargin)
+ }
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException(s"Failed to resolve the schema for $format for " +
+ s"the partition column: $partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return (null, partitionSchema)
+ }
+ val dataSchema = userSpecifiedSchema.map { schema =>
+ val equality = sparkSession.sessionState.conf.resolver
+ StructType(schema.filterNot(f => partitionSchema.exists(p => equality(p.name, f.name))))
+ }.orElse {
+ format.inferSchema(
--- End diff --
Yes.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tejasapatil <gi...@git.apache.org>.
Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89014484
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
--- End diff --
`Managed tables` is a thing in Hive (google for managed and external tables in hive). `Hive managed tables` can be confused to that. You could say 'tables managed by Hive`.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on the issue:
https://github.com/apache/spark/pull/15951
True. But there's no reason "part" and "id" can't be strings right?
On Nov 21, 2016 12:16 AM, "Xiao Li" <no...@github.com> wrote:
> The real issue is that a user that uses the spark.read code path can never
> clearly specify what the partition columns are. If you try to specify the
> fields in schema, we practically ignore what the user provides, and fall
> back to our inferred data types. What happens in the end is data corruption.
>
> For data source tables, the partition columns are part of data schema.
> Users do not need to know which columns are used for partitioning. If they
> can provide the right types, they should be able to see the expected data.
>
> In the test cases, we can get the correct result with the following
> changes:
>
> spark.range(4).select(createArray('id + 1) as 'ex, 'id, 'id % 4 as 'part).coalesce(1)
> val schema = new StructType()
> .add("part", LongType)
> .add("ex", ArrayType(StringType))
> .add("id", LongType)
> spark.read
> .schema(schema)
> .format("parquet")
> .load(src.toString).show()
>
> \u2014
> You are receiving this because you authored the thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/spark/pull/15951#issuecomment-261871343>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/AFACe7DmaOEHYPu22xbaNPZZl5OKZMwvks5rAVNagaJpZM4K3rkM>
> .
>
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tejasapatil <gi...@git.apache.org>.
Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89018962
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
--- End diff --
- can you include the name of the table / relation ? the computation can refer to multiple tables which might have same `partitionColumn`
- can you also print the specified schema ? seems like would be informative for debugging
- It would be good to include the inferred datatype used instead of just mentioning that we would fallback
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tejasapatil <gi...@git.apache.org>.
Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89018372
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
--- End diff --
Creating new Hadoop config object for every path seems expensive to me. Can it be created once outside the `flatMap` and re-used inside ?
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88994472
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,90 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ partitionSchema -> partitionSchema.map(_.name)
+ }
+ val tableSchema = userSpecifiedSchema.map { schema =>
--- End diff --
good catch. missed the `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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88984227
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,90 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
--- End diff --
can you add param docs to define what `justPartitioning` means?
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89011122
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -291,22 +360,24 @@ case class DataSource(
throw new AnalysisException(s"Path does not exist: $qualified")
}
// Sufficient to check head of the globPath seq for non-glob scenario
- if (checkFilesExist && !fs.exists(globPath.head)) {
+ // Don't need to check once again if files exist in streaming mode
+ if (!isStreaming && !fs.exists(globPath.head)) {
throw new AnalysisException(s"Path does not exist: ${globPath.head}")
}
globPath
}.toArray
- // If they gave a schema, then we try and figure out the types of the partition columns
- // from that schema.
- val partitionSchema = userSpecifiedSchema.map { schema =>
- StructType(
- partitionColumns.map { c =>
- // TODO: Case sensitivity.
- schema
- .find(_.name.toLowerCase() == c.toLowerCase())
- .getOrElse(throw new AnalysisException(s"Invalid partition column '$c'"))
- })
+ val (schema, inferredPartitionColumns) = getOrInferFileFormatSchema(format)
+ val dataSchema = if (isStreaming) {
+ schema
+ } else {
+ StructType(schema.dropRight(inferredPartitionColumns.length))
+ }
+
+ val partitionSchema = if (inferredPartitionColumns.isEmpty) {
--- End diff --
This is the one i dont understand. why should the behavior be different if this non-streaming source (HadoopFSRelation) is being create by user directly or by the file streaming source.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89242805
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
--- End diff --
Nit: `tempFileIndex`
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89030610
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
--- End diff --
reworded to `materialize 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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tejasapatil <gi...@git.apache.org>.
Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89017468
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
--- End diff --
Feels like you wanted to say `we will never have to re-compute this`. Typo ?
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tejasapatil <gi...@git.apache.org>.
Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89017872
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
--- End diff --
Every call of this method will re-compute `tempFileCatalog` which as per your comment is an expensive op. This kinda contradicts 3rd point in the method doc which says that subsequent invocations of this method would be cheap. Did I miss anything ?
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/15951
When users specify schemas, we do not want to infer the schemas due to its potentially expensive cost. Based on my understanding, data corruption issues are common when user-specified schemas do not provide correct types.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68979 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68979/consoleFull)** for PR 15951 at commit [`08566e7`](https://github.com/apache/spark/commit/08566e72d04f7f2334d676211caca8bc0ae99290).
* 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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88934128
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala ---
@@ -573,4 +573,39 @@ class DataFrameReaderWriterSuite extends QueryTest with SharedSQLContext with Be
}
}
}
+
+ test("SPARK-18510: Data corruption from user specified partition column schemas") {
+ import org.apache.spark.sql.functions.udf
+ import testImplicits._
+ withTempDir { src =>
+ val createArray = udf { (length: Long) =>
+ for (i <- 1 to length.toInt) yield i.toString
+ }
+ spark.range(4).select(createArray('id + 1) as 'ex, 'id, 'id % 4 as 'part).coalesce(1).write
+ .partitionBy("part", "id")
+ .mode("overwrite")
+ .parquet(src.toString)
+ // make sure to specify the schema in the wrong order. Partition column in the middle, etc.
--- End diff --
with the fix, it still does not matter. The comment is outdated, I thought something else was the problem. In terms of schema though, the output from Spark is always consistent, i.e. partition columns go last.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68909 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68909/consoleFull)** for PR 15951 at commit [`9080c4e`](https://github.com/apache/spark/commit/9080c4ed50e9ddf65d3e045e0604398618256fb4).
* 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89007519
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,96 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return partitionSchema -> partitionSchema.map(_.name)
+ }
+ if (catalogTable.isDefined && userSpecifiedSchema.isDefined) {
+ // backwards compatibility before SPARK-18510. Return the schema of catalog tables as is
+ return userSpecifiedSchema.get -> partitionSchema.map(_.name)
+ }
+ val tableSchema = userSpecifiedSchema.map { schema =>
+ val equality = sparkSession.sessionState.conf.resolver
+ StructType(schema.filterNot(f => partitionSchema.exists(p => equality(p.name, f.name))))
+ }.orElse {
+ format.inferSchema(
sparkSession,
caseInsensitiveOptions,
- fileCatalog.allFiles())
-
- inferred.map { inferredSchema =>
- StructType(inferredSchema ++ partitionSchema) -> partitionSchema.map(_.name)
- }
+ tempFileCatalog.allFiles())
}.getOrElse {
- throw new AnalysisException("Unable to infer schema. It must be specified manually.")
+ val exampleFiles = tempFileCatalog.allFiles().take(2).mkString(",")
--- End diff --
I tried to keep it consistent with before. I can remove it.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89007308
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -291,22 +360,24 @@ case class DataSource(
throw new AnalysisException(s"Path does not exist: $qualified")
}
// Sufficient to check head of the globPath seq for non-glob scenario
- if (checkFilesExist && !fs.exists(globPath.head)) {
+ // Don't need to check once again if files exist in streaming mode
+ if (!isStreaming && !fs.exists(globPath.head)) {
throw new AnalysisException(s"Path does not exist: ${globPath.head}")
}
globPath
}.toArray
- // If they gave a schema, then we try and figure out the types of the partition columns
- // from that schema.
- val partitionSchema = userSpecifiedSchema.map { schema =>
- StructType(
- partitionColumns.map { c =>
- // TODO: Case sensitivity.
- schema
- .find(_.name.toLowerCase() == c.toLowerCase())
- .getOrElse(throw new AnalysisException(s"Invalid partition column '$c'"))
- })
+ val (schema, inferredPartitionColumns) = getOrInferFileFormatSchema(format)
+ val dataSchema = if (isStreaming) {
+ schema
+ } else {
+ StructType(schema.dropRight(inferredPartitionColumns.length))
--- End diff --
It looks elegant, but it opens up the possibility of a new kind of bug where the types in partitioning columns in the returned partitionSchema and dataSchema are different.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68946 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68946/consoleFull)** for PR 15951 at commit [`de49ba5`](https://github.com/apache/spark/commit/de49ba516871013dcad35874d57cada4b8fd8daf).
* This patch **fails SparkR 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68965 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68965/consoleFull)** for PR 15951 at commit [`330e148`](https://github.com/apache/spark/commit/330e148114f26484aba5e9cb16aac2f286cd05ba).
* 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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88938054
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -272,14 +309,20 @@ case class DataSource(
HadoopFsRelation(
fileCatalog,
- partitionSchema = fileCatalog.partitionSpec().partitionColumns,
+ partitionSchema = fileCatalog.partitionSchema,
dataSchema = dataSchema,
bucketSpec = None,
format,
caseInsensitiveOptions)(sparkSession)
// This is a non-streaming file based datasource.
case (format: FileFormat, _) =>
+ val (schema, inferredPartitionColumns) = inferFileFormatSchema(format)
--- End diff --
Sounds good
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89030586
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
--- End diff --
Since everything is already defined in `userSpecifiedSchema` once we create the `FileStreamSource`, we will never have to materialize this variable, because we will not use it.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/15951
```
spark.read
.schema(someSchemaWherePartitionColumnsAreStrings)
```
I don't think this is valid use case, `DataFrameReader` can't specify partition columns, so we will always infer partitions.
I think the real problem is `HadoopFsRelation.schema`:
```
val schema: StructType = {
val dataSchemaColumnNames = dataSchema.map(_.name.toLowerCase).toSet
StructType(dataSchema ++ partitionSchema.filterNot { column =>
dataSchemaColumnNames.contains(column.name.toLowerCase)
})
}
```
It sliently drops the partition schema if the partition column names are duplicated in data schema.
I think the best solution is to add `partitionBy` in `DataFrameReader` so that we can skip inferring partitions really. But this maybe too late for 2.1, we should define a better semantic for the current "broken" API.
> Once we find what the partition columns are, we try to find them in the user specified schema and use the dataType provided there, or fall back to the smallest common data type.
This 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68946 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68946/consoleFull)** for PR 15951 at commit [`de49ba5`](https://github.com/apache/spark/commit/de49ba516871013dcad35874d57cada4b8fd8daf).
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/15951
Your concern is right. I just did another try. Using the string-type columns as partition columns. See the following code:
```Scala
val rowRdd: RDD[Row] = sparkContext.parallelize(1 to 10).map(i => Row(i, i.toString))
val inputSchema = StructType(Seq(
StructField("intCol", IntegerType),
StructField("stringCol", StringType)
))
spark.createDataFrame(rowRdd, inputSchema)
.write.partitionBy("stringCol").mode("overwrite").parquet(src.toString)
val schema = new StructType()
.add("intCol", IntegerType)
.add("stringCol", IntegerType)
spark.read
.schema(schema)
.format("parquet")
.load(src.toString).show()
```
Users have to use `IntegerType` for these partition columns even if the original data type is `StringType`. This looks werid. Otherwise, they will hit the following error:
```
Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 2, localhost, executor driver): java.lang.NullPointerException
at org.apache.spark.sql.execution.vectorized.OnHeapColumnVector.getArrayLength(OnHeapColumnVector.java:375)
```
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89224222
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -281,33 +361,25 @@ case class DataSource(
// This is a non-streaming file based datasource.
case (format: FileFormat, _) =>
val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.flatMap { path =>
val hdfsPath = new Path(path)
- val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val fs = hdfsPath.getFileSystem(hadoopConf)
--- End diff --
any need for this change? hadoopConf is not reused any where else
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on the issue:
https://github.com/apache/spark/pull/15951
@ericl I feel that would probably break 90% of production Spark jobs out there, therefore am a bit scared of something radical. I agree, it's confusing and annoying
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89012272
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -291,22 +360,24 @@ case class DataSource(
throw new AnalysisException(s"Path does not exist: $qualified")
}
// Sufficient to check head of the globPath seq for non-glob scenario
- if (checkFilesExist && !fs.exists(globPath.head)) {
+ // Don't need to check once again if files exist in streaming mode
+ if (!isStreaming && !fs.exists(globPath.head)) {
throw new AnalysisException(s"Path does not exist: ${globPath.head}")
}
globPath
}.toArray
- // If they gave a schema, then we try and figure out the types of the partition columns
- // from that schema.
- val partitionSchema = userSpecifiedSchema.map { schema =>
- StructType(
- partitionColumns.map { c =>
- // TODO: Case sensitivity.
- schema
- .find(_.name.toLowerCase() == c.toLowerCase())
- .getOrElse(throw new AnalysisException(s"Invalid partition column '$c'"))
- })
+ val (schema, inferredPartitionColumns) = getOrInferFileFormatSchema(format)
+ val dataSchema = if (isStreaming) {
+ schema
+ } else {
+ StructType(schema.dropRight(inferredPartitionColumns.length))
--- End diff --
good idea
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68975/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68964 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68964/consoleFull)** for PR 15951 at commit [`ff0a2c7`](https://github.com/apache/spark/commit/ff0a2c72aeb629fc8067caac9133536ed6e20853).
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68913 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68913/consoleFull)** for PR 15951 at commit [`46ab68a`](https://github.com/apache/spark/commit/46ab68a0cb752e0d1ac6f7d8ebf1bfc4f178dc13).
* 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68910 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68910/consoleFull)** for PR 15951 at commit [`f518405`](https://github.com/apache/spark/commit/f5184058469d89df1014008936f1b16000b0f88e).
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88996973
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,96 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return partitionSchema -> partitionSchema.map(_.name)
+ }
+ if (catalogTable.isDefined && userSpecifiedSchema.isDefined) {
+ // backwards compatibility before SPARK-18510. Return the schema of catalog tables as is
+ return userSpecifiedSchema.get -> partitionSchema.map(_.name)
+ }
+ val tableSchema = userSpecifiedSchema.map { schema =>
+ val equality = sparkSession.sessionState.conf.resolver
+ StructType(schema.filterNot(f => partitionSchema.exists(p => equality(p.name, f.name))))
+ }.orElse {
+ format.inferSchema(
sparkSession,
caseInsensitiveOptions,
- fileCatalog.allFiles())
-
- inferred.map { inferredSchema =>
- StructType(inferredSchema ++ partitionSchema) -> partitionSchema.map(_.name)
- }
+ tempFileCatalog.allFiles())
}.getOrElse {
- throw new AnalysisException("Unable to infer schema. It must be specified manually.")
+ val exampleFiles = tempFileCatalog.allFiles().take(2).mkString(",")
--- End diff --
This might generate some unwanted files. For example, in Json, [`inferSchema`](https://github.com/apache/spark/blob/01dd0083011741c2bbe5ae1d2a25f2c9a1302b76/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala#L60) filter out the unwanted files.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68975 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68975/consoleFull)** for PR 15951 at commit [`97003e2`](https://github.com/apache/spark/commit/97003e24ea377ae0d844e9f90f020b2111339661).
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/15951
> The real issue is that a user that uses the spark.read code path can never clearly specify what the partition columns are. If you try to specify the fields in schema, we practically ignore what the user provides, and fall back to our inferred data types. What happens in the end is data corruption.
For data source tables, the partition columns are part of data schema. Users do not need to know which columns are used for partitioning. If they can provide the right types, they should be able to see the expected data.
In the test cases, we can get the correct result with the following changes:
```
spark.range(4).select(createArray('id + 1) as 'ex, 'id, 'id % 4 as 'part).coalesce(1)
val schema = new StructType()
.add("part", LongType)
.add("ex", ArrayType(StringType))
.add("id", LongType)
spark.read
.schema(schema)
.format("parquet")
.load(src.toString).show()
```
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68956 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68956/consoleFull)** for PR 15951 at commit [`fde4f64`](https://github.com/apache/spark/commit/fde4f642979ba49fd191106783b7e00a596a93a8).
* 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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89262935
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala ---
@@ -274,7 +274,7 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
pathToPartitionedTable,
userSpecifiedSchema = Option("num int, str string"),
userSpecifiedPartitionCols = partitionCols,
- expectedSchema = new StructType().add("num", IntegerType).add("str", StringType),
+ expectedSchema = new StructType().add("str", StringType).add("num", IntegerType),
--- End diff --
I believe the original test case was incorrect. Although the schema check passes, if you really read rows out of the Dataset, you'll hit an exception, as shown in the following Spark shell session:
```scala
import org.apache.spark.sql.types._
val df0 = spark.range(10).select(
('id % 4) cast StringType as "part",
'id cast StringType as "data"
)
val path = "/tmp/part.parquet"
df0.write.mode("overwrite").partitionBy("part").parquet(path)
val df1 = spark.read.schema(
new StructType()
.add("part", StringType, nullable = true)
.add("data", StringType, nullable = true)
).parquet(path)
df1.printSchema()
// root
// |-- part: string (nullable = true)
// |-- data: string (nullable = true)
df1.show()
// 16/11/22 22:52:21 ERROR Executor: Exception in task 0.0 in stage 10.0 (TID 34)
// java.lang.NullPointerException
// at org.apache.spark.sql.execution.vectorized.OnHeapColumnVector.getArrayLength(OnHeapColumnVector.java:375)
// at org.apache.spark.sql.execution.vectorized.ColumnVector.getArray(ColumnVector.java:554)
// at org.apache.spark.sql.execution.vectorized.ColumnVector.getByteArray(ColumnVector.java:576)
// [...]
```
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68913/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68956 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68956/consoleFull)** for PR 15951 at commit [`fde4f64`](https://github.com/apache/spark/commit/fde4f642979ba49fd191106783b7e00a596a93a8).
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tejasapatil <gi...@git.apache.org>.
Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89019196
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
--- End diff --
can you include the name of the table / relation ? the computation can refer to multiple tables which might have same `partitionColumn`
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89252556
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
--- End diff --
I think it would be clearer if we can split this method into two: one for partition schema and the other for data schema. In this way, we can also remove the `justPartitioning` argument by calling the method you need at the right place.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88985887
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,90 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
--- End diff --
Please add docs on why this is lazy. It took me half-a-minute to trace down why this should be lazy.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on the issue:
https://github.com/apache/spark/pull/15951
Thanks @tejasapatil for the review. Addressed your comments
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the issue:
https://github.com/apache/spark/pull/15951
Would be good if @rxin can take a look.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89223421
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -144,8 +224,8 @@ case class DataSource(
"you may be able to create a static DataFrame on that directory with " +
"'spark.read.load(directory)' and infer schema from it.")
}
- val (schema, partCols) = inferFileFormatSchema(format)
- SourceInfo(s"FileSource[$path]", schema, partCols)
+ val (schema, partCols) = getOrInferFileFormatSchema(format)
--- End diff --
rename to dataSchema and partitionSchema
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68947/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68910 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68910/consoleFull)** for PR 15951 at commit [`f518405`](https://github.com/apache/spark/commit/f5184058469d89df1014008936f1b16000b0f88e).
* 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #69010 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/69010/consoleFull)** for PR 15951 at commit [`f3b42ff`](https://github.com/apache/spark/commit/f3b42ffbd69a4fae3aa9d4dd22d75bb9f63a4586).
* 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on the issue:
https://github.com/apache/spark/pull/15951
cc @rxin @marmbrus Don't know who's the best person to look at this, but git blame sais I mainly changed your code :)
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68946/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68909 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68909/consoleFull)** for PR 15951 at commit [`9080c4e`](https://github.com/apache/spark/commit/9080c4ed50e9ddf65d3e045e0604398618256fb4).
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by brkyvz <gi...@git.apache.org>.
Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88934282
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -272,14 +309,20 @@ case class DataSource(
HadoopFsRelation(
fileCatalog,
- partitionSchema = fileCatalog.partitionSpec().partitionColumns,
+ partitionSchema = fileCatalog.partitionSchema,
dataSchema = dataSchema,
bucketSpec = None,
format,
caseInsensitiveOptions)(sparkSession)
// This is a non-streaming file based datasource.
case (format: FileFormat, _) =>
+ val (schema, inferredPartitionColumns) = inferFileFormatSchema(format)
--- End diff --
not if you already specified the schema. I'm going to rename it to `getOrInfer` so that people don't think it's expensive all the time
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88991557
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala ---
@@ -274,7 +274,7 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
pathToPartitionedTable,
userSpecifiedSchema = Option("num int, str string"),
userSpecifiedPartitionCols = partitionCols,
- expectedSchema = new StructType().add("num", IntegerType).add("str", StringType),
+ expectedSchema = new StructType().add("str", StringType).add("num", IntegerType),
--- End diff --
so in this PR, for some cases, the order of fields in schema created after resolveRelation is changing?
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89012513
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,96 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
--- End diff --
Can you document what "least amount of work" is? That is, it will skip file scanning if .....
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88992024
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,90 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ partitionSchema -> partitionSchema.map(_.name)
+ }
+ val tableSchema = userSpecifiedSchema.map { schema =>
--- End diff --
Are you missing `else` here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89223766
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -281,33 +361,25 @@ case class DataSource(
// This is a non-streaming file based datasource.
case (format: FileFormat, _) =>
val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.flatMap { path =>
val hdfsPath = new Path(path)
- val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val fs = hdfsPath.getFileSystem(hadoopConf)
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
val globPath = SparkHadoopUtil.get.globPathIfNecessary(qualified)
if (globPath.isEmpty) {
throw new AnalysisException(s"Path does not exist: $qualified")
}
// Sufficient to check head of the globPath seq for non-glob scenario
+ // Don't need to check once again if files exist in streaming mode
if (checkFilesExist && !fs.exists(globPath.head)) {
throw new AnalysisException(s"Path does not exist: ${globPath.head}")
}
globPath
}.toArray
- // If they gave a schema, then we try and figure out the types of the partition columns
- // from that schema.
- val partitionSchema = userSpecifiedSchema.map { schema =>
- StructType(
- partitionColumns.map { c =>
- // TODO: Case sensitivity.
- schema
- .find(_.name.toLowerCase() == c.toLowerCase())
- .getOrElse(throw new AnalysisException(s"Invalid partition column '$c'"))
- })
- }
+ val (dataSchema, inferredPartitionSchema) = getOrInferFileFormatSchema(format)
--- End diff --
nit: this may not be inferred right? so just partitionSchema would be a better name.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89224653
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -281,33 +361,25 @@ case class DataSource(
// This is a non-streaming file based datasource.
case (format: FileFormat, _) =>
val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.flatMap { path =>
--- End diff --
Just to confirm, for paths with globs, this would expand here ONCE, and then expand them AGAIN in `getOrInferFileFormatSchema`, right?
If so, we dont have to fix it in this PR, but we should document this in a JIRA or something for fixing later.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68956/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68913 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68913/consoleFull)** for PR 15951 at commit [`46ab68a`](https://github.com/apache/spark/commit/46ab68a0cb752e0d1ac6f7d8ebf1bfc4f178dc13).
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89006009
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,96 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return partitionSchema -> partitionSchema.map(_.name)
+ }
+ if (catalogTable.isDefined && userSpecifiedSchema.isDefined) {
+ // backwards compatibility before SPARK-18510. Return the schema of catalog tables as is
+ return userSpecifiedSchema.get -> partitionSchema.map(_.name)
+ }
+ val tableSchema = userSpecifiedSchema.map { schema =>
+ val equality = sparkSession.sessionState.conf.resolver
+ StructType(schema.filterNot(f => partitionSchema.exists(p => equality(p.name, f.name))))
+ }.orElse {
+ format.inferSchema(
sparkSession,
caseInsensitiveOptions,
- fileCatalog.allFiles())
-
- inferred.map { inferredSchema =>
- StructType(inferredSchema ++ partitionSchema) -> partitionSchema.map(_.name)
- }
+ tempFileCatalog.allFiles())
}.getOrElse {
- throw new AnalysisException("Unable to infer schema. It must be specified manually.")
+ val exampleFiles = tempFileCatalog.allFiles().take(2).mkString(",")
--- End diff --
Why will it generate the files? at max it may print few file names that the json format actually ignores while inferring schema.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68965 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68965/consoleFull)** for PR 15951 at commit [`330e148`](https://github.com/apache/spark/commit/330e148114f26484aba5e9cb16aac2f286cd05ba).
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88991007
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala ---
@@ -126,7 +126,6 @@ case class AnalyzeCreateTable(sparkSession: SparkSession) extends Rule[LogicalPl
normalizeColumnName(tableDesc.identifier, schema, colName, "partition")
}
checkDuplication(normalizedPartitionCols, "partition")
-
--- End diff --
undo this change.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #69010 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/69010/consoleFull)** for PR 15951 at commit [`f3b42ff`](https://github.com/apache/spark/commit/f3b42ffbd69a4fae3aa9d4dd22d75bb9f63a4586).
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89263076
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
--- End diff --
Well, just realized that it might be hard to split because of the temporary `InMemoryFileIndex`.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by ericl <gi...@git.apache.org>.
Github user ericl commented on the issue:
https://github.com/apache/spark/pull/15951
This might change the behavior, but how about just raising an error if the partition types differ from those provided by the user, or the user failed to provide a partitioning schema? It seems confusing to partially infer a schema when the user does not provide it.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r88996397
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala ---
@@ -573,4 +573,40 @@ class DataFrameReaderWriterSuite extends QueryTest with SharedSQLContext with Be
}
}
}
+
+ test("SPARK-18510: Data corruption from user specified partition column schemas") {
--- End diff --
same comment as above.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89006242
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,96 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return partitionSchema -> partitionSchema.map(_.name)
+ }
+ if (catalogTable.isDefined && userSpecifiedSchema.isDefined) {
+ // backwards compatibility before SPARK-18510. Return the schema of catalog tables as is
+ return userSpecifiedSchema.get -> partitionSchema.map(_.name)
+ }
+ val tableSchema = userSpecifiedSchema.map { schema =>
+ val equality = sparkSession.sessionState.conf.resolver
+ StructType(schema.filterNot(f => partitionSchema.exists(p => equality(p.name, f.name))))
+ }.orElse {
+ format.inferSchema(
sparkSession,
caseInsensitiveOptions,
- fileCatalog.allFiles())
-
- inferred.map { inferredSchema =>
- StructType(inferredSchema ++ partitionSchema) -> partitionSchema.map(_.name)
- }
+ tempFileCatalog.allFiles())
}.getOrElse {
- throw new AnalysisException("Unable to infer schema. It must be specified manually.")
+ val exampleFiles = tempFileCatalog.allFiles().take(2).mkString(",")
--- End diff --
None the less, to avoid this, could you not just print the `allPaths`?
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68976 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68976/consoleFull)** for PR 15951 at commit [`6f741b6`](https://github.com/apache/spark/commit/6f741b617573613b62d6fe2f0c7eca1cbf66f660).
* This patch **fails SparkR 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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68976 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68976/consoleFull)** for PR 15951 at commit [`6f741b6`](https://github.com/apache/spark/commit/6f741b617573613b62d6fe2f0c7eca1cbf66f660).
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the issue:
https://github.com/apache/spark/pull/15951
Thanks @liancheng for your comments. Since these are mostly nits, I am going to merge this PR (since it fixes critical bug for 2.1) and address the final comments in a separate PR.
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68965/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68910/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68979/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68964/
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89249380
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
- val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val fs = hdfsPath.getFileSystem(hadoopConf)
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to materialize the lazy val below
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ val inferredOpt = inferredPartitions.find(_.name == partitionColumn)
+ if (inferredOpt.isDefined) {
+ logDebug(
+ s"""Type of partition column: $partitionColumn not found in specified schema
+ |for $format.
+ |User Specified Schema
+ |=====================
+ |${userSpecifiedSchema.orNull}
+ |
+ |Falling back to inferred dataType if it exists.
+ """.stripMargin)
+ }
+ inferredPartitions.find(_.name == partitionColumn)
--- End diff --
Duplicated code?
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89249078
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
- val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val fs = hdfsPath.getFileSystem(hadoopConf)
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to materialize the lazy val below
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
--- End diff --
Also need to use the resolver to handle case sensitivity here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89242786
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
--- End diff --
Nit: `tempFileIndex`
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68947 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68947/consoleFull)** for PR 15951 at commit [`f0a2754`](https://github.com/apache/spark/commit/f0a2754b244212af3e65601fad6b1e5c07aad3a2).
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89223308
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +88,106 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only managed tables by Hive provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * @param format the file format object for this DataSource
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning.
+ * @return A pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns. If `justPartitioning` is `true`, then the dataSchema will be provided as
+ * `null`.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
- val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
+ val fs = hdfsPath.getFileSystem(hadoopConf)
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to materialize the lazy val below
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ val inferredOpt = inferredPartitions.find(_.name == partitionColumn)
+ if (inferredOpt.isDefined) {
+ logDebug(
+ s"""Type of partition column: $partitionColumn not found in specified schema
+ |for $format.
+ |User Specified Schema
+ |=====================
+ |${userSpecifiedSchema.orNull}
+ |
+ |Falling back to inferred dataType if it exists.
+ """.stripMargin)
+ }
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException(s"Failed to resolve the schema for $format for " +
+ s"the partition column: $partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return (null, partitionSchema)
+ }
+ val dataSchema = userSpecifiedSchema.map { schema =>
+ val equality = sparkSession.sessionState.conf.resolver
+ StructType(schema.filterNot(f => partitionSchema.exists(p => equality(p.name, f.name))))
+ }.orElse {
+ format.inferSchema(
--- End diff --
just to confirm, inferschema returns schema without the partition columns?
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68909/
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68976/
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89010923
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -291,22 +360,24 @@ case class DataSource(
throw new AnalysisException(s"Path does not exist: $qualified")
}
// Sufficient to check head of the globPath seq for non-glob scenario
- if (checkFilesExist && !fs.exists(globPath.head)) {
+ // Don't need to check once again if files exist in streaming mode
+ if (!isStreaming && !fs.exists(globPath.head)) {
throw new AnalysisException(s"Path does not exist: ${globPath.head}")
}
globPath
}.toArray
- // If they gave a schema, then we try and figure out the types of the partition columns
- // from that schema.
- val partitionSchema = userSpecifiedSchema.map { schema =>
- StructType(
- partitionColumns.map { c =>
- // TODO: Case sensitivity.
- schema
- .find(_.name.toLowerCase() == c.toLowerCase())
- .getOrElse(throw new AnalysisException(s"Invalid partition column '$c'"))
- })
+ val (schema, inferredPartitionColumns) = getOrInferFileFormatSchema(format)
+ val dataSchema = if (isStreaming) {
+ schema
+ } else {
+ StructType(schema.dropRight(inferredPartitionColumns.length))
--- End diff --
nvm. with justPartitioning param, its more confusing as it is right now. so I like this suggestion
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/15951
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15951
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15951
**[Test build #68979 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68979/consoleFull)** for PR 15951 at commit [`08566e7`](https://github.com/apache/spark/commit/08566e72d04f7f2334d676211caca8bc0ae99290).
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89018539
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,95 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to skip file scanning whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of the data schema (excluding partition columns) and the schema of the partition
+ * columns.
+ *
+ * @param justPartitioning Whether to exit early and provide just the schema partitioning. The
+ * data schema is incorrect in this case and should not be used.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, StructType) = {
+ // the operations below are expensive therefore try not to do them if we don't need to
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return new StructType() -> partitionSchema
--- End diff --
better to return so that its obvious that this cannot be used.
---
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 #15951: [SPARK-18510] Fix data corruption from inferred p...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15951#discussion_r89001357
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala ---
@@ -84,30 +84,96 @@ case class DataSource(
private val caseInsensitiveOptions = new CaseInsensitiveMap(options)
/**
- * Infer the schema of the given FileFormat, returns a pair of schema and partition column names.
+ * Get the schema of the given FileFormat, if provided by `userSpecifiedSchema`, or try to infer
+ * it. In the read path, only Hive managed tables provide the partition columns properly when
+ * initializing this class. All other file based data sources will try to infer the partitioning,
+ * and then cast the inferred types to user specified dataTypes if the partition columns exist
+ * inside `userSpecifiedSchema`, otherwise we can hit data corruption bugs like SPARK-18510.
+ * This method will try to do the least amount of work given whether `userSpecifiedSchema` and
+ * `partitionColumns` are provided. Here are some code paths that use this method:
+ * 1. `spark.read` (no schema): Most amount of work. Infer both schema and partitioning columns
+ * 2. `spark.read.schema(userSpecifiedSchema)`: Parse partitioning columns, cast them to the
+ * dataTypes provided in `userSpecifiedSchema` if they exist or fallback to inferred
+ * dataType if they don't.
+ * 3. `spark.readStream.schema(userSpecifiedSchema)`: For streaming use cases, users have to
+ * provide the schema. Here, we also perform partition inference like 2, and try to use
+ * dataTypes in `userSpecifiedSchema`. All subsequent triggers for this stream will re-use
+ * this information, therefore calls to this method should be very cheap, i.e. there won't
+ * be any further inference in any triggers.
+ * 4. `df.saveAsTable(tableThatExisted)`: In this case, we call this method to resolve the
+ * existing table's partitioning scheme. This is achieved by not providing
+ * `userSpecifiedSchema`. For this case, we add the boolean `justPartitioning` for an early
+ * exit, if we don't care about the schema of the original table.
+ *
+ * Returns a pair of schema and partition column names.
*/
- private def inferFileFormatSchema(format: FileFormat): (StructType, Seq[String]) = {
- userSpecifiedSchema.map(_ -> partitionColumns).orElse {
- val allPaths = caseInsensitiveOptions.get("path")
+ private def getOrInferFileFormatSchema(
+ format: FileFormat,
+ justPartitioning: Boolean = false): (StructType, Seq[String]) = {
+ lazy val tempFileCatalog = {
+ val allPaths = caseInsensitiveOptions.get("path") ++ paths
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
- val fileCatalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
- val partitionSchema = fileCatalog.partitionSpec().partitionColumns
- val inferred = format.inferSchema(
+ new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
+ }
+ val partitionSchema = if (partitionColumns.isEmpty && catalogTable.isEmpty) {
+ // Try to infer partitioning, because no DataSource in the read path provides the partitioning
+ // columns properly unless it is a Hive DataSource
+ val resolved = tempFileCatalog.partitionSchema.map { partitionField =>
+ val equality = sparkSession.sessionState.conf.resolver
+ // SPARK-18510: try to get schema from userSpecifiedSchema, otherwise fallback to inferred
+ userSpecifiedSchema.flatMap(_.find(f => equality(f.name, partitionField.name))).getOrElse(
+ partitionField)
+ }
+ StructType(resolved)
+ } else {
+ // in streaming mode, we have already inferred and registered partition columns, we will
+ // never have to re-use this
+ lazy val inferredPartitions = tempFileCatalog.partitionSchema
+ // maintain old behavior before SPARK-18510. If userSpecifiedSchema is empty used inferred
+ // partitioning
+ if (userSpecifiedSchema.isEmpty) {
+ inferredPartitions
+ } else {
+ val partitionFields = partitionColumns.map { partitionColumn =>
+ userSpecifiedSchema.flatMap(_.find(_.name == partitionColumn)).orElse {
+ logDebug(s"Schema of partition column: $partitionColumn not found in specified " +
+ "schema. Falling back to inferred dataType if it exists.")
+ inferredPartitions.find(_.name == partitionColumn)
+ }.getOrElse {
+ throw new AnalysisException("Failed to resolve the schema for the partition column: " +
+ s"$partitionColumn. It must be specified manually.")
+ }
+ }
+ StructType(partitionFields)
+ }
+ }
+ if (justPartitioning) {
+ return partitionSchema -> partitionSchema.map(_.name)
+ }
+ if (catalogTable.isDefined && userSpecifiedSchema.isDefined) {
+ // backwards compatibility before SPARK-18510. Return the schema of catalog tables as is
+ return userSpecifiedSchema.get -> partitionSchema.map(_.name)
+ }
+ val tableSchema = userSpecifiedSchema.map { schema =>
--- End diff --
`tableSchema` -> `dataSchema`
---
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 issue #15951: [SPARK-18510] Fix data corruption from inferred partitio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/15951
shall we update the document of `DataSource.partitionColumns`? `When this ist is empty, the relation is unpartitioned.` this is wrong now, the relation can be partitioned even if `partitionColumns` is empty.
---
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