-
Notifications
You must be signed in to change notification settings - Fork 28.1k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[SPARK-20986] [SQL] Reset table's statistics after PruneFileSourcePartitions rule. #18205
Changes from 2 commits
20a6043
c53a0c7
120662e
f7c3dfc
d1513a8
6e28460
16a3f7e
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -17,6 +17,7 @@ | |
|
||
package org.apache.spark.sql.execution.datasources | ||
|
||
import org.apache.spark.sql.catalyst.catalog.CatalogStatistics | ||
import org.apache.spark.sql.catalyst.expressions._ | ||
import org.apache.spark.sql.catalyst.planning.PhysicalOperation | ||
import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project} | ||
|
@@ -59,8 +60,10 @@ private[sql] object PruneFileSourcePartitions extends Rule[LogicalPlan] { | |
val prunedFileIndex = catalogFileIndex.filterPartitions(partitionKeyFilters.toSeq) | ||
val prunedFsRelation = | ||
fsRelation.copy(location = prunedFileIndex)(sparkSession) | ||
val prunedLogicalRelation = logicalRelation.copy(relation = prunedFsRelation) | ||
|
||
val withStats = logicalRelation.catalogTable.map(_.copy( | ||
stats = Some(CatalogStatistics(sizeInBytes = BigInt(prunedFileIndex.sizeInBytes))))) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. do we ignore all column stats here? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yes, Now it replace stats of CatalogTable with new CatalogStatistics() like DetermineTableStats. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Column stats are collected as table-level, here we need partition-specific stats, so we can ignore column stats. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ah actually we have to, the column stats is table level and is invalid for partitions. |
||
val prunedLogicalRelation = logicalRelation.copy( | ||
relation = prunedFsRelation, catalogTable = withStats) | ||
// Keep partition-pruning predicates so that they are visible in physical planning | ||
val filterExpression = filters.reduceLeft(And) | ||
val filter = Filter(filterExpression, prunedLogicalRelation) | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.rules.RuleExecutor | |
import org.apache.spark.sql.execution.datasources.{CatalogFileIndex, HadoopFsRelation, LogicalRelation, PruneFileSourcePartitions} | ||
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat | ||
import org.apache.spark.sql.hive.test.TestHiveSingleton | ||
import org.apache.spark.sql.internal.SQLConf | ||
import org.apache.spark.sql.test.SQLTestUtils | ||
import org.apache.spark.sql.types.StructType | ||
|
||
|
@@ -66,4 +67,35 @@ class PruneFileSourcePartitionsSuite extends QueryTest with SQLTestUtils with Te | |
} | ||
} | ||
} | ||
|
||
test("SPARK-20986 Reset table's statistics after PruneFileSourcePartitions rule") { | ||
withTempView("tempTbl") { | ||
withTable("partTbl") { | ||
spark.range(1000).selectExpr("id").createOrReplaceTempView("tempTbl") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. For this test, we can use a much smaller size (e.g. 10) to accelerate testing. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yes, Thanks. |
||
sql("CREATE TABLE partTbl (id INT) PARTITIONED BY (part INT) STORED AS parquet") | ||
for (part <- Seq(1, 2, 3)) { | ||
sql( | ||
s""" | ||
|INSERT OVERWRITE TABLE partTbl PARTITION (part='$part') | ||
|select id from tempTbl | ||
""".stripMargin) | ||
} | ||
|
||
withSQLConf(SQLConf.ENABLE_FALL_BACK_TO_HDFS_FOR_STATS.key -> "true") { | ||
val df = sql("SELECT * FROM partTbl where part = 1") | ||
val query = df.queryExecution.analyzed.analyze | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: just There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Because there is SubqueryAlias plan, I think that we need analyze() to eliminate it. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. but why we need to eliminate SubqueryAlias here? |
||
val sizes1 = query.collect { | ||
case relation: LogicalRelation => relation.computeStats(conf).sizeInBytes | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We'd better not to compute stats for an analyzed plan. We can use There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yes, Thanks. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can we get catalog stats by There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yes, Thanks. |
||
} | ||
assert(sizes1.size === 1, s"Size wrong for:\n ${df.queryExecution}") | ||
assert(sizes1(0) > 5000, s"expected > 5000 for test table 'src', got: ${sizes1(0)}") | ||
val sizes2 = Optimize.execute(query).collect { | ||
case relation: LogicalRelation => relation.computeStats(conf).sizeInBytes | ||
} | ||
assert(sizes2.size === 1, s"Size wrong for:\n ${df.queryExecution}") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. assert the new size in catalog stats is larger than the previous one, and equal to There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I donot think that it have changed the stats of catalog. after the optimizer, the size in catalog stats is larger than computeStats(conf).sizeInBytes because the partition pruned. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Don't we reset the catalog stats using the pruned size here? |
||
assert(sizes2(0) < 5000, s"expected < 5000 for test table 'src', got: ${sizes2(0)}") | ||
} | ||
} | ||
} | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
add a comment here indicating we are reseting stats based on pruned file size?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, Thanks.