You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/12/23 21:40:23 UTC
[GitHub] [spark] viirya opened a new pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
viirya opened a new pull request #30912:
URL: https://github.com/apache/spark/pull/30912
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This patch proposes to do column pruning for CsvToStructs expression if we only require some fields from it.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
`CsvToStructs` takes a schema parameter used to tell CSV Parser what fields are needed to parse. If `CsvToStructs` is followed by GetStructField. We can prune the schema to only parse certain field.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
Unit test
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751656052
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38024/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751124947
**[Test build #133357 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133357/testReport)** for PR 30912 at commit [`0461b8d`](https://github.com/apache/spark/commit/0461b8da388931891c504308c2e404abcff7195f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750713269
**[Test build #133331 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133331/testReport)** for PR 30912 at commit [`4353f4e`](https://github.com/apache/spark/commit/4353f4efa054980de0a2905dbcf7000fd32524e4).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751602459
**[Test build #133417 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133417/testReport)** for PR 30912 at commit [`6c51f88`](https://github.com/apache/spark/commit/6c51f88f43595d78d06ea1556cf1640f70adc8d4).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751761845
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133431/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548755328
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
Review comment:
Technically, I'm neutral in terms of separating.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751761845
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133431/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750741601
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133338/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750745148
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37929/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750951822
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750485155
**[Test build #133322 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133322/testReport)** for PR 30912 at commit [`73263ba`](https://github.com/apache/spark/commit/73263ba542d3972bb46f9507f1224e41438a7957).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750698157
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133322/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750785864
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37934/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750760830
**[Test build #133324 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133324/testReport)** for PR 30912 at commit [`b500344`](https://github.com/apache/spark/commit/b500344c7a5b0bf72b9b868d3f7a6ff211a328bf).
* This patch **fails from timeout after a configured wait of `500m`**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751126348
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133357/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751629279
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38021/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751547776
**[Test build #133417 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133417/testReport)** for PR 30912 at commit [`6c51f88`](https://github.com/apache/spark/commit/6c51f88f43595d78d06ea1556cf1640f70adc8d4).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751558628
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38008/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750936403
**[Test build #133358 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133358/testReport)** for PR 30912 at commit [`4d6f60b`](https://github.com/apache/spark/commit/4d6f60b0f34fa0895cc2530da0c784fe0ff52ed6).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549688081
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ private def nameOfCorruptRecord = SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+ private val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ private val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(c @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
+ if schema.length > 1 && c.options.isEmpty && schema(ordinal).name != nameOfCorruptRecord =>
Review comment:
```suggestion
if schema.length > 1 && c.options.isEmpty && schema(ordinal).name != nameOfCorruptRecord =>
// When the parse mode is permissive, and corrupt column is not selected, we can prune here.
// from `GetStructField`. To be more conservative, it does not optimize when any option
// is set.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750807618
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133343/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751618945
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38021/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548401034
##########
File path: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvExprsSuite.scala
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+class OptimizeCsvExprsSuite extends PlanTest with ExpressionEvalHelper {
Review comment:
Added e2e tests now. Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750741601
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133338/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751703446
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133433/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750722241
**[Test build #133331 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133331/testReport)** for PR 30912 at commit [`4353f4e`](https://github.com/apache/spark/commit/4353f4efa054980de0a2905dbcf7000fd32524e4).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750741485
**[Test build #133338 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133338/testReport)** for PR 30912 at commit [`58cc2ba`](https://github.com/apache/spark/commit/58cc2ba4fc5a032cefb285a8c35f4bfb8dff9a07).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon closed pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon closed pull request #30912:
URL: https://github.com/apache/spark/pull/30912
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750948906
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751558615
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38008/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751083121
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133358/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750698157
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133322/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549688081
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ private def nameOfCorruptRecord = SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+ private val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ private val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(c @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
+ if schema.length > 1 && c.options.isEmpty && schema(ordinal).name != nameOfCorruptRecord =>
Review comment:
```suggestion
if schema.length > 1 && c.options.isEmpty && schema(ordinal).name != nameOfCorruptRecord =>
// When the parse mode is permissive, and corrupt column is not selected, we can prune here
// from `GetStructField`. To be more conservative, it does not optimize when any option
// is set.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548355934
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
Review comment:
nit but maybe private.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750750935
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548371264
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
Review comment:
Hm, rethinking about it, I think it is safer to do pruning only if `options` is empty. This optimization might hide a parsing error when parsing entire schema under "failfast" mode.
Although it is arguable if it still should throw the same error if we only need a field that could be parsed correctly. But I don't want to change the behavior before the optimization.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750773611
**[Test build #133343 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133343/testReport)** for PR 30912 at commit [`0461b8d`](https://github.com/apache/spark/commit/0461b8da388931891c504308c2e404abcff7195f).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549235257
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+ private val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ private val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
+ if schema.length > 1 && j.options.isEmpty =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(requiredSchema = Some(prunedSchema)), ordinal = 0)
Review comment:
Just added a test. I also checked json. Json parser seems smarter and doesn't have the same issue.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750935770
**[Test build #133357 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133357/testReport)** for PR 30912 at commit [`0461b8d`](https://github.com/apache/spark/commit/0461b8da388931891c504308c2e404abcff7195f).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751603763
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133417/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750936403
**[Test build #133358 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133358/testReport)** for PR 30912 at commit [`4d6f60b`](https://github.com/apache/spark/commit/4d6f60b0f34fa0895cc2530da0c784fe0ff52ed6).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750794245
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37934/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751748009
**[Test build #133431 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133431/testReport)** for PR 30912 at commit [`d98f780`](https://github.com/apache/spark/commit/d98f7801d9a09f6402a4d3011ac29e202c46c81a).
* This patch **fails from timeout after a configured wait of `500m`**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548744919
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
Review comment:
I re-checked Json optimization. Json parser behavior seems different than Csv here. Pruned schema doesn't change failfast behavor of json parser.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750727016
**[Test build #133338 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133338/testReport)** for PR 30912 at commit [`58cc2ba`](https://github.com/apache/spark/commit/58cc2ba4fc5a032cefb285a8c35f4bfb8dff9a07).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750944301
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37948/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750951822
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon edited a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon edited a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751611488
I have no more comments for now. I don't mind if somebody merges FWIW. I can take a look again tonight or early tomorrow but don't block it by me :-).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750486073
**[Test build #133324 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133324/testReport)** for PR 30912 at commit [`b500344`](https://github.com/apache/spark/commit/b500344c7a5b0bf72b9b868d3f7a6ff211a328bf).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549186677
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+ private val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ private val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
+ if schema.length > 1 && j.options.isEmpty =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(requiredSchema = Some(prunedSchema)), ordinal = 0)
Review comment:
@viirya, shall we exclude this optiomization when malformed column is selected?
```scala
spark.sparkContext.parallelize(Seq("a,b,c,d")).toDF("csv")
.selectExpr("from_csv(csv, 'a string, b string, _corrupt_record string') as parsed")
.selectExpr("parsed._corrupt_record").show()
```
Before:
```
+---------------+
|_corrupt_record|
+---------------+
| a,b,c,d|
+---------------+
```
After:
```
java.lang.ArrayIndexOutOfBoundsException: 0
at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.genericGet(rows.scala:201)
at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getAs(rows.scala:35)
at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.get(rows.scala:37)
at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.get$(rows.scala:37)
at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.get(rows.scala:195)
at org.apache.spark.sql.catalyst.util.FailureSafeParser.$anonfun$toResultRow$2(FailureSafeParser.scala:47)
at scala.Option.map(Option.scala:230)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751341329
I'm okay but allow me to review one more time before merging it in. CSV/JSON ones have many corner cases which are easy to overlook ..
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549240959
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ val nameOfCorruptRecord = SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)
Review comment:
no big deal but I would do this as a `private def` in case it doesn't respect runtime config.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750750934
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750794254
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/37934/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750716613
Should we maybe separate the refactoring out of this PR? Now we have one generalized rule but have two configurations to turn on and off each.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549240959
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ val nameOfCorruptRecord = SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)
Review comment:
no big deal but I would do this as a `private lazy val` in case it doesn't respect runtime config.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751637946
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38021/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750737032
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37929/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751083121
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133358/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750771518
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751547776
**[Test build #133417 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133417/testReport)** for PR 30912 at commit [`6c51f88`](https://github.com/apache/spark/commit/6c51f88f43595d78d06ea1556cf1640f70adc8d4).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750495126
cc @HyukjinKwon @dongjoon-hyun @maropu
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750935770
**[Test build #133357 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133357/testReport)** for PR 30912 at commit [`0461b8d`](https://github.com/apache/spark/commit/0461b8da388931891c504308c2e404abcff7195f).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750761612
For Json's e2e tests, I will work in other PR. Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548369089
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
Review comment:
Good catch! `from_csv` needs to pass in both data schema and required schema for pruning usage.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-752177321
Thanks @dongjoon-hyun @HyukjinKwon
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548315495
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
Review comment:
Thank you for generalizing this. This renaming looks fine to me.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548360407
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
Review comment:
Ah I see. Since we generalized the rule, it might more make sense to have one configuration to enable and disable this ..
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751703446
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133433/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548367638
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
Review comment:
Hm .. okay. I still prefer to separate refactoring from this PR but it's fine if both you or @dongjoon-hyun prefer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751126348
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133357/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750728757
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/37922/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548357636
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
Review comment:
Not dependent, but if so `p` only goes to match first case when both are enabled.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750935055
retest this please
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751605456
**[Test build #133431 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133431/testReport)** for PR 30912 at commit [`d98f780`](https://github.com/apache/spark/commit/d98f7801d9a09f6402a4d3011ac29e202c46c81a).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750486073
**[Test build #133324 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133324/testReport)** for PR 30912 at commit [`b500344`](https://github.com/apache/spark/commit/b500344c7a5b0bf72b9b868d3f7a6ff211a328bf).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548316158
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
Review comment:
indentation?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548366483
##########
File path: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvExprsSuite.scala
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+class OptimizeCsvExprsSuite extends PlanTest with ExpressionEvalHelper {
Review comment:
I think we should have an e2e test too since there's runtime options available in CSV that affects output and how to parse (e.g., https://github.com/apache/spark/pull/30912/files/4353f4efa054980de0a2905dbcf7000fd32524e4#r548365520).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751554339
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38008/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750727016
**[Test build #133338 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133338/testReport)** for PR 30912 at commit [`58cc2ba`](https://github.com/apache/spark/commit/58cc2ba4fc5a032cefb285a8c35f4bfb8dff9a07).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548365520
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
Review comment:
@viirya, this has a behaviour about parse mode. For example, previously it worked in failfast mode when other columns can be passed but. Looks like now it fails, for example:
```scala
spark.sparkContext.parallelize(Seq("a,b")).toDF("csv")
.selectExpr("from_csv(csv, 'a string, b string', map('mode', 'failfast')) as parsed")
.selectExpr("parsed.a").show()
```
Before this PR:
```
+---+
| a|
+---+
| a|
+---+
```
After this PR:
```
org.apache.spark.SparkException: Malformed records are detected in record parsing. Parse Mode: FAILFAST. To process malformed records as null result, try setting the option 'mode' as 'PERMISSIVE'.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750572931
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/37916/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549186677
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+ private val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ private val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
+ if schema.length > 1 && j.options.isEmpty =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(requiredSchema = Some(prunedSchema)), ordinal = 0)
Review comment:
@viirya, shall we exclude this optiomization when malformed column is selected? Looks like the corrupt column doesn't work here.
```scala
spark.sparkContext.parallelize(Seq("a,b,c,d")).toDF("csv")
.selectExpr("from_csv(csv, 'a string, b string, _corrupt_record string') as parsed")
.selectExpr("parsed._corrupt_record").show()
```
Before:
```
+---------------+
|_corrupt_record|
+---------------+
| a,b,c,d|
+---------------+
```
After:
```
java.lang.ArrayIndexOutOfBoundsException: 0
at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.genericGet(rows.scala:201)
at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getAs(rows.scala:35)
at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.get(rows.scala:37)
at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.get$(rows.scala:37)
at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.get(rows.scala:195)
at org.apache.spark.sql.catalyst.util.FailureSafeParser.$anonfun$toResultRow$2(FailureSafeParser.scala:47)
at scala.Option.map(Option.scala:230)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549190001
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+ private val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ private val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
+ if schema.length > 1 && j.options.isEmpty =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(requiredSchema = Some(prunedSchema)), ordinal = 0)
Review comment:
Hm, interesting. I will also add test for this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751637946
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38021/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751558628
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38008/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548367638
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
Review comment:
Hm .. okay. I still prefer to separate refactoring from this PR but it's fine if both you and @dongjoon-hyun prefer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548316620
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+
+ }
+
+ val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _), ordinal, _)
Review comment:
Just a question. Do you have more optimizations in this area in the near future?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751656052
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38024/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750722707
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37922/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548355759
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
Review comment:
What about:
```scala
case p if SQLConf.get.jsonExpressionOptimization => p.transformExpressions(jsonOptimization)
case p if SQLConf.get.csvExpressionOptimization => p.transformExpressions(csvOptimization)
case p => p
```
? Or is the optimization dependent on each other?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750771519
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750691251
**[Test build #133322 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133322/testReport)** for PR 30912 at commit [`73263ba`](https://github.com/apache/spark/commit/73263ba542d3972bb46f9507f1224e41438a7957).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750725172
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133331/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548351376
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+
+ }
+
+ val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _), ordinal, _)
Review comment:
Yea, I'll add a few more Csv optimization similar to Json ones.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750728750
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37922/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751603763
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133417/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751622345
**[Test build #133433 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133433/testReport)** for PR 30912 at commit [`55fb78f`](https://github.com/apache/spark/commit/55fb78fbcc59be2d8569093f70ddb2f329127b03).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548366884
##########
File path: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvExprsSuite.scala
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+class OptimizeCsvExprsSuite extends PlanTest with ExpressionEvalHelper {
Review comment:
Okay. I will add e2e test later.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750943726
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37949/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548358639
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
Review comment:
@viirya, I know this is not the part of this PR but nit: `v.nameExprs.head.semanticEquals(c.valExprs.head.children.head)`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750807618
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133343/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751611488
I have no more comments for now. I don't mind if somebody merges FWIW.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548361945
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
Review comment:
To enable/disable entire rule, `spark.sql.optimizer.excludedRules` should be able to help. So the purpose here is to have individual configs for enabling csv/json.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750572931
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/37916/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548317077
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+
Review comment:
Although this comes from the old file, shall we remove this empty line at this PR?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751696069
**[Test build #133433 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133433/testReport)** for PR 30912 at commit [`55fb78f`](https://github.com/apache/spark/commit/55fb78fbcc59be2d8569093f70ddb2f329127b03).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750738156
**[Test build #133339 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133339/testReport)** for PR 30912 at commit [`fc5dd38`](https://github.com/apache/spark/commit/fc5dd38c1c0f8c90fe083f0e7f603ccdcc6e7b19).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r548360767
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+
+ val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ val csvOptimization: PartialFunction[Expression, Expression] = {
Review comment:
here too. nit but maybe private
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750794254
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/37934/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751056846
**[Test build #133358 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133358/testReport)** for PR 30912 at commit [`4d6f60b`](https://github.com/apache/spark/commit/4d6f60b0f34fa0895cc2530da0c784fe0ff52ed6).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750485155
**[Test build #133322 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133322/testReport)** for PR 30912 at commit [`73263ba`](https://github.com/apache/spark/commit/73263ba542d3972bb46f9507f1224e41438a7957).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-752060672
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a change in pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #30912:
URL: https://github.com/apache/spark/pull/30912#discussion_r549183877
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, StructType}
+
+/**
+ * Simplify redundant csv/json related expressions.
+ *
+ * The optimization includes:
+ * 1. JsonToStructs(StructsToJson(child)) => child.
+ * 2. Prune unnecessary columns from GetStructField/GetArrayStructFields + JsonToStructs.
+ * 3. CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) =>
+ * If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))
+ * if JsonToStructs(json) is shared among all fields of CreateNamedStruct. `prunedSchema`
+ * contains all accessed fields in original CreateNamedStruct.
+ * 4. Prune unnecessary columns from GetStructField + CsvToStructs.
+ */
+object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ case p =>
+ val optimized = if (SQLConf.get.jsonExpressionOptimization) {
+ p.transformExpressions(jsonOptimization)
+ } else {
+ p
+ }
+
+ if (SQLConf.get.csvExpressionOptimization) {
+ optimized.transformExpressions(csvOptimization)
+ } else {
+ optimized
+ }
+ }
+
+ private val jsonOptimization: PartialFunction[Expression, Expression] = {
+ case c: CreateNamedStruct
+ // If we create struct from various fields of the same `JsonToStructs`.
+ if c.valExprs.forall { v =>
+ v.isInstanceOf[GetStructField] &&
+ v.asInstanceOf[GetStructField].child.isInstanceOf[JsonToStructs] &&
+ v.children.head.semanticEquals(c.valExprs.head.children.head)
+ } =>
+ val jsonToStructs = c.valExprs.map(_.children.head)
+ val sameFieldName = c.names.zip(c.valExprs).forall {
+ case (name, valExpr: GetStructField) =>
+ name.toString == valExpr.childSchema(valExpr.ordinal).name
+ case _ => false
+ }
+
+ // Although `CreateNamedStruct` allows duplicated field names, e.g. "a int, a int",
+ // `JsonToStructs` does not support parsing json with duplicated field names.
+ val duplicateFields = c.names.map(_.toString).distinct.length != c.names.length
+
+ // If we create struct from various fields of the same `JsonToStructs` and we don't
+ // alias field names and there is no duplicated field in the struct.
+ if (sameFieldName && !duplicateFields) {
+ val fromJson = jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
+ val nullFields = c.children.grouped(2).flatMap {
+ case Seq(name, value) => Seq(name, Literal(null, value.dataType))
+ }.toSeq
+
+ If(IsNull(fromJson.child), c.copy(children = nullFields), KnownNotNull(fromJson))
+ } else {
+ c
+ }
+
+ case jsonToStructs @ JsonToStructs(_, options1,
+ StructsToJson(options2, child, timeZoneId2), timeZoneId1)
+ if options1.isEmpty && options2.isEmpty && timeZoneId1 == timeZoneId2 &&
+ jsonToStructs.dataType == child.dataType =>
+ // `StructsToJson` only fails when `JacksonGenerator` encounters data types it
+ // cannot convert to JSON. But `StructsToJson.checkInputDataTypes` already
+ // verifies its child's data types is convertible to JSON. But in
+ // `StructsToJson(JsonToStructs(...))` case, we cannot verify input json string
+ // so `JsonToStructs` might throw error in runtime. Thus we cannot optimize
+ // this case similarly.
+ child
+
+ case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _), ordinal, _)
+ if schema.length > 1 =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
+
+ case g @ GetArrayStructFields(j @ JsonToStructs(schema: ArrayType, _, _, _), _, _, _, _)
+ if schema.elementType.asInstanceOf[StructType].length > 1 =>
+ val prunedSchema = ArrayType(StructType(Seq(g.field)), g.containsNull)
+ g.copy(child = j.copy(schema = prunedSchema), ordinal = 0, numFields = 1)
+ }
+
+ private val csvOptimization: PartialFunction[Expression, Expression] = {
+ case g @ GetStructField(j @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
+ if schema.length > 1 && j.options.isEmpty =>
+ val prunedSchema = StructType(Seq(schema(ordinal)))
+ g.copy(child = j.copy(requiredSchema = Some(prunedSchema)), ordinal = 0)
Review comment:
```suggestion
case g @ GetStructField(c @ CsvToStructs(schema: StructType, _, _, _, None), ordinal, _)
if schema.length > 1 && c.options.isEmpty =>
val prunedSchema = StructType(Seq(schema(ordinal)))
g.copy(child = c.copy(requiredSchema = Some(prunedSchema)), ordinal = 0)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751622345
**[Test build #133433 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133433/testReport)** for PR 30912 at commit [`55fb78f`](https://github.com/apache/spark/commit/55fb78fbcc59be2d8569093f70ddb2f329127b03).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750725172
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-751605456
**[Test build #133431 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133431/testReport)** for PR 30912 at commit [`d98f780`](https://github.com/apache/spark/commit/d98f7801d9a09f6402a4d3011ac29e202c46c81a).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #30912: [SPARK-32968][SQL] Prune unnecessary columns from CsvToStructs
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #30912:
URL: https://github.com/apache/spark/pull/30912#issuecomment-750713269
**[Test build #133331 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/133331/testReport)** for PR 30912 at commit [`4353f4e`](https://github.com/apache/spark/commit/4353f4efa054980de0a2905dbcf7000fd32524e4).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org