[SPARK-26188][SQL] FileIndex: don't infer data types of partition columns if user specifies schema #23165

gengliangwang · 2018-11-28T16:21:46Z

What changes were proposed in this pull request?

This PR is to fix a regression introduced in: https://github.com/apache/spark/pull/21004/files#r236998030

If user specifies schema, Spark don't need to infer data type for of partition columns, otherwise the data type might not match with the one user provided.
E.g. for partition directory p=4d, after data type inference the column value will be 4.0.
See https://issues.apache.org/jira/browse/SPARK-26188 for more details.

Note that user specified schema might not cover all the data columns:

val schema = new StructType()
  .add("id", StringType)
  .add("ex", ArrayType(StringType))
val df = spark.read
  .schema(schema)
  .format("parquet")
  .load(src.toString)

assert(df.schema.toList === List(
  StructField("ex", ArrayType(StringType)),
  StructField("part", IntegerType), // inferred partitionColumn dataType
  StructField("id", StringType))) // used user provided partitionColumn dataType

For the missing columns in user specified schema, Spark still need to infer their data types if partitionColumnTypeInferenceEnabled is enabled.

To implement the partially inference, refactor PartitioningUtils.parsePartitions and pass the user specified schema as parameter to cast partition values.

How was this patch tested?

Add unit test.

gengliangwang · 2018-11-28T16:22:10Z

@cloud-fan @mgaido91

SparkQA · 2018-11-28T18:54:05Z

Test build #99385 has finished for PR 23165 at commit 2866a9e.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-11-29T03:00:01Z

