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.

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

Closed
wants to merge 7 commits into from
Closed
Show file tree
Hide file tree
Changes from 3 commits
Commits
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
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.

add a comment here indicating we are reseting stats based on pruned file size?

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, Thanks.

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
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,15 @@
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}
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

Expand Down Expand Up @@ -66,4 +68,42 @@ 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")
Copy link
Contributor

Choose a reason for hiding this comment

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

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, 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)
}

val tableName = "partTbl"
sql(s"analyze table partTbl compute STATISTICS")
Copy link
Contributor

Choose a reason for hiding this comment

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

nit: ANALYZE TABLE partTbl COMPUTE STATISTICS

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, Thanks.


val tableStats =
spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName)).stats
assert(tableStats.isDefined && tableStats.get.sizeInBytes > 0, "tableStats is lost")

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
Copy link
Contributor

@wzhfy wzhfy Jun 11, 2017

Choose a reason for hiding this comment

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

nit: just df.queryExecution.analyzed?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Copy link
Contributor

Choose a reason for hiding this comment

The 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
Copy link
Contributor

@wzhfy wzhfy Jun 10, 2017

Choose a reason for hiding this comment

The 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 spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName)).stats to query the catalog stats.

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, Thanks.

Copy link
Contributor

@wzhfy wzhfy Jun 11, 2017

Choose a reason for hiding this comment

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

Can we get catalog stats by relation.catalogTable.get.stats.get here and check it? I just think we need to cover this reset code path

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, Thanks.

}
assert(sizes1.size === 1, s"Size wrong for:\n ${df.queryExecution}")
assert(sizes1(0) == tableStats.get.sizeInBytes)
val sizes2 = Optimize.execute(query).collect {
case relation: LogicalRelation => relation.computeStats(conf).sizeInBytes
}
assert(sizes2.size === 1, s"Size wrong for:\n ${df.queryExecution}")
Copy link
Contributor

Choose a reason for hiding this comment

The 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 computeStats(conf).sizeInBytes?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Copy link
Contributor

Choose a reason for hiding this comment

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

LogicalRelation overrides computeStats and it will use CatalogStatistics if it exists

Copy link
Contributor

Choose a reason for hiding this comment

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

I donot think that it have changed the stats of catalog.

Don't we reset the catalog stats using the pruned size here?

assert(sizes2(0) < tableStats.get.sizeInBytes)
}
}
}
}
}