[jira] [Commented] (SPARK-47230) The column pruning is not working as expected for nested struct in an array

["Ohad Raviv (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-47230?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131089#comment-18131089 ] 

Ohad Raviv commented on SPARK-47230:
------------------------------------

PR: https://github.com/apache/spark/pull/59706

It prunes nested fields read through chains of explode/posexplode in SchemaPruning (V1) and V2ScanRelationPushDown, which also covers SPARK-34956 (several fields of one exploded element). It stays inside the two existing scan pruning rules, adds no expression and no per-row work, and is behind an internal flag with a fallback to the unpruned plan.

On Parquet (JDK 17, NestedSchemaPruningBenchmark), two fields of one explode go from 1001 ms to 296 ms. With the same pruning as an optimizer extension on Databricks Runtime, two fields read through two chained explodes went from 557 MB to 15.6 MB read.

Joint work by Tom Sisso and me, building on Igor Berman's earlier PRs here (#52288, #52598, #54283): Igor contributed the query-shape test suite from Taboola's workloads, tested the change on them, and reviewed it.

> The column pruning is not working as expected for nested struct in an array
> ---------------------------------------------------------------------------
>
>                 Key: SPARK-47230
>                 URL: https://issues.apache.org/jira/browse/SPARK-47230
>             Project: Spark
>          Issue Type: Improvement
>          Components: Optimizer, Spark Core, Spark Shell
>    Affects Versions: 3.3.0, 3.4.0, 3.5.0, 3.5.1
>            Reporter: Ling Qin
>            Priority: Major
>              Labels: pull-request-available
>
> See the code snippet below, when explode an array of struct and select one field in the struct,  some unexpected behaviour observed:
>  * If the field in the struct is in the select clause, not in the where clause, the column pruning works as expected.
>  * If the field in the struct is in the select clause and in the where clause, the column pruning not working.
>  * If the field in the struct is not even selected, the column pruning not working
> {code:java}
> from pyspark.sql import SparkSession
> from pyspark.sql.types import IntegerType, StructField, StructType, ArrayType
> import random
> spark = SparkSession.builder.appName("example").getOrCreate()# Create an RDD with an array of structs, each array having a random size between 5 to 10
> rdd = spark.range(1000).rdd.map(lambda x: (x.id + 3, [(x.id + i, x.id - i) for i in range(1, random.randint(5, 11))]))
> # Define a new schema
> schema = StructType([
>     StructField("a", IntegerType(), True),
>     StructField("b", ArrayType(StructType([
>         StructField("x", IntegerType(), True),
>         StructField("y", IntegerType(), True)
>     ])), True)
> ])
> # Create a DataFrame with the new schema
> df = spark.createDataFrame(rdd, schema=schema)
> # Write the DataFrame to a parquet file
> df.repartition(1).write.mode('overwrite').parquet('test.parquet')
> # Read the parquet file back into a DataFrame
> df = spark.read.parquet('test.parquet') 
> df.createOrReplaceTempView("df_view")
> spark.conf.set('spark.sql.optimizer.nestedSchemaPruning.enabled', 'true')
> # case 1, as expected
> sql_query = """
> SELECT a, EXPLODE(b.x) AS bb
> FROM df_view
> """
> spark.sql(sql_query).explain()
> # ReadSchema: struct<a:int,b:array<struct<x:int>>>
> # case 2, as expected
> sql_query = """
> SELECT a, bb.x 
> FROM df_view 
> lateral view explode(b) as bb
> """
> spark.sql(sql_query).explain()
> # ReadSchema: struct<a:int,b:array<struct<x:int>>>
> # case 3, bug: should only read b.x
> sql_query = """
> SELECT a, bb.x 
> FROM df_view 
> lateral view explode(b) as bb
> where bb.x is not null
> """
> spark.sql(sql_query).explain()
> #ReadSchema: struct<a:int,b:array<struct<x:int,y:int>>>
> #case 4, bug? seems no need to read both a and b
> sql_query = """
> SELECT a
> FROM df_view 
> lateral view explode(b) as bb
> """
> spark.sql(sql_query).explain()
> #ReadSchema: struct<a:int,b:array<struct<x:int,y:int>>>{code}
>  
>  
>  



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org