...e/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala

    userSpecifiedSchema match {
      case Some(userProvidedSchema) if userProvidedSchema.nonEmpty =>
+        val inferredPartitionSpec = PartitioningUtils.parsePartitions(
+          leafDirs,
+          typeInference = false,


can you add some comment, so that we don't make the same mistake in the future?

cloud-fan · 2018-11-29T03:00:53Z

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala

+      stringToFile(file, "text")
+      val path = new Path(dir.getCanonicalPath)
+      val schema = StructType(Seq(StructField("a", StringType, false)))
+      val catalog = new InMemoryFileIndex(spark, Seq(path), Map.empty, Some(schema))


catalog ->fileIndex

mgaido91

we need also to revert the changes in castPartitionValuesToUserSchema...but I am not sure they are causing the test failures...

gengliangwang · 2018-11-29T11:39:55Z

@mgaido91 The user specified schema might not match the full data schema. For the missing columns, we still need to infer their data types if partitionColumnTypeInferenceEnabled is enabled.
I will come up with a solution soon.

cloud-fan · 2018-11-29T14:30:25Z

...e/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala

-    userSpecifiedSchema match {
-      case Some(userProvidedSchema) if userProvidedSchema.nonEmpty =>
-        val userPartitionSchema =
-          combineInferredAndUserSpecifiedPartitionSchema(inferredPartitionSpec)


we can remove combineInferredAndUserSpecifiedPartitionSchema now

cloud-fan · 2018-11-29T14:35:24Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala

-      val literal = inferPartitionColumnValue(rawColumnValue, typeInference, timeZone)
+      val literal = if (userSpecifiedDataTypes.contains(columnName)) {
+        // SPARK-26188: if user provides corresponding column schema, process the column as String
+        //              type and cast it as user specified data type later.


can we do the cast here? It's a good practise to put related code together

the function returns Option[(String, Literal)]

the function inferPartitionColumnValue is quite complex, don't want change it or write duplicated logic.

cloud-fan · 2018-11-29T14:36:46Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala

      timeZone: TimeZone): PartitionSpec = {
+    val userSpecifiedDataTypes = if (userSpecifiedSchema.isDefined) {


can we build this at the caller side out of PartitioningUtils? Then we only need one extra parameter.

Personally I prefer to make the parameter simple and easy to understand. So that the logic of caller(outside the PartitioningUtils) looks cleaner.

mgaido91 · 2018-11-29T14:35:36Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala

+    val userSpecifiedDataTypes = if (userSpecifiedSchema.isDefined) {
+      val nameToDataType = userSpecifiedSchema.get.fields.map(f => f.name -> f.dataType).toMap
+      if (caseSensitive) {
+        CaseInsensitiveMap(nameToDataType)


isn't this if !caseSensitive?

Yes, thanks for pointing it out :)

mgaido91 · 2018-11-29T14:36:51Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala

+      val literal = if (userSpecifiedDataTypes.contains(columnName)) {
+        // SPARK-26188: if user provides corresponding column schema, process the column as String
+        //              type and cast it as user specified data type later.
+        inferPartitionColumnValue(rawColumnValue, false, timeZone)


can't we add the cast here?

See my reasons above.

can't we make it returning Option[(String, Literal)]? If not, what about Literal(Cast(...).eval())?

Thanks. I will use the Cast one.

cloud-fan · 2018-11-29T15:27:48Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala

        }
      }

      // Finally, we create `Partition`s based on paths and resolved partition values.
      val partitions = resolvedPartitionValues.zip(pathsWithPartitionValues).map {
-        case (PartitionValues(_, literals), (path, _)) =>
+        case (PartitionValues(columnNames, literals), (path, _)) =>


unnecessary change?

SparkQA · 2018-11-29T16:59:50Z

Test build #99452 has finished for PR 23165 at commit 0a20c9a.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-11-29T17:18:16Z

Test build #99451 has finished for PR 23165 at commit f25730e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-11-29T18:53:52Z

Test build #99458 has finished for PR 23165 at commit 7c5bdd1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-11-29T19:17:44Z

Test build #99460 has finished for PR 23165 at commit 36c65e5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-11-30T04:01:18Z

thanks, merging to master/2.4!

…umns if user specifies schema ## What changes were proposed in this pull request? This PR is to fix a regression introduced in: https://github.com/apache/spark/pull/21004/files#r236998030 If user specifies schema, Spark don't need to infer data type for of partition columns, otherwise the data type might not match with the one user provided. E.g. for partition directory `p=4d`, after data type inference the column value will be `4.0`. See https://issues.apache.org/jira/browse/SPARK-26188 for more details. Note that user specified schema **might not cover all the data columns**: ``` val schema = new StructType() .add("id", StringType) .add("ex", ArrayType(StringType)) val df = spark.read .schema(schema) .format("parquet") .load(src.toString) assert(df.schema.toList === List( StructField("ex", ArrayType(StringType)), StructField("part", IntegerType), // inferred partitionColumn dataType StructField("id", StringType))) // used user provided partitionColumn dataType ``` For the missing columns in user specified schema, Spark still need to infer their data types if `partitionColumnTypeInferenceEnabled` is enabled. To implement the partially inference, refactor `PartitioningUtils.parsePartitions` and pass the user specified schema as parameter to cast partition values. ## How was this patch tested? Add unit test. Closes #23165 from gengliangwang/fixFileIndex. Authored-by: Gengliang Wang <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 9cfc3ee) Signed-off-by: Wenchen Fan <[email protected]>

…umns if user specifies schema ## What changes were proposed in this pull request? This PR is to fix a regression introduced in: https://github.com/apache/spark/pull/21004/files#r236998030 If user specifies schema, Spark don't need to infer data type for of partition columns, otherwise the data type might not match with the one user provided. E.g. for partition directory `p=4d`, after data type inference the column value will be `4.0`. See https://issues.apache.org/jira/browse/SPARK-26188 for more details. Note that user specified schema **might not cover all the data columns**: ``` val schema = new StructType() .add("id", StringType) .add("ex", ArrayType(StringType)) val df = spark.read .schema(schema) .format("parquet") .load(src.toString) assert(df.schema.toList === List( StructField("ex", ArrayType(StringType)), StructField("part", IntegerType), // inferred partitionColumn dataType StructField("id", StringType))) // used user provided partitionColumn dataType ``` For the missing columns in user specified schema, Spark still need to infer their data types if `partitionColumnTypeInferenceEnabled` is enabled. To implement the partially inference, refactor `PartitioningUtils.parsePartitions` and pass the user specified schema as parameter to cast partition values. ## How was this patch tested? Add unit test. Closes apache#23165 from gengliangwang/fixFileIndex. Authored-by: Gengliang Wang <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

bersprockets · 2019-02-25T21:33:00Z

Hi @gengliangwang @cloud-fan

I noticed this PR changed how mixed-cased partition columns are handled when the user provides a schema.

Say I have this file structure (note that each instance of pS is mixed case):

bash-3.2$ find partitioned5 -type d
partitioned5
partitioned5/pi=2
partitioned5/pi=2/pS=foo
partitioned5/pi=2/pS=bar
partitioned5/pi=1
partitioned5/pi=1/pS=foo
partitioned5/pi=1/pS=bar
bash-3.2$

If I load the file with a user-provided schema in 2.4 (before this PR was committed) or 2.3, I see:

scala> val df = spark.read.schema("intField int, pi int, ps string").parquet("partitioned5")
df: org.apache.spark.sql.DataFrame = [intField: int, pi: int ... 1 more field]
scala> df.printSchema
root
 |-- intField: integer (nullable = true)
 |-- pi: integer (nullable = true)
 |-- ps: string (nullable = true)
scala>

However, with this PR I see:

scala> val df = spark.read.schema("intField int, pi int, ps string").parquet("partitioned5")
df: org.apache.spark.sql.DataFrame = [intField: int, pi: int ... 1 more field]
scala> df.printSchema
root
 |-- intField: integer (nullable = true)
 |-- pi: integer (nullable = true)
 |-- pS: string (nullable = true)
scala>

Spark is picking up the mixed-case column name pS from the directory name, not the lower-case ps from my specified schema.

In all cases, spark.sql.caseSensitive is set to the default (false).

Not sure is this is an issue, but it is a difference.

Edit: I filed a Jira: https://issues.apache.org/jira/browse/SPARK-26990

gengliangwang · 2019-02-26T07:16:07Z

@bersprockets Thanks for reporting the issue. I have created #23894 for it.

…umns if user specifies schema ## What changes were proposed in this pull request? This PR is to fix a regression introduced in: https://github.com/apache/spark/pull/21004/files#r236998030 If user specifies schema, Spark don't need to infer data type for of partition columns, otherwise the data type might not match with the one user provided. E.g. for partition directory `p=4d`, after data type inference the column value will be `4.0`. See https://issues.apache.org/jira/browse/SPARK-26188 for more details. Note that user specified schema **might not cover all the data columns**: ``` val schema = new StructType() .add("id", StringType) .add("ex", ArrayType(StringType)) val df = spark.read .schema(schema) .format("parquet") .load(src.toString) assert(df.schema.toList === List( StructField("ex", ArrayType(StringType)), StructField("part", IntegerType), // inferred partitionColumn dataType StructField("id", StringType))) // used user provided partitionColumn dataType ``` For the missing columns in user specified schema, Spark still need to infer their data types if `partitionColumnTypeInferenceEnabled` is enabled. To implement the partially inference, refactor `PartitioningUtils.parsePartitions` and pass the user specified schema as parameter to cast partition values. ## How was this patch tested? Add unit test. Closes apache#23165 from gengliangwang/fixFileIndex. Authored-by: Gengliang Wang <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 9cfc3ee) Signed-off-by: Wenchen Fan <[email protected]>

* [SPARK-26709][SQL] OptimizeMetadataOnlyQuery does not handle empty records correctly ## What changes were proposed in this pull request? When reading from empty tables, the optimization `OptimizeMetadataOnlyQuery` may return wrong results: ``` sql("CREATE TABLE t (col1 INT, p1 INT) USING PARQUET PARTITIONED BY (p1)") sql("INSERT INTO TABLE t PARTITION (p1 = 5) SELECT ID FROM range(1, 1)") sql("SELECT MAX(p1) FROM t") ``` The result is supposed to be `null`. However, with the optimization the result is `5`. The rule is originally ported from https://issues.apache.org/jira/browse/HIVE-1003 in apache#13494. In Hive, the rule is disabled by default in a later release(https://issues.apache.org/jira/browse/HIVE-15397), due to the same problem. It is hard to completely avoid the correctness issue. Because data sources like Parquet can be metadata-only. Spark can't tell whether it is empty or not without actually reading it. This PR disable the optimization by default. ## How was this patch tested? Unit test Closes apache#23635 from gengliangwang/optimizeMetadata. Lead-authored-by: Gengliang Wang <[email protected]> Co-authored-by: Xiao Li <[email protected]> Signed-off-by: gatorsmile <[email protected]> (cherry picked from commit f5b9370) Signed-off-by: gatorsmile <[email protected]> * [SPARK-26080][PYTHON] Skips Python resource limit on Windows in Python worker ## What changes were proposed in this pull request? `resource` package is a Unix specific package. See https://docs.python.org/2/library/resource.html and https://docs.python.org/3/library/resource.html. Note that we document Windows support: > Spark runs on both Windows and UNIX-like systems (e.g. Linux, Mac OS). This should be backported into branch-2.4 to restore Windows support in Spark 2.4.1. ## How was this patch tested? Manually mocking the changed logics. Closes apache#23055 from HyukjinKwon/SPARK-26080. Lead-authored-by: hyukjinkwon <[email protected]> Co-authored-by: Hyukjin Kwon <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]> (cherry picked from commit 9cda9a8) Signed-off-by: Hyukjin Kwon <[email protected]> * [SPARK-26873][SQL] Use a consistent timestamp to build Hadoop Job IDs. ## What changes were proposed in this pull request? Updates FileFormatWriter to create a consistent Hadoop Job ID for a write. ## How was this patch tested? Existing tests for regressions. Closes apache#23777 from rdblue/SPARK-26873-fix-file-format-writer-job-ids. Authored-by: Ryan Blue <[email protected]> Signed-off-by: Marcelo Vanzin <[email protected]> (cherry picked from commit 33334e2) Signed-off-by: Marcelo Vanzin <[email protected]> * [SPARK-26745][SPARK-24959][SQL][BRANCH-2.4] Revert count optimization in JSON datasource by ## What changes were proposed in this pull request? This PR reverts JSON count optimization part of apache#21909. We cannot distinguish the cases below without parsing: ``` [{...}, {...}] ``` ``` [] ``` ``` {...} ``` ```bash # empty string ``` when we `count()`. One line (input: IN) can be, 0 record, 1 record and multiple records and this is dependent on each input. See also apache#23665 (comment). ## How was this patch tested? Manually tested. Closes apache#23708 from HyukjinKwon/SPARK-26745-backport. Authored-by: Hyukjin Kwon <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]> * [SPARK-26677][BUILD] Update Parquet to 1.10.1 with notEq pushdown fix. ## What changes were proposed in this pull request? Update to Parquet Java 1.10.1. ## How was this patch tested? Added a test from HyukjinKwon that validates the notEq case from SPARK-26677. Closes apache#23704 from rdblue/SPARK-26677-fix-noteq-parquet-bug. Lead-authored-by: Ryan Blue <[email protected]> Co-authored-by: Hyukjin Kwon <[email protected]> Co-authored-by: Ryan Blue <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> (cherry picked from commit f72d217) Signed-off-by: Dongjoon Hyun <[email protected]> * [SPARK-26677][FOLLOWUP][BRANCH-2.4] Update Parquet manifest with Hadoop-2.6 ## What changes were proposed in this pull request? During merging Parquet upgrade PR, `hadoop-2.6` profile dependency manifest is missed. ## How was this patch tested? Manual. ``` ./dev/test-dependencies.sh ``` Also, this will recover `branch-2.4` with `hadoop-2.6` build. - https://amplab.cs.berkeley.edu/jenkins/view/Spark%20QA%20Test%20(Dashboard)/job/spark-branch-2.4-test-sbt-hadoop-2.6/281/ Closes apache#23738 from dongjoon-hyun/SPARK-26677-2. Authored-by: Dongjoon Hyun <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> * [SPARK-26708][SQL][BRANCH-2.4] Incorrect result caused by inconsistency between a SQL cache's cached RDD and its physical plan ## What changes were proposed in this pull request? When performing non-cascading cache invalidation, `recache` is called on the other cache entries which are dependent on the cache being invalidated. It leads to the the physical plans of those cache entries being re-compiled. For those cache entries, if the cache RDD has already been persisted, chances are there will be inconsistency between the data and the new plan. It can cause a correctness issue if the new plan's `outputPartitioning` or `outputOrdering` is different from the that of the actual data, and meanwhile the cache is used by another query that asks for specific `outputPartitioning` or `outputOrdering` which happens to match the new plan but not the actual data. The fix is to keep the cache entry as it is if the data has been loaded, otherwise re-build the cache entry, with a new plan and an empty cache buffer. ## How was this patch tested? Added UT. Closes apache#23678 from maryannxue/spark-26708-2.4. Authored-by: maryannxue <[email protected]> Signed-off-by: Takeshi Yamamuro <[email protected]> * [SPARK-26267][SS] Retry when detecting incorrect offsets from Kafka (2.4) ## What changes were proposed in this pull request? Backport apache#23324 to branch-2.4. ## How was this patch tested? Jenkins Closes apache#23365 from zsxwing/SPARK-26267-2.4. Authored-by: Shixiong Zhu <[email protected]> Signed-off-by: Shixiong Zhu <[email protected]> * [SPARK-26706][SQL] Fix `illegalNumericPrecedence` for ByteType This PR contains a minor change in `Cast$mayTruncate` that fixes its logic for bytes. Right now, `mayTruncate(ByteType, LongType)` returns `false` while `mayTruncate(ShortType, LongType)` returns `true`. Consequently, `spark.range(1, 3).as[Byte]` and `spark.range(1, 3).as[Short]` behave differently. Potentially, this bug can silently corrupt someone's data. ```scala // executes silently even though Long is converted into Byte spark.range(Long.MaxValue - 10, Long.MaxValue).as[Byte] .map(b => b - 1) .show() +-----+ |value| +-----+ | -12| | -11| | -10| | -9| | -8| | -7| | -6| | -5| | -4| | -3| +-----+ // throws an AnalysisException: Cannot up cast `id` from bigint to smallint as it may truncate spark.range(Long.MaxValue - 10, Long.MaxValue).as[Short] .map(s => s - 1) .show() ``` This PR comes with a set of unit tests. Closes apache#23632 from aokolnychyi/cast-fix. Authored-by: Anton Okolnychyi <[email protected]> Signed-off-by: DB Tsai <[email protected]> * [SPARK-26078][SQL][BACKPORT-2.4] Dedup self-join attributes on IN subqueries ## What changes were proposed in this pull request? When there is a self-join as result of a IN subquery, the join condition may be invalid, resulting in trivially true predicates and return wrong results. The PR deduplicates the subquery output in order to avoid the issue. ## How was this patch tested? added UT Closes apache#23449 from mgaido91/SPARK-26078_2.4. Authored-by: Marco Gaido <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> * [SPARK-26233][SQL][BACKPORT-2.4] CheckOverflow when encoding a decimal value ## What changes were proposed in this pull request? When we encode a Decimal from external source we don't check for overflow. That method is useful not only in order to enforce that we can represent the correct value in the specified range, but it also changes the underlying data to the right precision/scale. Since in our code generation we assume that a decimal has exactly the same precision and scale of its data type, missing to enforce it can lead to corrupted output/results when there are subsequent transformations. ## How was this patch tested? added UT Closes apache#23232 from mgaido91/SPARK-26233_2.4. Authored-by: Marco Gaido <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> * [SPARK-27097][CHERRY-PICK 2.4] Avoid embedding platform-dependent offsets literally in whole-stage generated code ## What changes were proposed in this pull request? Spark SQL performs whole-stage code generation to speed up query execution. There are two steps to it: - Java source code is generated from the physical query plan on the driver. A single version of the source code is generated from a query plan, and sent to all executors. - It's compiled to bytecode on the driver to catch compilation errors before sending to executors, but currently only the generated source code gets sent to the executors. The bytecode compilation is for fail-fast only. - Executors receive the generated source code and compile to bytecode, then the query runs like a hand-written Java program. In this model, there's an implicit assumption about the driver and executors being run on similar platforms. Some code paths accidentally embedded platform-dependent object layout information into the generated code, such as: ```java Platform.putLong(buffer, /* offset */ 24, /* value */ 1); ``` This code expects a field to be at offset +24 of the `buffer` object, and sets a value to that field. But whole-stage code generation generally uses platform-dependent information from the driver. If the object layout is significantly different on the driver and executors, the generated code can be reading/writing to wrong offsets on the executors, causing all kinds of data corruption. One code pattern that leads to such problem is the use of `Platform.XXX` constants in generated code, e.g. `Platform.BYTE_ARRAY_OFFSET`. Bad: ```scala val baseOffset = Platform.BYTE_ARRAY_OFFSET // codegen template: s"Platform.putLong($buffer, $baseOffset, $value);" ``` This will embed the value of `Platform.BYTE_ARRAY_OFFSET` on the driver into the generated code. Good: ```scala val baseOffset = "Platform.BYTE_ARRAY_OFFSET" // codegen template: s"Platform.putLong($buffer, $baseOffset, $value);" ``` This will generate the offset symbolically -- `Platform.putLong(buffer, Platform.BYTE_ARRAY_OFFSET, value)`, which will be able to pick up the correct value on the executors. Caveat: these offset constants are declared as runtime-initialized `static final` in Java, so they're not compile-time constants from the Java language's perspective. It does lead to a slightly increased size of the generated code, but this is necessary for correctness. NOTE: there can be other patterns that generate platform-dependent code on the driver which is invalid on the executors. e.g. if the endianness is different between the driver and the executors, and if some generated code makes strong assumption about endianness, it would also be problematic. ## How was this patch tested? Added a new test suite `WholeStageCodegenSparkSubmitSuite`. This test suite needs to set the driver's extraJavaOptions to force the driver and executor use different Java object layouts, so it's run as an actual SparkSubmit job. Authored-by: Kris Mok <kris.mokdatabricks.com> Closes apache#24032 from gatorsmile/testFailure. Lead-authored-by: Kris Mok <[email protected]> Co-authored-by: gatorsmile <[email protected]> Signed-off-by: DB Tsai <[email protected]> * [SPARK-26188][SQL] FileIndex: don't infer data types of partition columns if user specifies schema ## What changes were proposed in this pull request? This PR is to fix a regression introduced in: https://github.com/apache/spark/pull/21004/files#r236998030 If user specifies schema, Spark don't need to infer data type for of partition columns, otherwise the data type might not match with the one user provided. E.g. for partition directory `p=4d`, after data type inference the column value will be `4.0`. See https://issues.apache.org/jira/browse/SPARK-26188 for more details. Note that user specified schema **might not cover all the data columns**: ``` val schema = new StructType() .add("id", StringType) .add("ex", ArrayType(StringType)) val df = spark.read .schema(schema) .format("parquet") .load(src.toString) assert(df.schema.toList === List( StructField("ex", ArrayType(StringType)), StructField("part", IntegerType), // inferred partitionColumn dataType StructField("id", StringType))) // used user provided partitionColumn dataType ``` For the missing columns in user specified schema, Spark still need to infer their data types if `partitionColumnTypeInferenceEnabled` is enabled. To implement the partially inference, refactor `PartitioningUtils.parsePartitions` and pass the user specified schema as parameter to cast partition values. ## How was this patch tested? Add unit test. Closes apache#23165 from gengliangwang/fixFileIndex. Authored-by: Gengliang Wang <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 9cfc3ee) Signed-off-by: Wenchen Fan <[email protected]> * [SPARK-25921][PYSPARK] Fix barrier task run without BarrierTaskContext while python worker reuse ## What changes were proposed in this pull request? Running a barrier job after a normal spark job causes the barrier job to run without a BarrierTaskContext. This is because while python worker reuse, BarrierTaskContext._getOrCreate() will still return a TaskContext after firstly submit a normal spark job, we'll get a `AttributeError: 'TaskContext' object has no attribute 'barrier'`. Fix this by adding check logic in BarrierTaskContext._getOrCreate() and make sure it will return BarrierTaskContext in this scenario. ## How was this patch tested? Add new UT in pyspark-core. Closes apache#22962 from xuanyuanking/SPARK-25921. Authored-by: Yuanjian Li <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit c00e72f) Signed-off-by: Wenchen Fan <[email protected]>

fix

2866a9e

cloud-fan reviewed Nov 29, 2018

View reviewed changes

mgaido91 reviewed Nov 29, 2018

View reviewed changes

gengliangwang added 2 commits November 29, 2018 21:32

fix

f25730e

revise

0a20c9a

cloud-fan reviewed Nov 29, 2018

View reviewed changes

mgaido91 reviewed Nov 29, 2018

View reviewed changes

address comments

7c5bdd1

cloud-fan reviewed Nov 29, 2018

View reviewed changes

remove one unnecessary change

36c65e5

cloud-fan approved these changes Nov 29, 2018

View reviewed changes

asfgit closed this in 9cfc3ee Nov 30, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-26188][SQL] FileIndex: don't infer data types of partition columns if user specifies schema #23165

[SPARK-26188][SQL] FileIndex: don't infer data types of partition columns if user specifies schema #23165

gengliangwang commented Nov 28, 2018 •

edited

Loading

gengliangwang commented Nov 28, 2018

SparkQA commented Nov 28, 2018

cloud-fan Nov 29, 2018

cloud-fan Nov 29, 2018

mgaido91 left a comment •

edited

Loading

gengliangwang commented Nov 29, 2018 •

edited

Loading

cloud-fan Nov 29, 2018

cloud-fan Nov 29, 2018

gengliangwang Nov 29, 2018

cloud-fan Nov 29, 2018

gengliangwang Nov 29, 2018

mgaido91 Nov 29, 2018

gengliangwang Nov 29, 2018

mgaido91 Nov 29, 2018

gengliangwang Nov 29, 2018

mgaido91 Nov 29, 2018

gengliangwang Nov 29, 2018

cloud-fan Nov 29, 2018

SparkQA commented Nov 29, 2018

SparkQA commented Nov 29, 2018

SparkQA commented Nov 29, 2018

SparkQA commented Nov 29, 2018

cloud-fan commented Nov 30, 2018

bersprockets commented Feb 25, 2019 •

edited

Loading

gengliangwang commented Feb 26, 2019

		timeZone: TimeZone): PartitionSpec = {
		val userSpecifiedDataTypes = if (userSpecifiedSchema.isDefined) {

[SPARK-26188][SQL] FileIndex: don't infer data types of partition columns if user specifies schema #23165

[SPARK-26188][SQL] FileIndex: don't infer data types of partition columns if user specifies schema #23165

Conversation

gengliangwang commented Nov 28, 2018 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

gengliangwang commented Nov 28, 2018

SparkQA commented Nov 28, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mgaido91 left a comment • edited Loading

Choose a reason for hiding this comment

gengliangwang commented Nov 29, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Nov 29, 2018

SparkQA commented Nov 29, 2018

SparkQA commented Nov 29, 2018

SparkQA commented Nov 29, 2018

cloud-fan commented Nov 30, 2018

bersprockets commented Feb 25, 2019 • edited Loading

gengliangwang commented Feb 26, 2019

gengliangwang commented Nov 28, 2018 •

edited

Loading

mgaido91 left a comment •

edited

Loading

gengliangwang commented Nov 29, 2018 •

edited

Loading

bersprockets commented Feb 25, 2019 •

edited

Loading