[spark] branch master updated (72cce5c39da -> 1c55e94d6a9)
This is an automated email from the ASF dual-hosted git repository. yao pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git from 72cce5c39da [SPARK-40872] Fallback to original shuffle block when a push-merged shuffle chunk is zero-size add 1c55e94d6a9 [SPARK-41211][BUILD] Upgrade ZooKeeper from 3.6.2 to 3.6.3 No new revisions were added by this update. Summary of changes: dev/deps/spark-deps-hadoop-2-hive-2.3 | 4 ++-- dev/deps/spark-deps-hadoop-3-hive-2.3 | 4 ++-- pom.xml | 2 +- 3 files changed, 5 insertions(+), 5 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-41154][SQL] Incorrect relation caching for queries with time travel spec
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 07427b854be [SPARK-41154][SQL] Incorrect relation caching for queries with time travel spec 07427b854be is described below commit 07427b854be58810bd485c00c5e5c576d5aa404e Author: ulysses-you AuthorDate: Mon Nov 21 18:19:10 2022 +0800 [SPARK-41154][SQL] Incorrect relation caching for queries with time travel spec ### What changes were proposed in this pull request? Add TimeTravelSpec to the key of relation cache in AnalysisContext. ### Why are the changes needed? Correct the relation resolution for the same table but different TimeTravelSpec. ### Does this PR introduce _any_ user-facing change? yes, bug fix ### How was this patch tested? add test Closes #38687 from ulysses-you/time-travel-spec. Authored-by: ulysses-you Signed-off-by: Wenchen Fan --- .../apache/spark/sql/catalyst/analysis/Analyzer.scala | 11 ++- .../spark/sql/connector/DataSourceV2SQLSuite.scala | 17 + 2 files changed, 23 insertions(+), 5 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 104c5c1e080..8cdf83f4be7 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 @@ -113,9 +113,9 @@ object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog { * @param nestedViewDepth The nested depth in the view resolution, this enables us to limit the *depth of nested views. * @param maxNestedViewDepth The maximum allowed depth of nested view resolution. - * @param relationCache A mapping from qualified table names to resolved relations. This can ensure - * that the table is resolved only once if a table is used multiple times - * in a query. + * @param relationCache A mapping from qualified table names and time travel spec to resolved + * relations. This can ensure that the table is resolved only once if a table + * is used multiple times in a query. * @param referredTempViewNames All the temp view names referred by the current view we are * resolving. It's used to make sure the relation resolution is * consistent between view creation and view resolution. For example, @@ -129,7 +129,8 @@ case class AnalysisContext( catalogAndNamespace: Seq[String] = Nil, nestedViewDepth: Int = 0, maxNestedViewDepth: Int = -1, -relationCache: mutable.Map[Seq[String], LogicalPlan] = mutable.Map.empty, +relationCache: mutable.Map[(Seq[String], Option[TimeTravelSpec]), LogicalPlan] = + mutable.Map.empty, referredTempViewNames: Seq[Seq[String]] = Seq.empty, // 1. If we are resolving a view, this field will be restored from the view metadata, //by calling `AnalysisContext.withAnalysisContext(viewDesc)`. @@ -1239,7 +1240,7 @@ class Analyzer(override val catalogManager: CatalogManager) resolveTempView(u.multipartIdentifier, u.isStreaming, timeTravelSpec.isDefined).orElse { expandIdentifier(u.multipartIdentifier) match { case CatalogAndIdentifier(catalog, ident) => -val key = catalog.name +: ident.namespace :+ ident.name +val key = ((catalog.name +: ident.namespace :+ ident.name).toSeq, timeTravelSpec) AnalysisContext.get.relationCache.get(key).map(_.transform { case multi: MultiInstanceRelation => val newRelation = multi.newInstance() diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala index de8612c3348..ea93367b3db 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala @@ -2775,6 +2775,23 @@ class DataSourceV2SQLSuiteV1Filter extends DataSourceV2SQLSuite with AlterTableT } } + test("SPARK-41154: Incorrect relation caching for queries with time travel spec") { +sql("use testcat") +val t1 = "testcat.t1" +val t2 = "testcat.t2" +withTable(t1, t2) { + sql(s"CREATE TABLE $t1 USING foo AS SELECT 1 as c") + sql(s"CREATE TABLE $t2 USING foo AS SELECT 2 as c") + assert( +sql(""" + |SELECT * FROM t VERSION AS OF '1' + |UNION ALL + |SELECT * F
[spark] branch master updated (07427b854be -> d9abbfe5f71)
This is an automated email from the ASF dual-hosted git repository. gengliang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git from 07427b854be [SPARK-41154][SQL] Incorrect relation caching for queries with time travel spec add d9abbfe5f71 [SPARK-41195][SQL] Support PIVOT/UNPIVOT with join children No new revisions were added by this update. Summary of changes: docs/sql-ref-syntax-qry-select.md | 2 + .../spark/sql/catalyst/parser/SqlBaseParser.g4 | 8 +- .../spark/sql/catalyst/parser/AstBuilder.scala | 105 --- .../spark/sql/errors/QueryParsingErrors.scala | 8 +- .../sql/catalyst/parser/UnpivotParserSuite.scala | 148 + 5 files changed, 219 insertions(+), 52 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-41005][COLLECT][FOLLOWUP] Remove JSON code path and use `RDD.collect` in Arrow code path
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 908adca2b22 [SPARK-41005][COLLECT][FOLLOWUP] Remove JSON code path and use `RDD.collect` in Arrow code path 908adca2b22 is described below commit 908adca2b229b05b2ae0dd31cbaaa1fdcde16290 Author: Ruifeng Zheng AuthorDate: Tue Nov 22 09:43:44 2022 +0900 [SPARK-41005][COLLECT][FOLLOWUP] Remove JSON code path and use `RDD.collect` in Arrow code path ### What changes were proposed in this pull request? 1, Remove JSON code path; 2, use RDD.collect in Arrow code path, since existing tests were already broken in Arrow code path; 3, reenable `test_fill_na` ### Why are the changes needed? existing Arrow code path is still problematic and it fails and fallback to JSON code path, which change the output datatypes of `test_fill_na` ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? reenabled test and added UT Closes #38706 from zhengruifeng/collect_disable_json. Authored-by: Ruifeng Zheng Signed-off-by: Hyukjin Kwon --- .../src/main/protobuf/spark/connect/base.proto | 14 +- .../service/SparkConnectStreamHandler.scala| 156 ++--- python/pyspark/sql/connect/client.py | 5 - python/pyspark/sql/connect/proto/base_pb2.py | 41 ++ python/pyspark/sql/connect/proto/base_pb2.pyi | 51 +-- .../sql/tests/connect/test_connect_basic.py| 55 +++- 6 files changed, 82 insertions(+), 240 deletions(-) diff --git a/connector/connect/src/main/protobuf/spark/connect/base.proto b/connector/connect/src/main/protobuf/spark/connect/base.proto index 66e27187153..277da6b2431 100644 --- a/connector/connect/src/main/protobuf/spark/connect/base.proto +++ b/connector/connect/src/main/protobuf/spark/connect/base.proto @@ -139,11 +139,7 @@ message ExecutePlanRequest { message ExecutePlanResponse { string client_id = 1; - // Result type - oneof result_type { -ArrowBatch arrow_batch = 2; -JSONBatch json_batch = 3; - } + ArrowBatch arrow_batch = 2; // Metrics for the query execution. Typically, this field is only present in the last // batch of results and then represent the overall state of the query execution. @@ -155,14 +151,6 @@ message ExecutePlanResponse { bytes data = 2; } - // Message type when the result is returned as JSON. This is essentially a bulk wrapper - // for the JSON result of a Spark DataFrame. All rows are returned in the JSON record format - // of `{col -> row}`. - message JSONBatch { -int64 row_count = 1; -bytes data = 2; - } - message Metrics { repeated MetricObject metrics = 1; diff --git a/connector/connect/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala b/connector/connect/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala index 50ff08f997c..092bdd00dc1 100644 --- a/connector/connect/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala +++ b/connector/connect/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala @@ -18,12 +18,10 @@ package org.apache.spark.sql.connect.service import scala.collection.JavaConverters._ -import scala.util.control.NonFatal import com.google.protobuf.ByteString import io.grpc.stub.StreamObserver -import org.apache.spark.SparkException import org.apache.spark.connect.proto import org.apache.spark.connect.proto.{ExecutePlanRequest, ExecutePlanResponse} import org.apache.spark.internal.Logging @@ -34,7 +32,6 @@ import org.apache.spark.sql.execution.{SparkPlan, SQLExecution} import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AdaptiveSparkPlanHelper, QueryStageExec} import org.apache.spark.sql.execution.arrow.ArrowConverters import org.apache.spark.sql.types.StructType -import org.apache.spark.util.ThreadUtils class SparkConnectStreamHandler(responseObserver: StreamObserver[ExecutePlanResponse]) extends Logging { @@ -57,75 +54,7 @@ class SparkConnectStreamHandler(responseObserver: StreamObserver[ExecutePlanResp // Extract the plan from the request and convert it to a logical plan val planner = new SparkConnectPlanner(session) val dataframe = Dataset.ofRows(session, planner.transformRelation(request.getPlan.getRoot)) -try { - processAsArrowBatches(request.getClientId, dataframe) -} catch { - case e: Exception => -logWarning(e.getMessage) -processAsJsonBatches(request.getClientId, dataframe) -} - } - - def processAsJsonBatches(clientId: String, dataframe: DataFrame): Unit = { -// Only process up to 10MB of data. -val sb = new StringBuilder -
[spark] branch master updated: [SPARK-41151][SQL] Keep built-in file `_metadata` column nullable value consistent
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 77a8fb445cb [SPARK-41151][SQL] Keep built-in file `_metadata` column nullable value consistent 77a8fb445cb is described below commit 77a8fb445cb099e11ed486447959b3de0a625b6d Author: yaohua AuthorDate: Tue Nov 22 09:54:14 2022 +0900 [SPARK-41151][SQL] Keep built-in file `_metadata` column nullable value consistent ### What changes were proposed in this pull request? In FileSourceStrategy, we add an Alias node to wrap the file metadata fields (e.g. file_name, file_size) in a NamedStruct ([here](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala#L279)). But `CreateNamedStruct` has an override `nullable` value `false` ([here](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypeCreator.scala#L44 [...] This PR fixes this, by changing `_metadata` column to be always not nullable. Rationale: 1. By definition, `_metadata` for file-based sources is always not null; 2. If users have already persisted this nullable `_metadata` somewhere, then it's totally fine to write non-nullable data to this nullable column. ### Why are the changes needed? For stateful streaming, we store the schema in the state store and [check consistency across batches](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateSchemaCompatibilityChecker.scala#L47). To avoid state schema compatibility mismatched, we should keep nullable consistent in `_metadata`. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? New UT Closes #38683 from Yaohua628/spark-41151. Authored-by: yaohua Signed-off-by: Jungtaek Lim --- .../sql/catalyst/expressions/namedExpressions.scala | 5 +++-- .../sql/execution/datasources/FileSourceStrategy.scala | 6 +- .../execution/datasources/FileMetadataStructSuite.scala | 17 - 3 files changed, 24 insertions(+), 4 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala index 99e5f411bdb..8dd28e9aaae 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala @@ -464,8 +464,9 @@ object FileSourceMetadataAttribute { val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col" - def apply(name: String, dataType: DataType, nullable: Boolean = true): AttributeReference = -AttributeReference(name, dataType, nullable, + def apply(name: String, dataType: DataType): AttributeReference = +// Metadata column for file sources is always not nullable. +AttributeReference(name, dataType, nullable = false, new MetadataBuilder() .putBoolean(METADATA_COL_ATTR_KEY, value = true) .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true).build())() diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala index 576801d3dd5..476d6579b38 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala @@ -275,8 +275,12 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging { .get.withName(FileFormat.ROW_INDEX) } } +// SPARK-41151: metadata column is not nullable for file sources. +// Here, we *explicitly* enforce the not null to `CreateStruct(structColumns)` +// to avoid any risk of inconsistent schema nullability val metadataAlias = - Alias(CreateStruct(structColumns), METADATA_NAME)(exprId = metadataStruct.exprId) + Alias(KnownNotNull(CreateStruct(structColumns)), +METADATA_NAME)(exprId = metadataStruct.exprId) execution.ProjectExec( readDataColumns ++ partitionColumns :+ metadataAlias, scan) }.getOrElse(scan) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala index e0e208b62f1..a39a36a4f83 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/data
[spark] branch master updated: [SPARK-41196][CONNECT][INFRA][FOLLOW-UP] Change protobuf versions in CI
This is an automated email from the ASF dual-hosted git repository. ruifengz 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 1734d9b1a1e [SPARK-41196][CONNECT][INFRA][FOLLOW-UP] Change protobuf versions in CI 1734d9b1a1e is described below commit 1734d9b1a1e72153b72687b643ae1597ff15c2cc Author: Ruifeng Zheng AuthorDate: Tue Nov 22 09:12:03 2022 +0800 [SPARK-41196][CONNECT][INFRA][FOLLOW-UP] Change protobuf versions in CI ### What changes were proposed in this pull request? pin `protobuf==3.19.4` in tests ### Why are the changes needed? versions were already changed in https://github.com/apache/spark/pull/38693 ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? updated CI Closes #38729 from zhengruifeng/connect_infra_protobuf. Authored-by: Ruifeng Zheng Signed-off-by: Ruifeng Zheng --- .github/workflows/build_and_test.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 0521270cee7..646a096b72d 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -243,7 +243,7 @@ jobs: - name: Install Python packages (Python 3.8) if: (contains(matrix.modules, 'sql') && !contains(matrix.modules, 'sql-')) run: | -python3.8 -m pip install 'numpy>=1.20.0' pyarrow pandas scipy unittest-xml-reporting 'grpcio==1.48.1' 'protobuf==4.21.6' +python3.8 -m pip install 'numpy>=1.20.0' pyarrow pandas scipy unittest-xml-reporting 'grpcio==1.48.1' 'protobuf==3.19.4' python3.8 -m pip list # Run the tests. - name: Run tests @@ -589,7 +589,7 @@ jobs: # See also https://issues.apache.org/jira/browse/SPARK-38279. python3.9 -m pip install 'sphinx<3.1.0' mkdocs pydata_sphinx_theme ipython nbsphinx numpydoc 'jinja2<3.0.0' 'markupsafe==2.0.1' 'pyzmq<24.0.0' python3.9 -m pip install ipython_genutils # See SPARK-38517 -python3.9 -m pip install sphinx_plotly_directive 'numpy>=1.20.0' pyarrow pandas 'plotly>=4.8' 'grpcio==1.48.1' 'protobuf==4.21.6' 'mypy-protobuf==3.3.0' +python3.9 -m pip install sphinx_plotly_directive 'numpy>=1.20.0' pyarrow pandas 'plotly>=4.8' 'grpcio==1.48.1' 'protobuf==3.19.4' 'mypy-protobuf==3.3.0' python3.9 -m pip install 'docutils<0.18.0' # See SPARK-39421 apt-get update -y apt-get install -y ruby ruby-dev - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch branch-3.3 updated: [SPARK-41154][SQL][3.3] Incorrect relation caching for queries with time travel spec
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 ace3c69ef18 [SPARK-41154][SQL][3.3] Incorrect relation caching for queries with time travel spec ace3c69ef18 is described below commit ace3c69ef18d648bb15855c40c8b7e44987dede4 Author: ulysses-you AuthorDate: Tue Nov 22 10:07:17 2022 +0800 [SPARK-41154][SQL][3.3] Incorrect relation caching for queries with time travel spec backport https://github.com/apache/spark/pull/38687 for branch-3.3 ### What changes were proposed in this pull request? Add TimeTravelSpec to the key of relation cache in AnalysisContext. ### Why are the changes needed? Correct the relation resolution for the same table but different TimeTravelSpec. ### Does this PR introduce _any_ user-facing change? yes, bug fix ### How was this patch tested? add test Closes #38741 from ulysses-you/time-travel-spec-3.3. Authored-by: ulysses-you Signed-off-by: Wenchen Fan --- .../apache/spark/sql/catalyst/analysis/Analyzer.scala | 11 ++- .../spark/sql/connector/DataSourceV2SQLSuite.scala | 17 + 2 files changed, 23 insertions(+), 5 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 2a2fe6f2957..0c68dd8839d 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 @@ -112,9 +112,9 @@ object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog { * @param nestedViewDepth The nested depth in the view resolution, this enables us to limit the *depth of nested views. * @param maxNestedViewDepth The maximum allowed depth of nested view resolution. - * @param relationCache A mapping from qualified table names to resolved relations. This can ensure - * that the table is resolved only once if a table is used multiple times - * in a query. + * @param relationCache A mapping from qualified table names and time travel spec to resolved + * relations. This can ensure that the table is resolved only once if a table + * is used multiple times in a query. * @param referredTempViewNames All the temp view names referred by the current view we are * resolving. It's used to make sure the relation resolution is * consistent between view creation and view resolution. For example, @@ -128,7 +128,8 @@ case class AnalysisContext( catalogAndNamespace: Seq[String] = Nil, nestedViewDepth: Int = 0, maxNestedViewDepth: Int = -1, -relationCache: mutable.Map[Seq[String], LogicalPlan] = mutable.Map.empty, +relationCache: mutable.Map[(Seq[String], Option[TimeTravelSpec]), LogicalPlan] = + mutable.Map.empty, referredTempViewNames: Seq[Seq[String]] = Seq.empty, // 1. If we are resolving a view, this field will be restored from the view metadata, //by calling `AnalysisContext.withAnalysisContext(viewDesc)`. @@ -1188,7 +1189,7 @@ class Analyzer(override val catalogManager: CatalogManager) lookupTempView(u.multipartIdentifier, u.isStreaming, timeTravelSpec.isDefined).orElse { expandIdentifier(u.multipartIdentifier) match { case CatalogAndIdentifier(catalog, ident) => -val key = catalog.name +: ident.namespace :+ ident.name +val key = ((catalog.name +: ident.namespace :+ ident.name).toSeq, timeTravelSpec) AnalysisContext.get.relationCache.get(key).map(_.transform { case multi: MultiInstanceRelation => val newRelation = multi.newInstance() diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala index 44f97f55713..7470911c9e5 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala @@ -2576,6 +2576,23 @@ class DataSourceV2SQLSuite } } + test("SPARK-41154: Incorrect relation caching for queries with time travel spec") { +sql("use testcat") +val t1 = "testcat.t1" +val t2 = "testcat.t2" +withTable(t1, t2) { + sql(s"CREATE TABLE $t1 USING foo AS SELECT 1 as c") + sql(s"CREATE TABLE $t2 USING foo AS SELECT 2 as c") + assert( +sql(""" + |SELECT * FROM t VERSION AS OF '1' +
[spark] branch master updated: [SPARK-41017][SQL][FOLLOWUP] Push Filter with both deterministic and nondeterministic predicates
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 0cdbda13b08 [SPARK-41017][SQL][FOLLOWUP] Push Filter with both deterministic and nondeterministic predicates 0cdbda13b08 is described below commit 0cdbda13b08ad79c91d1d7d5912fb1120191dc56 Author: Wenchen Fan AuthorDate: Tue Nov 22 10:40:24 2022 +0800 [SPARK-41017][SQL][FOLLOWUP] Push Filter with both deterministic and nondeterministic predicates ### What changes were proposed in this pull request? This PR fixes a regression caused by https://github.com/apache/spark/pull/38511 . For `FROM t WHERE rand() > 0.5 AND col = 1`, we can still push down `col = 1` because we don't guarantee the predicates evaluation order within a `Filter`. This PR updates `ScanOperation` to consider this case and bring back the previous pushdown behavior. ### Why are the changes needed? fix perf regression ### Does this PR introduce _any_ user-facing change? no ### How was this patch tested? new tests Closes #38746 from cloud-fan/filter. Lead-authored-by: Wenchen Fan Co-authored-by: Wenchen Fan Signed-off-by: Wenchen Fan --- .../spark/sql/catalyst/planning/patterns.scala | 21 - .../spark/sql/FileBasedDataSourceSuite.scala | 22 -- 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala index 3c35ba9b600..bbda9eb76b1 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala @@ -137,11 +137,22 @@ object ScanOperation extends OperationHelper { val alwaysInline = SQLConf.get.getConf(SQLConf.COLLAPSE_PROJECT_ALWAYS_INLINE) val (fields, filters, child, _) = collectProjectsAndFilters(plan, alwaysInline) // `collectProjectsAndFilters` transforms the plan bottom-up, so the bottom-most filter are -// placed at the beginning of `filters` list. According to the SQL semantic, we can only -// push down the bottom deterministic filters. -val filtersCanPushDown = filters.takeWhile(_.deterministic).flatMap(splitConjunctivePredicates) -val filtersStayUp = filters.dropWhile(_.deterministic) -Some((fields.getOrElse(child.output), filtersStayUp, filtersCanPushDown, child)) +// placed at the beginning of `filters` list. According to the SQL semantic, we cannot merge +// Filters if one or more of them are nondeterministic. This means we can only push down the +// bottom-most Filter, or more following deterministic Filters if the bottom-most Filter is +// also deterministic. +if (filters.isEmpty) { + Some((fields.getOrElse(child.output), Nil, Nil, child)) +} else if (filters.head.deterministic) { + val filtersCanPushDown = filters.takeWhile(_.deterministic) +.flatMap(splitConjunctivePredicates) + val filtersStayUp = filters.dropWhile(_.deterministic) + Some((fields.getOrElse(child.output), filtersStayUp, filtersCanPushDown, child)) +} else { + val filtersCanPushDown = splitConjunctivePredicates(filters.head) + val filtersStayUp = filters.drop(1) + Some((fields.getOrElse(child.output), filtersStayUp, filtersCanPushDown, child)) +} } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala index 98cb54ccbbc..cf5f8d990f7 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala @@ -30,10 +30,10 @@ import org.apache.hadoop.fs.{LocalFileSystem, Path} import org.apache.spark.SparkException import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.TestingUDT.{IntervalUDT, NullData, NullUDT} -import org.apache.spark.sql.catalyst.expressions.AttributeReference +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GreaterThan, Literal} import org.apache.spark.sql.catalyst.expressions.IntegralLiteralTestUtils.{negativeInt, positiveInt} import org.apache.spark.sql.catalyst.plans.logical.Filter -import org.apache.spark.sql.execution.SimpleMode +import org.apache.spark.sql.execution.{FileSourceScanLike, SimpleMode} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.execution.datasources.FilePartition import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileScan} @@ -1074,6
[spark] branch branch-3.3 updated (ace3c69ef18 -> abc343fcab8)
This is an automated email from the ASF dual-hosted git repository. kabhwan pushed a change to branch branch-3.3 in repository https://gitbox.apache.org/repos/asf/spark.git from ace3c69ef18 [SPARK-41154][SQL][3.3] Incorrect relation caching for queries with time travel spec add abc343fcab8 [SPARK-41151][SQL][3.3] Keep built-in file `_metadata` column nullable value consistent No new revisions were added by this update. Summary of changes: .../sql/catalyst/expressions/namedExpressions.scala | 5 +++-- .../sql/execution/datasources/FileSourceStrategy.scala | 6 +- .../execution/datasources/FileMetadataStructSuite.scala | 17 - 3 files changed, 24 insertions(+), 4 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-41213][CONNECT][PYTHON] Implement `DataFrame.__repr__` and `DataFrame.dtypes`
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 55addb38f2a [SPARK-41213][CONNECT][PYTHON] Implement `DataFrame.__repr__` and `DataFrame.dtypes` 55addb38f2a is described below commit 55addb38f2a9cc2521a2e7908bb8ad3d49c9a8bb Author: Ruifeng Zheng AuthorDate: Tue Nov 22 15:44:28 2022 +0900 [SPARK-41213][CONNECT][PYTHON] Implement `DataFrame.__repr__` and `DataFrame.dtypes` ### What changes were proposed in this pull request? Implement `DataFrame.__repr__` and `DataFrame.dtypes` ### Why are the changes needed? For api coverage ### Does this PR introduce _any_ user-facing change? yes ### How was this patch tested? added UT Closes #38735 from zhengruifeng/connect_df_repr. Authored-by: Ruifeng Zheng Signed-off-by: Hyukjin Kwon --- python/pyspark/sql/connect/dataframe.py | 19 ++- .../pyspark/sql/tests/connect/test_connect_basic.py | 13 ++--- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/python/pyspark/sql/connect/dataframe.py b/python/pyspark/sql/connect/dataframe.py index 15aa028b11b..275a6d2668d 100644 --- a/python/pyspark/sql/connect/dataframe.py +++ b/python/pyspark/sql/connect/dataframe.py @@ -115,6 +115,9 @@ class DataFrame(object): self._cache: Dict[str, Any] = {} self._session: "RemoteSparkSession" = session +def __repr__(self) -> str: +return "DataFrame[%s]" % (", ".join("%s: %s" % c for c in self.dtypes)) + @classmethod def withPlan(cls, plan: plan.LogicalPlan, session: "RemoteSparkSession") -> "DataFrame": """Main initialization method used to construct a new data frame with a child plan.""" @@ -137,13 +140,26 @@ class DataFrame(object): def colRegex(self, regex: str) -> "DataFrame": ... +@property +def dtypes(self) -> List[Tuple[str, str]]: +"""Returns all column names and their data types as a list. + +.. versionadded:: 3.4.0 + +Returns +--- +list +List of columns as tuple pairs. +""" +return [(str(f.name), f.dataType.simpleString()) for f in self.schema.fields] + @property def columns(self) -> List[str]: """Returns the list of columns of the current data frame.""" if self._plan is None: return [] -return self.schema().names +return self.schema.names def sparkSession(self) -> "RemoteSparkSession": """Returns Spark session that created this :class:`DataFrame`. @@ -736,6 +752,7 @@ class DataFrame(object): query = self._plan.to_proto(self._session) return self._session._to_pandas(query) +@property def schema(self) -> StructType: """Returns the schema of this :class:`DataFrame` as a :class:`pyspark.sql.types.StructType`. diff --git a/python/pyspark/sql/tests/connect/test_connect_basic.py b/python/pyspark/sql/tests/connect/test_connect_basic.py index 9e7a5f2f4a5..917cce0ebb0 100644 --- a/python/pyspark/sql/tests/connect/test_connect_basic.py +++ b/python/pyspark/sql/tests/connect/test_connect_basic.py @@ -137,7 +137,7 @@ class SparkConnectTests(SparkConnectSQLTestCase): self.assertGreater(len(result), 0) def test_schema(self): -schema = self.connect.read.table(self.tbl_name).schema() +schema = self.connect.read.table(self.tbl_name).schema self.assertEqual( StructType( [StructField("id", LongType(), True), StructField("name", StringType(), True)] @@ -333,6 +333,14 @@ class SparkConnectTests(SparkConnectSQLTestCase): expected = "+---+---+\n| X| Y|\n+---+---+\n| 1| 2|\n+---+---+\n" self.assertEqual(show_str, expected) +def test_repr(self): +# SPARK-41213: Test the __repr__ method +query = """SELECT * FROM VALUES (1L, NULL), (3L, "Z") AS tab(a, b)""" +self.assertEqual( +self.connect.sql(query).__repr__(), +self.spark.sql(query).__repr__(), +) + def test_explain_string(self): # SPARK-41122: test explain API. plan_str = self.connect.sql("SELECT 1").explain(extended=True) @@ -380,8 +388,7 @@ class SparkConnectTests(SparkConnectSQLTestCase): col0 = ( self.connect.range(1, 10) .select(col("id").alias("name", metadata={"max": 99})) -.schema() -.names[0] +.schema.names[0] ) self.assertEqual("name", col0) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (55addb38f2a -> 51b04406028)
This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git from 55addb38f2a [SPARK-41213][CONNECT][PYTHON] Implement `DataFrame.__repr__` and `DataFrame.dtypes` add 51b04406028 [SPARK-41209][PYTHON] Improve PySpark type inference in _merge_type method No new revisions were added by this update. Summary of changes: python/pyspark/sql/tests/test_types.py | 46 +++--- python/pyspark/sql/types.py| 4 +++ 2 files changed, 36 insertions(+), 14 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (51b04406028 -> 8496059966e)
This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git from 51b04406028 [SPARK-41209][PYTHON] Improve PySpark type inference in _merge_type method add 8496059966e [SPARK-41193][SQL][TESTS] Ignore `collect data with single partition larger than 2GB bytes array limit` in `DatasetLargeResultCollectingSuite` No new revisions were added by this update. Summary of changes: sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala | 6 +- 1 file changed, 5 insertions(+), 1 deletion(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (8496059966e -> b38a1158870)
This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git from 8496059966e [SPARK-41193][SQL][TESTS] Ignore `collect data with single partition larger than 2GB bytes array limit` in `DatasetLargeResultCollectingSuite` add b38a1158870 [SPARK-41169][CONNECT][PYTHON] Implement `DataFrame.drop` No new revisions were added by this update. Summary of changes: .../main/protobuf/spark/connect/relations.proto| 14 ++ .../org/apache/spark/sql/connect/dsl/package.scala | 22 +++ .../sql/connect/planner/SparkConnectPlanner.scala | 16 ++- .../connect/planner/SparkConnectProtoSuite.scala | 17 +++ python/pyspark/sql/connect/dataframe.py| 19 ++- python/pyspark/sql/connect/plan.py | 43 ++ python/pyspark/sql/connect/proto/relations_pb2.py | 150 +++-- python/pyspark/sql/connect/proto/relations_pb2.pyi | 45 +++ .../sql/tests/connect/test_connect_basic.py| 27 .../sql/tests/connect/test_connect_plan_only.py| 16 +++ 10 files changed, 296 insertions(+), 73 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (b38a1158870 -> d453598a428)
This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git from b38a1158870 [SPARK-41169][CONNECT][PYTHON] Implement `DataFrame.drop` add d453598a428 [SPARK-40809][CONNECT][FOLLOW-UP] Do not use Buffer to make Scala 2.13 test pass No new revisions were added by this update. Summary of changes: python/pyspark/sql/tests/connect/test_connect_basic.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch branch-3.3 updated (abc343fcab8 -> 09000408fbb)
This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a change to branch branch-3.3 in repository https://gitbox.apache.org/repos/asf/spark.git from abc343fcab8 [SPARK-41151][SQL][3.3] Keep built-in file `_metadata` column nullable value consistent add 09000408fbb [SPARK-41118][SQL][3.3] to_number`/`try_to_number` should return `null` when format is `null` No new revisions were added by this update. Summary of changes: .../expressions/numberFormatExpressions.scala | 122 +++-- .../expressions/StringExpressionsSuite.scala | 11 ++ 2 files changed, 73 insertions(+), 60 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (d453598a428 -> 40b7d29e14c)
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 d453598a428 [SPARK-40809][CONNECT][FOLLOW-UP] Do not use Buffer to make Scala 2.13 test pass add 40b7d29e14c [SPARK-41217][SQL] Add the error class `FAILED_FUNCTION_CALL` No new revisions were added by this update. Summary of changes: core/src/main/resources/error/error-classes.json | 5 ++ .../sql/catalyst/analysis/FunctionRegistry.scala | 9 +-- .../spark/sql/errors/QueryCompilationErrors.scala | 13 ++- .../results/ansi/string-functions.sql.out | 16 +++- .../sql-tests/results/csv-functions.sql.out| 93 +- .../resources/sql-tests/results/extract.sql.out| 75 +++-- .../sql-tests/results/json-functions.sql.out | 93 +- .../sql-tests/results/postgreSQL/int8.sql.out | 2 +- .../sql-tests/results/string-functions.sql.out | 16 +++- .../results/table-valued-functions.sql.out | 2 +- 10 files changed, 269 insertions(+), 55 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (40b7d29e14c -> a80899f8bef)
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 40b7d29e14c [SPARK-41217][SQL] Add the error class `FAILED_FUNCTION_CALL` add a80899f8bef [SPARK-41206][SQL] Rename the error class `_LEGACY_ERROR_TEMP_1233` to `COLUMN_ALREADY_EXISTS` No new revisions were added by this update. Summary of changes: .../connect/planner/SparkConnectProtoSuite.scala | 12 ++-- core/src/main/resources/error/error-classes.json | 10 +-- .../spark/sql/catalyst/analysis/Analyzer.scala | 3 +- .../sql/catalyst/analysis/CheckAnalysis.scala | 1 - .../spark/sql/catalyst/analysis/ResolveUnion.scala | 2 - .../spark/sql/errors/QueryCompilationErrors.scala | 8 +-- .../apache/spark/sql/util/PartitioningUtils.scala | 3 +- .../org/apache/spark/sql/util/SchemaUtils.scala| 40 --- .../apache/spark/sql/util/SchemaUtilsSuite.scala | 52 +++--- .../main/scala/org/apache/spark/sql/Dataset.scala | 2 - .../spark/sql/execution/command/CommandCheck.scala | 2 +- .../spark/sql/execution/command/tables.scala | 1 - .../apache/spark/sql/execution/command/views.scala | 3 +- .../sql/execution/datasources/DataSource.scala | 10 +-- .../InsertIntoHadoopFsRelationCommand.scala| 1 - .../execution/datasources/PartitioningUtils.scala | 3 +- .../sql/execution/datasources/jdbc/JdbcUtils.scala | 8 +-- .../spark/sql/execution/datasources/rules.scala| 9 +-- .../sql/execution/datasources/v2/FileTable.scala | 6 +- .../sql/execution/datasources/v2/FileWrite.scala | 3 +- .../spark/sql/DataFrameSetOperationsSuite.scala| 21 +++--- .../org/apache/spark/sql/DataFrameSuite.scala | 67 +- .../apache/spark/sql/NestedDataSourceSuite.scala | 5 +- .../org/apache/spark/sql/SQLInsertTestSuite.scala | 7 +- .../spark/sql/StatisticsCollectionSuite.scala | 10 +-- .../spark/sql/connector/AlterTableTests.scala | 34 + .../spark/sql/connector/DataSourceV2SQLSuite.scala | 80 -- .../connector/V2CommandsCaseSensitivitySuite.scala | 21 -- .../spark/sql/execution/command/DDLSuite.scala | 80 +- .../datasources/jdbc/JdbcUtilsSuite.scala | 6 +- .../sql/execution/datasources/json/JsonSuite.scala | 10 +-- .../org/apache/spark/sql/jdbc/JDBCWriteSuite.scala | 10 +-- .../spark/sql/sources/PartitionedWriteSuite.scala | 9 ++- .../spark/sql/streaming/FileStreamSinkSuite.scala | 10 +-- .../sql/test/DataFrameReaderWriterSuite.scala | 65 ++ .../hive/execution/InsertIntoHiveDirCommand.scala | 1 - .../org/apache/spark/sql/hive/InsertSuite.scala| 9 ++- .../spark/sql/hive/execution/HiveDDLSuite.scala| 16 +++-- 38 files changed, 326 insertions(+), 314 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org