[spark] branch master updated: [SPARK-40965][SQL] Rename the error class `_LEGACY_ERROR_TEMP_1208 ` to `FIELD_NOT_FOUND`
This is an automated email from the ASF dual-hosted git repository.
maxgekk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new f0d7b31ad61 [SPARK-40965][SQL] Rename the error class
`_LEGACY_ERROR_TEMP_1208 ` to `FIELD_NOT_FOUND`
f0d7b31ad61 is described below
commit f0d7b31ad617d9c112f81eef2130040adbcf454d
Author: Max Gekk
AuthorDate: Sun Oct 30 12:29:00 2022 +0300
[SPARK-40965][SQL] Rename the error class `_LEGACY_ERROR_TEMP_1208 ` to
`FIELD_NOT_FOUND`
### What changes were proposed in this pull request?
In the PR, I propose to assign the proper name `FIELD_NOT_FOUND ` to the
legacy error class `_LEGACY_ERROR_TEMP_1208 `, and modify test suite to use
`checkError()` which checks the error class name, context and etc.
### Why are the changes needed?
Proper name improves user experience w/ Spark SQL.
### Does this PR introduce _any_ user-facing change?
Yes, the PR changes an user-facing error message.
### How was this patch tested?
By running the modified test suites:
```
$ build/sbt "test:testOnly *AnalysisErrorSuite"
$ build/sbt "test:testOnly *ResolveSubquerySuite"
$ build/sbt "test:testOnly *EncoderResolutionSuite"
$ build/sbt "test:testOnly *SQLQuerySuite"
```
Closes #38435 from MaxGekk/field-not-found-error-class.
Authored-by: Max Gekk
Signed-off-by: Max Gekk
---
core/src/main/resources/error/error-classes.json | 10 +-
.../spark/sql/errors/QueryCompilationErrors.scala | 6 +-
.../sql/catalyst/analysis/AnalysisErrorSuite.scala | 5 +-
.../catalyst/analysis/ResolveSubquerySuite.scala | 6 +-
.../catalyst/encoders/EncoderResolutionSuite.scala | 10 +-
.../expressions/AttributeResolutionSuite.scala | 10 +-
.../results/typeCoercion/native/mapZipWith.sql.out | 1 +
.../apache/spark/sql/ColumnExpressionSuite.scala | 107 ++---
.../spark/sql/DataFrameSetOperationsSuite.scala| 12 ++-
.../scala/org/apache/spark/sql/SQLQuerySuite.scala | 20 ++--
.../spark/sql/execution/SQLViewTestSuite.scala | 12 ++-
.../FileMetadataStructRowIndexSuite.scala | 12 ++-
.../datasources/FileMetadataStructSuite.scala | 32 +++---
13 files changed, 154 insertions(+), 89 deletions(-)
diff --git a/core/src/main/resources/error/error-classes.json
b/core/src/main/resources/error/error-classes.json
index 4797ee0d0d0..e45b6e3bdb6 100644
--- a/core/src/main/resources/error/error-classes.json
+++ b/core/src/main/resources/error/error-classes.json
@@ -356,6 +356,11 @@
],
"sqlState" : "22023"
},
+ "FIELD_NOT_FOUND" : {
+"message" : [
+ "No such struct field in ."
+]
+ },
"FORBIDDEN_OPERATION" : {
"message" : [
"The operation is not allowed on the :
"
@@ -2479,11 +2484,6 @@
"The duration and time inputs to window must be an integer, long or
string literal."
]
},
- "_LEGACY_ERROR_TEMP_1208" : {
-"message" : [
- "No such struct field in ."
-]
- },
"_LEGACY_ERROR_TEMP_1209" : {
"message" : [
"Ambiguous reference to fields ."
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index cf7e3524d5b..f97888f046b 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -2080,10 +2080,10 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase {
def noSuchStructFieldInGivenFieldsError(
fieldName: String, fields: Array[StructField]): Throwable = {
new AnalysisException(
- errorClass = "_LEGACY_ERROR_TEMP_1208",
+ errorClass = "FIELD_NOT_FOUND",
messageParameters = Map(
-"fieldName" -> fieldName,
-"fields" -> fields.map(_.name).mkString(", ")))
+"fieldName" -> toSQLId(fieldName),
+"fields" -> fields.map(f => toSQLId(f.name)).mkString(", ")))
}
def ambiguousReferenceToFieldsError(fields: String): Throwable = {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala
index d530be5f5e4..f3bca030380 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala
@@ -370,10 +370,11 @@ class AnalysisErrorSuite extends AnalysisTest {
"Ambiguous reference to fields" :: "differentCase" :: "differentcase" ::
Nil,
caseSensitive = false)
- errorTest(
+ errorClassTest(
"missing field",
nestedRelation2.select($"top.c"),
-
[spark] branch master updated: [SPARK-40953][CONNECT][PYTHON] Fix `DataFrame.head` to collect specified number of rows
This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git The following commit(s) were added to refs/heads/master by this push: new 41ccae0922f [SPARK-40953][CONNECT][PYTHON] Fix `DataFrame.head` to collect specified number of rows 41ccae0922f is described below commit 41ccae0922f39aff5dbef236bab78959269ddd5c Author: Ruifeng Zheng AuthorDate: Sun Oct 30 20:34:57 2022 +0900 [SPARK-40953][CONNECT][PYTHON] Fix `DataFrame.head` to collect specified number of rows ### What changes were proposed in this pull request? `head` should return `self.limit(n).toPandas()` instead of `self.toPandas()` ### Why are the changes needed? bugfix ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? existing tests Closes #38424 from zhengruifeng/connect_head_fix. Authored-by: Ruifeng Zheng Signed-off-by: Hyukjin Kwon --- python/pyspark/sql/connect/dataframe.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/python/pyspark/sql/connect/dataframe.py b/python/pyspark/sql/connect/dataframe.py index bf9ed83615b..03a766aff30 100644 --- a/python/pyspark/sql/connect/dataframe.py +++ b/python/pyspark/sql/connect/dataframe.py @@ -215,8 +215,7 @@ class DataFrame(object): return GroupingFrame(self, *cols) def head(self, n: int) -> Optional["pandas.DataFrame"]: -self.limit(n) -return self.toPandas() +return self.limit(n).toPandas() # TODO(martin.grund) fix mypu def join(self, other: "DataFrame", on: Any, how: Optional[str] = None) -> "DataFrame": - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated: [SPARK-40925][SQL][SS] Fix stateful operator late record filtering
This is an automated email from the ASF dual-hosted git repository.
kabhwan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 406d0e243cf [SPARK-40925][SQL][SS] Fix stateful operator late record
filtering
406d0e243cf is described below
commit 406d0e243cfec9b29df946e1a0e20ed5fe25e152
Author: Alex Balikov <91913242+alex-bali...@users.noreply.github.com>
AuthorDate: Mon Oct 31 10:13:50 2022 +0900
[SPARK-40925][SQL][SS] Fix stateful operator late record filtering
### What changes were proposed in this pull request?
This PR fixes the input late record filtering done by stateful operators to
allow for chaining of stateful operators. Currently stateful operators are
initialized with the current microbatch watermark and perform both input late
record filtering and state eviction (e.g. producing aggregations) using the
same watermark value. The state evicted (or aggregates produced) due to
watermark advancing is behind the watermark and thus effectively late - if a
following stateful operator consume [...]
This PR provides two watermark values to the stateful operators - one from
the previous microbatch to be used for late record filtering and the one from
the current microbatch (as in the existing code) to be used for state eviction.
This solves the above problem of the broken late record filtering.
Note that this PR still does not solve the issue of time-interval stream
join producing records delayed against the watermark. Therefore time-interval
streaming join followed by stateful operators is still not supported. That will
be fixed in a follow up PR (and a SPIP) effectively replacing the single global
watermark with conceptually watermarks per operator.
Also, the stateful operator chains unblocked by this PR (e.g. a chain of
window aggregations) are still blocked by the unsupported operations checker.
The new test for these scenarios - MultiStatefulOperatorsSuite has to
explicitly disable the unsupported ops check. This again will be fixed in a
follow-up PR.
### Why are the changes needed?
The PR allows Spark Structured Streaming to support chaining of stateful
operators e.g. chaining of time window aggregations which is a meaningful
streaming scenario.
### Does this PR introduce _any_ user-facing change?
With this PR, chains of stateful operators will be supported in Spark
Structured Streaming.
### How was this patch tested?
Added a new test suite - MultiStatefulOperatorsSuite
Closes #38405 from alex-balikov/multiple_stateful-ops-base.
Lead-authored-by: Alex Balikov
<91913242+alex-bali...@users.noreply.github.com>
Co-authored-by: Alex Balikov
Signed-off-by: Jungtaek Lim
---
.../sql/catalyst/expressions/SessionWindow.scala | 24 +-
.../sql/catalyst/expressions/TimeWindow.scala | 20 +-
.../org/apache/spark/sql/internal/SQLConf.scala| 16 +
.../spark/sql/execution/QueryExecution.scala | 2 +-
.../spark/sql/execution/SparkStrategies.scala | 6 +-
.../spark/sql/execution/aggregate/AggUtils.scala | 9 +-
.../FlatMapGroupsInPandasWithStateExec.scala | 13 +-
.../streaming/FlatMapGroupsWithStateExec.scala | 27 +-
.../execution/streaming/IncrementalExecution.scala | 44 ++-
.../execution/streaming/MicroBatchExecution.scala | 1 +
.../sql/execution/streaming/OffsetSeqLog.scala | 4 +
.../streaming/StreamingSymmetricHashJoinExec.scala | 15 +-
.../StreamingSymmetricHashJoinHelper.scala | 6 +-
.../streaming/continuous/ContinuousExecution.scala | 1 +
.../streaming/sources/MicroBatchWrite.scala| 2 +-
.../execution/streaming/statefulOperators.scala| 108 +++--
.../streaming/FlatMapGroupsWithStateSuite.scala| 2 +-
.../streaming/MultiStatefulOperatorsSuite.scala| 440 +
.../spark/sql/streaming/StreamingJoinSuite.scala | 2 +-
19 files changed, 661 insertions(+), 81 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala
index 77e8dfde87b..02273b0c461 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala
@@ -68,11 +68,29 @@ case class SessionWindow(timeColumn: Expression,
gapDuration: Expression) extend
with Unevaluable
with NonSQLExpression {
+ private def inputTypeOnTimeColumn: AbstractDataType = {
+TypeCollection(
+ AnyTimestampType,
+ // Below two types cover both time window & session window, since they
produce the same type
+ // of output as window column.
+ new StructType()
+.add(Struct
[spark] branch master updated: [SPARK-40963][SQL] Set nullable correctly in project created by `ExtractGenerator`
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 90d31541fb0 [SPARK-40963][SQL] Set nullable correctly in project
created by `ExtractGenerator`
90d31541fb0 is described below
commit 90d31541fb0313d762cc36067060e6445c04a9b6
Author: Bruce Robbins
AuthorDate: Mon Oct 31 10:45:17 2022 +0900
[SPARK-40963][SQL] Set nullable correctly in project created by
`ExtractGenerator`
### What changes were proposed in this pull request?
When creating the project list for the new projection In
`ExtractGenerator`, take into account whether the generator is outer when
setting nullable on generator-related output attributes.
### Why are the changes needed?
This PR fixes an issue that can produce either incorrect results or a
`NullPointerException`. It's a bit of an obscure issue in that I am
hard-pressed to reproduce without using a subquery that has a inline table.
Example:
```
select c1, explode(c4) as c5 from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(1, 2)),
(2, array(2, 3)),
(3, null)
as data(c1, c2)
)
);
+---+---+
|c1 |c5 |
+---+---+
|1 |1 |
|1 |2 |
|2 |2 |
|2 |3 |
|3 |0 |
+---+---+
```
In the last row, `c5` is 0, but should be `NULL`.
Another example:
```
select c1, exists(c4, x -> x is null) as c5 from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(1, 2)),
(2, array(2, 3)),
(3, array())
as data(c1, c2)
)
);
+---+-+
|c1 |c5 |
+---+-+
|1 |false|
|1 |false|
|2 |false|
|2 |false|
|3 |false|
+---+-+
```
In the last row, `false` should be `true`.
In both cases, at the time `CreateArray(c3)` is instantiated, `c3`'s
nullability is incorrect because the new projection created by
`ExtractGenerator` uses `generatorOutput` from `explode_outer(c2)` as a
projection list. `generatorOutput` doesn't take into account that
`explode_outer(c2)` is an _outer_ explode, so the nullability setting is lost.
`UpdateAttributeNullability` will eventually fix the nullable setting for
attributes referring to `c3`, but it doesn't fix the `containsNull` setting for
`c4` in `explode(c4)` (from the first example) or `exists(c4, x -> x is null)`
(from the second example).
This example fails with a `NullPointerException`:
```
select c1, inline_outer(c4) from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(named_struct('a', 1, 'b', 2))),
(2, array(named_struct('a', 3, 'b', 4), named_struct('a', 5, 'b', 6))),
(3, array())
as data(c1, c2)
)
);
22/10/30 17:34:42 ERROR Executor: Exception in task 1.0 in stage 8.0 (TID
14)
java.lang.NullPointerException
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.generate_doConsume_1$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.generate_doConsume_0$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)
at
org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:364)
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit test.
Closes #38440 from bersprockets/SPARK-40963.
Authored-by: Bruce Robbins
Signed-off-by: Hyukjin Kwon
---
.../apache/spark/sql/catalyst/analysis/Analyzer.scala | 2 +-
.../plans/logical/basicLogicalOperators.scala | 17 +
.../org/apache/spark/sql/GeneratorFunctionSuite.scala | 19 +++
3 files changed, 29 insertions(+), 9 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index fc12b6522b4..c7b84405412 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -2910,7 +2910,7 @@ class Analyzer(override val catalogManag
[spark] branch branch-3.3 updated: [SPARK-40963][SQL] Set nullable correctly in project created by `ExtractGenerator`
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new c7ef5600a75 [SPARK-40963][SQL] Set nullable correctly in project
created by `ExtractGenerator`
c7ef5600a75 is described below
commit c7ef5600a75789152af40804f539a3d075cf2c0c
Author: Bruce Robbins
AuthorDate: Mon Oct 31 10:45:17 2022 +0900
[SPARK-40963][SQL] Set nullable correctly in project created by
`ExtractGenerator`
### What changes were proposed in this pull request?
When creating the project list for the new projection In
`ExtractGenerator`, take into account whether the generator is outer when
setting nullable on generator-related output attributes.
### Why are the changes needed?
This PR fixes an issue that can produce either incorrect results or a
`NullPointerException`. It's a bit of an obscure issue in that I am
hard-pressed to reproduce without using a subquery that has a inline table.
Example:
```
select c1, explode(c4) as c5 from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(1, 2)),
(2, array(2, 3)),
(3, null)
as data(c1, c2)
)
);
+---+---+
|c1 |c5 |
+---+---+
|1 |1 |
|1 |2 |
|2 |2 |
|2 |3 |
|3 |0 |
+---+---+
```
In the last row, `c5` is 0, but should be `NULL`.
Another example:
```
select c1, exists(c4, x -> x is null) as c5 from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(1, 2)),
(2, array(2, 3)),
(3, array())
as data(c1, c2)
)
);
+---+-+
|c1 |c5 |
+---+-+
|1 |false|
|1 |false|
|2 |false|
|2 |false|
|3 |false|
+---+-+
```
In the last row, `false` should be `true`.
In both cases, at the time `CreateArray(c3)` is instantiated, `c3`'s
nullability is incorrect because the new projection created by
`ExtractGenerator` uses `generatorOutput` from `explode_outer(c2)` as a
projection list. `generatorOutput` doesn't take into account that
`explode_outer(c2)` is an _outer_ explode, so the nullability setting is lost.
`UpdateAttributeNullability` will eventually fix the nullable setting for
attributes referring to `c3`, but it doesn't fix the `containsNull` setting for
`c4` in `explode(c4)` (from the first example) or `exists(c4, x -> x is null)`
(from the second example).
This example fails with a `NullPointerException`:
```
select c1, inline_outer(c4) from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(named_struct('a', 1, 'b', 2))),
(2, array(named_struct('a', 3, 'b', 4), named_struct('a', 5, 'b', 6))),
(3, array())
as data(c1, c2)
)
);
22/10/30 17:34:42 ERROR Executor: Exception in task 1.0 in stage 8.0 (TID
14)
java.lang.NullPointerException
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.generate_doConsume_1$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.generate_doConsume_0$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)
at
org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:364)
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit test.
Closes #38440 from bersprockets/SPARK-40963.
Authored-by: Bruce Robbins
Signed-off-by: Hyukjin Kwon
(cherry picked from commit 90d31541fb0313d762cc36067060e6445c04a9b6)
Signed-off-by: Hyukjin Kwon
---
.../apache/spark/sql/catalyst/analysis/Analyzer.scala | 2 +-
.../plans/logical/basicLogicalOperators.scala | 17 +
.../org/apache/spark/sql/GeneratorFunctionSuite.scala | 19 +++
3 files changed, 29 insertions(+), 9 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 3a3997ff9c7..ad40f924ef8 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/or
[spark] branch branch-3.2 updated: [SPARK-40963][SQL] Set nullable correctly in project created by `ExtractGenerator`
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new 815f32fdf4f [SPARK-40963][SQL] Set nullable correctly in project
created by `ExtractGenerator`
815f32fdf4f is described below
commit 815f32fdf4f6ea61d1eddb5f2d4d05b26a30f671
Author: Bruce Robbins
AuthorDate: Mon Oct 31 10:45:17 2022 +0900
[SPARK-40963][SQL] Set nullable correctly in project created by
`ExtractGenerator`
### What changes were proposed in this pull request?
When creating the project list for the new projection In
`ExtractGenerator`, take into account whether the generator is outer when
setting nullable on generator-related output attributes.
### Why are the changes needed?
This PR fixes an issue that can produce either incorrect results or a
`NullPointerException`. It's a bit of an obscure issue in that I am
hard-pressed to reproduce without using a subquery that has a inline table.
Example:
```
select c1, explode(c4) as c5 from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(1, 2)),
(2, array(2, 3)),
(3, null)
as data(c1, c2)
)
);
+---+---+
|c1 |c5 |
+---+---+
|1 |1 |
|1 |2 |
|2 |2 |
|2 |3 |
|3 |0 |
+---+---+
```
In the last row, `c5` is 0, but should be `NULL`.
Another example:
```
select c1, exists(c4, x -> x is null) as c5 from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(1, 2)),
(2, array(2, 3)),
(3, array())
as data(c1, c2)
)
);
+---+-+
|c1 |c5 |
+---+-+
|1 |false|
|1 |false|
|2 |false|
|2 |false|
|3 |false|
+---+-+
```
In the last row, `false` should be `true`.
In both cases, at the time `CreateArray(c3)` is instantiated, `c3`'s
nullability is incorrect because the new projection created by
`ExtractGenerator` uses `generatorOutput` from `explode_outer(c2)` as a
projection list. `generatorOutput` doesn't take into account that
`explode_outer(c2)` is an _outer_ explode, so the nullability setting is lost.
`UpdateAttributeNullability` will eventually fix the nullable setting for
attributes referring to `c3`, but it doesn't fix the `containsNull` setting for
`c4` in `explode(c4)` (from the first example) or `exists(c4, x -> x is null)`
(from the second example).
This example fails with a `NullPointerException`:
```
select c1, inline_outer(c4) from (
select c1, array(c3) as c4 from (
select c1, explode_outer(c2) as c3
from values
(1, array(named_struct('a', 1, 'b', 2))),
(2, array(named_struct('a', 3, 'b', 4), named_struct('a', 5, 'b', 6))),
(3, array())
as data(c1, c2)
)
);
22/10/30 17:34:42 ERROR Executor: Exception in task 1.0 in stage 8.0 (TID
14)
java.lang.NullPointerException
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.generate_doConsume_1$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.generate_doConsume_0$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)
at
org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:364)
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit test.
Closes #38440 from bersprockets/SPARK-40963.
Authored-by: Bruce Robbins
Signed-off-by: Hyukjin Kwon
(cherry picked from commit 90d31541fb0313d762cc36067060e6445c04a9b6)
Signed-off-by: Hyukjin Kwon
---
.../apache/spark/sql/catalyst/analysis/Analyzer.scala | 2 +-
.../plans/logical/basicLogicalOperators.scala | 17 +
.../org/apache/spark/sql/GeneratorFunctionSuite.scala | 19 +++
3 files changed, 29 insertions(+), 9 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 8d6261a7847..c5b2229db31 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/or
[spark] branch branch-3.3 updated: [SPARK-40918][SQL][3.3] Mismatch between FileSourceScanExec and Orc and ParquetFileFormat on producing columnar output
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new fb2bdeaa234 [SPARK-40918][SQL][3.3] Mismatch between
FileSourceScanExec and Orc and ParquetFileFormat on producing columnar output
fb2bdeaa234 is described below
commit fb2bdeaa234b76fdd1e2b6101f714891daa89af9
Author: Juliusz Sompolski
AuthorDate: Mon Oct 31 14:00:09 2022 +0800
[SPARK-40918][SQL][3.3] Mismatch between FileSourceScanExec and Orc and
ParquetFileFormat on producing columnar output
### What changes were proposed in this pull request?
We move the decision about supporting columnar output based on WSCG one
level from ParquetFileFormat / OrcFileFormat up to FileSourceScanExec, and pass
it as a new required option for ParquetFileFormat / OrcFileFormat. Now the
semantics is as follows:
* `ParquetFileFormat.supportsBatch` and `OrcFileFormat.supportsBatch`
returns whether it **can**, not necessarily **will** return columnar output.
* To return columnar output, an option `FileFormat.OPTION_RETURNING_BATCH`
needs to be passed to `buildReaderWithPartitionValues` in these two file
formats. It should only be set to `true` if `supportsBatch` is also `true`, but
it can be set to `false` if we don't want columnar output nevertheless - this
way, `FileSourceScanExec` can set it to false when there are more than 100
columsn for WSCG, and `ParquetFileFormat` / `OrcFileFormat` doesn't have to
concern itself about WSCG limits.
* To avoid not passing it by accident, this option is made required. Making
it required requires updating a few places that use it, but an error resulting
from this is very obscure. It's better to fail early and explicitly here.
### Why are the changes needed?
This explains it for `ParquetFileFormat`. `OrcFileFormat` had exactly the
same issue.
`java.lang.ClassCastException:
org.apache.spark.sql.vectorized.ColumnarBatch cannot be cast to
org.apache.spark.sql.catalyst.InternalRow` was being thrown because
ParquetReader was outputting columnar batches, while FileSourceScanExec
expected row output.
The mismatch comes from the fact that `ParquetFileFormat.supportBatch`
depends on `WholeStageCodegenExec.isTooManyFields(conf, schema)`, where the
threshold is 100 fields.
When this is used in `FileSourceScanExec`:
```
override lazy val supportsColumnar: Boolean = {
relation.fileFormat.supportBatch(relation.sparkSession, schema)
}
```
the `schema` comes from output attributes, which includes extra metadata
attributes.
However, inside `ParquetFileFormat.buildReaderWithPartitionValues` it was
calculated again as
```
relation.fileFormat.buildReaderWithPartitionValues(
sparkSession = relation.sparkSession,
dataSchema = relation.dataSchema,
partitionSchema = relation.partitionSchema,
requiredSchema = requiredSchema,
filters = pushedDownFilters,
options = options,
hadoopConf = hadoopConf
...
val resultSchema = StructType(requiredSchema.fields ++
partitionSchema.fields)
...
val returningBatch = supportBatch(sparkSession, resultSchema)
```
Where `requiredSchema` and `partitionSchema` wouldn't include the metadata
columns:
```
FileSourceScanExec: output: List(c1#4608L, c2#4609L, ..., c100#4707L,
file_path#6388)
FileSourceScanExec: dataSchema:
StructType(StructField(c1,LongType,true),StructField(c2,LongType,true),...,StructField(c100,LongType,true))
FileSourceScanExec: partitionSchema: StructType()
FileSourceScanExec: requiredSchema:
StructType(StructField(c1,LongType,true),StructField(c2,LongType,true),...,StructField(c100,LongType,true))
```
Column like `file_path#6388` are added by the scan, and contain metadata
added by the scan, not by the file reader which concerns itself with what is
within the file.
### Does this PR introduce _any_ user-facing change?
Not a public API change, but it is now required to pass
`FileFormat.OPTION_RETURNING_BATCH` in `options` to
`ParquetFileFormat.buildReaderWithPartitionValues`. The only user of this API
in Apache Spark is `FileSourceScanExec`.
### How was this patch tested?
Tests added
Backports #38397 from juliuszsompolski/SPARK-40918.
Authored-by: Juliusz Sompolski
Signed-off-by: Wenchen Fan
Closes #38431 from juliuszsompolski/SPARK-40918-3.3.
Authored-by: Juliusz Sompolski
Signed-off-by: Wenchen Fan
---
.../spark/sql/execution/DataSourceScanExec.scala | 11 +--
.../sql/execution/datasources/FileFormat.scala | 13
.../execution/datasources/orc/OrcFileFormat.scala | 33 +---
.../datasou
[spark] branch master updated (90d31541fb0 -> 3c967f06e6c)
This is an automated email from the ASF dual-hosted git repository. maxgekk pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git from 90d31541fb0 [SPARK-40963][SQL] Set nullable correctly in project created by `ExtractGenerator` add 3c967f06e6c [SPARK-40967][SQL] Migrate `failAnalysis()` onto error classes No new revisions were added by this update. Summary of changes: core/src/main/resources/error/error-classes.json | 165 ++ .../org/apache/spark/sql/AnalysisException.scala | 14 + .../spark/sql/catalyst/analysis/Analyzer.scala | 24 +- .../sql/catalyst/analysis/CheckAnalysis.scala | 83 +- .../catalyst/analysis/ResolveInlineTables.scala| 20 +- .../catalyst/analysis/higherOrderFunctions.scala | 10 +- .../spark/sql/catalyst/analysis/package.scala | 24 +- .../main/scala/org/apache/spark/sql/Dataset.scala |9 +- .../spark/sql/execution/datasources/rules.scala|9 +- .../resources/sql-tests/results/ansi/array.sql.out | 32 +- .../results/ansi/higher-order-functions.sql.out| 14 +- .../test/resources/sql-tests/results/array.sql.out | 32 +- .../test/resources/sql-tests/results/count.sql.out | 16 +- .../results/higher-order-functions.sql.out | 14 +- .../sql-tests/results/inline-table.sql.out | 74 +- .../sql-tests/results/postgreSQL/boolean.sql.out | 19 +- .../sql-tests/results/postgreSQL/numeric.sql.out | 32 +- .../results/postgreSQL/window_part3.sql.out| 51 +- .../results/postgreSQL/window_part4.sql.out| 19 +- .../results/subquery/in-subquery/in-basic.sql.out | 24 +- .../negative-cases/subq-input-typecheck.sql.out| 72 +- .../results/table-valued-functions.sql.out | 14 +- .../typeCoercion/native/caseWhenCoercion.sql.out | 1120 +- .../results/typeCoercion/native/ifCoercion.sql.out | 1120 +- .../typeCoercion/native/inConversion.sql.out | 2240 ++-- .../sql-tests/results/udf/udf-inline-table.sql.out | 74 +- .../execution/command/PlanResolutionSuite.scala| 22 +- 27 files changed, 4915 insertions(+), 432 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated: [SPARK-40713][CONNECT] Improve SET operation support in the proto and the server
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 31bff52431d [SPARK-40713][CONNECT] Improve SET operation support in
the proto and the server
31bff52431d is described below
commit 31bff52431d25dd36253d3cafa17c233ff6a54f2
Author: Rui Wang
AuthorDate: Mon Oct 31 14:24:38 2022 +0800
[SPARK-40713][CONNECT] Improve SET operation support in the proto and the
server
### What changes were proposed in this pull request?
1. Rename `Union` to `SetOperation` and make it cover `except`,
`intersect`, `union`.
2. Improve server side support for `SetOperation`.
### Why are the changes needed?
Improve SET operation support in the proto and the server.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UT
Closes #38166 from amaliujia/SPARK-40713.
Authored-by: Rui Wang
Signed-off-by: Wenchen Fan
---
.../main/protobuf/spark/connect/relations.proto| 24 +++--
.../org/apache/spark/sql/connect/dsl/package.scala | 47 ++
.../sql/connect/planner/SparkConnectPlanner.scala | 40 ++--
.../connect/planner/SparkConnectPlannerSuite.scala | 39 ++--
.../connect/planner/SparkConnectProtoSuite.scala | 36 +++
python/pyspark/sql/connect/plan.py | 6 +-
python/pyspark/sql/connect/proto/relations_pb2.py | 104 ++---
python/pyspark/sql/connect/proto/relations_pb2.pyi | 95 ---
8 files changed, 279 insertions(+), 112 deletions(-)
diff --git a/connector/connect/src/main/protobuf/spark/connect/relations.proto
b/connector/connect/src/main/protobuf/spark/connect/relations.proto
index 94010487ee5..8421535dd8e 100644
--- a/connector/connect/src/main/protobuf/spark/connect/relations.proto
+++ b/connector/connect/src/main/protobuf/spark/connect/relations.proto
@@ -35,7 +35,7 @@ message Relation {
Project project = 3;
Filter filter = 4;
Join join = 5;
-Union union = 6;
+SetOperation set_op = 6;
Sort sort = 7;
Limit limit = 8;
Aggregate aggregate = 9;
@@ -127,15 +127,19 @@ message Join {
}
}
-// Relation of type [[Union]], at least one input must be set.
-message Union {
- repeated Relation inputs = 1;
- UnionType union_type = 2;
-
- enum UnionType {
-UNION_TYPE_UNSPECIFIED = 0;
-UNION_TYPE_DISTINCT = 1;
-UNION_TYPE_ALL = 2;
+// Relation of type [[SetOperation]]
+message SetOperation {
+ Relation left_input = 1;
+ Relation right_input = 2;
+ SetOpType set_op_type = 3;
+ bool is_all = 4;
+ bool by_name = 5;
+
+ enum SetOpType {
+SET_OP_TYPE_UNSPECIFIED = 0;
+SET_OP_TYPE_INTERSECT = 1;
+SET_OP_TYPE_UNION = 2;
+SET_OP_TYPE_EXCEPT = 3;
}
}
diff --git
a/connector/connect/src/main/scala/org/apache/spark/sql/connect/dsl/package.scala
b/connector/connect/src/main/scala/org/apache/spark/sql/connect/dsl/package.scala
index a9a97e740d8..9ffc4c4a1fe 100644
---
a/connector/connect/src/main/scala/org/apache/spark/sql/connect/dsl/package.scala
+++
b/connector/connect/src/main/scala/org/apache/spark/sql/connect/dsl/package.scala
@@ -21,6 +21,7 @@ import scala.language.implicitConversions
import org.apache.spark.connect.proto
import org.apache.spark.connect.proto.Join.JoinType
+import org.apache.spark.connect.proto.SetOperation.SetOpType
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.connect.planner.DataTypeProtoConverter
@@ -322,6 +323,52 @@ package object dsl {
// resolution only by name in the analyzer.
proto.Relation.newBuilder().setAggregate(agg.build()).build()
}
+
+ def except(otherPlan: proto.Relation, isAll: Boolean): proto.Relation = {
+proto.Relation
+ .newBuilder()
+ .setSetOp(
+createSetOperation(logicalPlan, otherPlan,
SetOpType.SET_OP_TYPE_EXCEPT, isAll))
+ .build()
+ }
+
+ def intersect(otherPlan: proto.Relation, isAll: Boolean): proto.Relation
=
+proto.Relation
+ .newBuilder()
+ .setSetOp(
+createSetOperation(logicalPlan, otherPlan,
SetOpType.SET_OP_TYPE_INTERSECT, isAll))
+ .build()
+
+ def union(
+ otherPlan: proto.Relation,
+ isAll: Boolean = true,
+ byName: Boolean = false): proto.Relation =
+proto.Relation
+ .newBuilder()
+ .setSetOp(
+createSetOperation(
+ logicalPlan,
+ otherPlan,
+ SetOpType.SET_OP_TYPE_UNION,
+ isAll,
+ byName))
+ .build()
+
+ private def createSetOperation(
+ left: proto.Relation,
+ right: proto.Relation,
+ t: SetOpType,
+ isAll: Boolean = true,
+ byName: Boolean =
