Skip to content
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.

Sign up for GitHub

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

Jump to bottom

[SPARK-20986] [SQL] Reset table's statistics after PruneFileSourcePartitions rule. #18205

Closed
wants to merge 7 commits into from
Closed

[SPARK-20986] [SQL] Reset table's statistics after PruneFileSourcePartitions rule. #18205

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
20a6043
init commit
lianhuiwang Jun 5, 2017
c53a0c7
address comments.
lianhuiwang Jun 6, 2017
120662e
address comments.
lianhuiwang Jun 11, 2017
f7c3dfc
address comments.
lianhuiwang Jun 12, 2017
d1513a8
address comments for UT.
lianhuiwang Jun 13, 2017
6e28460
address comments for UT.
lianhuiwang Jun 13, 2017
16a3f7e
remove unused import.
lianhuiwang Jun 13, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions ...src/main/scala/org/apache/spark/sql/execution/datasources/PruneFileSourcePartitions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -59,8 +60,11 @@ 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)

// Change table stats based on the sizeInBytes of pruned files
val withStats = logicalRelation.catalogTable.map(_.copy(
stats = Some(CatalogStatistics(sizeInBytes = BigInt(prunedFileIndex.sizeInBytes)))))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we ignore all column stats here?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, Now it replace stats of CatalogTable with new CatalogStatistics() like DetermineTableStats.

Copy link
Contributor

Choose a reason for hiding this comment

The 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.

Copy link
Contributor

Choose a reason for hiding this comment

The 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)
Expand Down
25 changes: 25 additions & 0 deletions ...e/src/test/scala/org/apache/spark/sql/hive/execution/PruneFileSourcePartitionsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.sql.hive.execution

import org.apache.spark.sql.QueryTest
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
Expand Down Expand Up @@ -66,4 +67,28 @@ class PruneFileSourcePartitionsSuite extends QueryTest with SQLTestUtils with Te
}
}
}

test("SPARK-20986 Reset table's statistics after PruneFileSourcePartitions rule") {
withTable("tbl") {
spark.range(10).selectExpr("id", "id % 3 as p").write.partitionBy("p").saveAsTable("tbl")
sql(s"ANALYZE TABLE tbl COMPUTE STATISTICS")
val tableStats = spark.sessionState.catalog.getTableMetadata(TableIdentifier("tbl")).stats
assert(tableStats.isDefined && tableStats.get.sizeInBytes > 0, "tableStats is lost")

val df = sql("SELECT * FROM tbl WHERE p = 1")
val sizes1 = df.queryExecution.analyzed.collect {
case relation: LogicalRelation => relation.catalogTable.get.stats.get.sizeInBytes
}
assert(sizes1.size === 1, s"Size wrong for:\n ${df.queryExecution}")
assert(sizes1(0) == tableStats.get.sizeInBytes)

val relations = df.queryExecution.optimizedPlan.collect {
case relation: LogicalRelation => relation
}
assert(relations.size === 1, s"Size wrong for:\n ${df.queryExecution}")
val size2 = relations(0).computeStats(conf).sizeInBytes
assert(size2 == relations(0).catalogTable.get.stats.get.sizeInBytes)
assert(size2 < tableStats.get.sizeInBytes)
}
}
}