Skip to content

[WIP][SPARK-2961][SQL] Use statistics to skip cached partitions #1883

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

Closed
marmbrus wants to merge 3 commits into from
Closed

[WIP][SPARK-2961][SQL] Use statistics to skip cached partitions #1883

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0c15b4c
First draft of partition skipping based on statistics
marmbrus Aug 11, 2014
58a4e78
better logging
marmbrus Aug 11, 2014
f6419da
WIP
marmbrus Aug 12, 2014
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
Prev Previous commit
WIP
  • Loading branch information
@marmbrus
marmbrus committed Aug 12, 2014
commit f6419da35430e677a01d916992435444fc7baa57
6 changes: 3 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeMap.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,15 +34,15 @@ object AttributeMap {
new AttributeMap(kvs.map(kv => (kv._1.exprId, (kv._1, kv._2))).toMap)
}

class AttributeMap[A](baseMap: Map[ExprId, (Attribute, A)])
class AttributeMap[A] protected (baseMap: Map[ExprId, (Attribute, A)])
extends Map[Attribute, A] with Serializable {

override def get(k: Attribute): Option[A] = baseMap.get(k.exprId).map(_._2)

override def +[B1 >: A](kv: (Attribute, B1)): Map[Attribute, B1] =
override def +[B1 >: A](kv: (Attribute, B1)): Map[Attribute, B1] = // scalastyle:ignore
(baseMap.map(_._2) + kv).toMap

override def iterator: Iterator[(Attribute, A)] = baseMap.map(_._2).iterator

override def -(key: Attribute): Map[Attribute, A] = (baseMap.map(_._2) - key).toMap
}
}
8 changes: 6 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnStats.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,19 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.{AttributeMap, Attribute, AttributeReference}
import org.apache.spark.sql.catalyst.types._

class ColumnStatisticsSchema(a: Attribute) extends Serializable {
/** Provides the Attributes that can be used to query statistics for a given column. */
private[sql] class ColumnStatisticsSchema(a: Attribute) extends Serializable {
val upperBound = AttributeReference(a.name + ".upperBound", a.dataType, nullable = false)()
val lowerBound = AttributeReference(a.name + ".lowerBound", a.dataType, nullable = false)()
val nullCount = AttributeReference(a.name + ".nullCount", IntegerType, nullable = false)()

val schema = Seq(lowerBound, upperBound, nullCount)
}

class PartitionStatistics(tableSchema: Seq[Attribute]) extends Serializable {
/**
* Provides the Attributes that can be used to access
*/
private[sql] class PartitionStatistics(tableSchema: Seq[Attribute]) extends Serializable {
val (forAttribute, schema) = {
val allStats = tableSchema.map(a => a -> new ColumnStatisticsSchema(a))
(AttributeMap(allStats), allStats.map(_._2.schema).foldLeft(Seq.empty[Attribute])(_ ++ _))
Expand Down