Skip to content

[SPARK-25306][SQL] Avoid skewed filter trees to speed up createFilter in ORC #22313

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
dongjoon-hyun wants to merge 6 commits into from
Closed

[SPARK-25306][SQL] Avoid skewed filter trees to speed up createFilter in ORC #22313

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ac06b0c
[SPARK-25306][SQL] Use cache to speed up `createFilter`
dongjoon-hyun Sep 1, 2018
7720134
Revert "[SPARK-25306][SQL] Use cache to speed up `createFilter`"
dongjoon-hyun Sep 3, 2018
4acbaf8
Address comments
dongjoon-hyun Sep 3, 2018
5c46693
Address comments
dongjoon-hyun Sep 4, 2018
4a372a3
fix typo
dongjoon-hyun Sep 4, 2018
3cd4443
Use old style.
dongjoon-hyun Sep 4, 2018
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
59 changes: 49 additions & 10 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,14 @@

package org.apache.spark.sql.execution.datasources.orc

import java.util.concurrent.TimeUnit

import com.google.common.cache.{CacheBuilder, CacheLoader}
import org.apache.orc.storage.ql.io.sarg.{PredicateLeaf, SearchArgument, SearchArgumentFactory}
import org.apache.orc.storage.ql.io.sarg.SearchArgument.Builder
import org.apache.orc.storage.serde2.io.HiveDecimalWritable

import org.apache.spark.SparkEnv
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types._

Expand Down Expand Up @@ -54,7 +58,37 @@ import org.apache.spark.sql.types._
* builder methods mentioned above can only be found in test code, where all tested filters are
* known to be convertible.
*/
private[orc] object OrcFilters {
private[sql] object OrcFilters {

case class FilterWithTypeMap(filter: Filter, typeMap: Map[String, DataType])

private val defaultCacheExpireTimeout = TimeUnit.SECONDS.toSeconds(20)

lazy val cacheExpireTimeout: Long =
Option(SparkEnv.get).map(_.conf.getTimeAsSeconds(
"spark.sql.orc.cache.sarg.timeout",
s"${defaultCacheExpireTimeout}s")).getOrElse(defaultCacheExpireTimeout)

private lazy val searchArgumentCache = CacheBuilder.newBuilder()
.expireAfterAccess(cacheExpireTimeout, TimeUnit.SECONDS)
.build(
new CacheLoader[FilterWithTypeMap, Option[Builder]]() {
override def load(typeMapAndFilter: FilterWithTypeMap): Option[Builder] = {
buildSearchArgument(
typeMapAndFilter.typeMap, typeMapAndFilter.filter, SearchArgumentFactory.newBuilder())
}
})

private def getOrBuildSearchArgumentWithNewBuilder(
dataTypeMap: Map[String, DataType],
expression: Filter): Option[Builder] = {
// When `spark.sql.orc.cache.sarg.timeout` is 0, cache is disabled.
if (cacheExpireTimeout > 0) {
searchArgumentCache.get(FilterWithTypeMap(expression, dataTypeMap))
} else {
buildSearchArgument(dataTypeMap, expression, SearchArgumentFactory.newBuilder())
}
}

/**
* Create ORC filter as a SearchArgument instance.
Expand All @@ -66,12 +100,19 @@ private[orc] object OrcFilters {
// collect all convertible ones to build the final `SearchArgument`.
val convertibleFilters = for {
filter <- filters
_ <- buildSearchArgument(dataTypeMap, filter, SearchArgumentFactory.newBuilder())
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, filter)
} yield filter

for {
// Combines all convertible filters using `And` to produce a single conjunction
conjunction <- convertibleFilters.reduceOption(org.apache.spark.sql.sources.And)
conjunction <- convertibleFilters.reduceOption { (x, y) =>
val newFilter = org.apache.spark.sql.sources.And(x, y)
if (cacheExpireTimeout > 0) {
// Build in a bottom-up manner
getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, newFilter)
}
newFilter
}
// Then tries to build a single ORC `SearchArgument` for the conjunction predicate
builder <- buildSearchArgument(dataTypeMap, conjunction, SearchArgumentFactory.newBuilder())
} yield builder.build()
Expand Down Expand Up @@ -127,8 +168,6 @@ private[orc] object OrcFilters {
dataTypeMap: Map[String, DataType],
expression: Filter,
builder: Builder): Option[Builder] = {
def newBuilder = SearchArgumentFactory.newBuilder()

def getType(attribute: String): PredicateLeaf.Type =
getPredicateLeafType(dataTypeMap(attribute))

Expand All @@ -144,23 +183,23 @@ private[orc] object OrcFilters {
// Pushing one side of AND down is only safe to do at the top level.
// You can see ParquetRelation's initializeLocalJobFunc method as an example.
for {
_ <- buildSearchArgument(dataTypeMap, left, newBuilder)
_ <- buildSearchArgument(dataTypeMap, right, newBuilder)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, left)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, right)
lhs <- buildSearchArgument(dataTypeMap, left, builder.startAnd())
rhs <- buildSearchArgument(dataTypeMap, right, lhs)
} yield rhs.end()

case Or(left, right) =>
for {
_ <- buildSearchArgument(dataTypeMap, left, newBuilder)
_ <- buildSearchArgument(dataTypeMap, right, newBuilder)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, left)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, right)
lhs <- buildSearchArgument(dataTypeMap, left, builder.startOr())
rhs <- buildSearchArgument(dataTypeMap, right, lhs)
} yield rhs.end()

case Not(child) =>
for {
_ <- buildSearchArgument(dataTypeMap, child, newBuilder)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, child)
negate <- buildSearchArgument(dataTypeMap, child, builder.startNot())
} yield negate.end()

Expand Down
16 changes: 13 additions & 3 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,10 @@ package org.apache.spark.sql.execution.datasources.orc
import java.nio.charset.StandardCharsets
import java.sql.{Date, Timestamp}

import scala.collection.JavaConverters._

import org.apache.orc.storage.ql.io.sarg.{PredicateLeaf, SearchArgument}
import org.scalatest.concurrent.TimeLimits
import org.scalatest.time.SpanSugar._
import scala.collection.JavaConverters._

import org.apache.spark.sql.{Column, DataFrame}
import org.apache.spark.sql.catalyst.dsl.expressions._
Expand All @@ -39,7 +40,7 @@ import org.apache.spark.sql.types._
* - OrcFilterSuite uses 'org.apache.orc.storage.ql.io.sarg' package.
* - HiveOrcFilterSuite uses 'org.apache.hadoop.hive.ql.io.sarg' package.
*/
class OrcFilterSuite extends OrcTest with SharedSQLContext {
class OrcFilterSuite extends OrcTest with SharedSQLContext with TimeLimits {

private def checkFilterPredicate(
df: DataFrame,
Expand Down Expand Up @@ -383,4 +384,13 @@ class OrcFilterSuite extends OrcTest with SharedSQLContext {
)).get.toString
}
}

test("createFilter should not hang") {
import org.apache.spark.sql.sources._
val schema = new StructType(Array(StructField("a", IntegerType, nullable = true)))
val filters = (1 to 500).map(LessThan("a", _)).toArray[Filter]
failAfter(2 seconds) {
OrcFilters.createFilter(schema, filters)

@cloud-fan cloud-fan Sep 4, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

This test looks tricky... It's a bad practice to assume some code will return in a certain time. Can we just add a microbenchmark for it?

@dongjoon-hyun dongjoon-hyun Sep 4, 2018 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Sure.

  1. Something like the test code in the PR description? And marked as ignore(...) instead of test(...) here?
  2. Or, do you want another test case in FilterPushdownBenchmark?

@dongjoon-hyun dongjoon-hyun Sep 4, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I'll choose (2), @cloud-fan .

}
}
}
53 changes: 44 additions & 9 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,13 @@

package org.apache.spark.sql.hive.orc

import java.util.concurrent.TimeUnit

import com.google.common.cache.{CacheBuilder, CacheLoader}
import org.apache.hadoop.hive.ql.io.sarg.{SearchArgument, SearchArgumentFactory}
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument.Builder

import org.apache.spark.SparkEnv
import org.apache.spark.internal.Logging
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._
Expand Down Expand Up @@ -55,19 +59,52 @@ import org.apache.spark.sql.types._
* known to be convertible.
*/
private[orc] object OrcFilters extends Logging {
case class FilterWithTypeMap(filter: Filter, typeMap: Map[String, DataType])

private lazy val cacheExpireTimeout =
org.apache.spark.sql.execution.datasources.orc.OrcFilters.cacheExpireTimeout

private lazy val searchArgumentCache = CacheBuilder.newBuilder()
.expireAfterAccess(cacheExpireTimeout, TimeUnit.SECONDS)
.build(
new CacheLoader[FilterWithTypeMap, Option[Builder]]() {
override def load(typeMapAndFilter: FilterWithTypeMap): Option[Builder] = {
buildSearchArgument(
typeMapAndFilter.typeMap, typeMapAndFilter.filter, SearchArgumentFactory.newBuilder())
}
})

private def getOrBuildSearchArgumentWithNewBuilder(

@xuanyuanking xuanyuanking Sep 2, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Just a little question about is any possible to reuse code with https://github.com/apache/spark/pull/22313/files#diff-224b8cbedf286ecbfdd092d1e2e2f237R61?

@dongjoon-hyun dongjoon-hyun Sep 2, 2018 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

@xuanyuanking . This already reuses cacheExpireTimeout.

For the cache value, SearchArgument, SearchArgumentFactory and Builder are different classes. (They only share the same names.)

  • Here, they comes from org.apache.hadoop.hive.ql.io.sarg.*.
  • There, they comes from org.apache.orc.storage.ql.io.sarg.*.

The only exception I made is FilterWithTypeMap. I wanted to keep them separately since it's also related to cache key.

dataTypeMap: Map[String, DataType],
expression: Filter): Option[Builder] = {
// When `spark.sql.orc.cache.sarg.timeout` is 0, cache is disabled.
if (cacheExpireTimeout > 0) {
searchArgumentCache.get(FilterWithTypeMap(expression, dataTypeMap))
} else {
buildSearchArgument(dataTypeMap, expression, SearchArgumentFactory.newBuilder())

@viirya viirya Sep 2, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

When we set timeout to zero on the cache, the loaded element can be removed immediately. Maybe we don't need to check timeout like this and we can simplify the code.

@dongjoon-hyun dongjoon-hyun Sep 3, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Ya. It's possible. But, if we create a Guava loading cache and pass through all the cache management logic in Guava, it means a more overhead than this PR. In this PR, spark.sql.orc.cache.sarg.timeout=0 means not creating the loading cache at all.

}
}

def createFilter(schema: StructType, filters: Array[Filter]): Option[SearchArgument] = {
val dataTypeMap = schema.map(f => f.name -> f.dataType).toMap

// First, tries to convert each filter individually to see whether it's convertible, and then
// collect all convertible ones to build the final `SearchArgument`.
val convertibleFilters = for {
filter <- filters
_ <- buildSearchArgument(dataTypeMap, filter, SearchArgumentFactory.newBuilder())
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, filter)
} yield filter

for {
// Combines all convertible filters using `And` to produce a single conjunction
conjunction <- convertibleFilters.reduceOption(And)
conjunction <- convertibleFilters.reduceOption { (x, y) =>
val newFilter = org.apache.spark.sql.sources.And(x, y)
if (cacheExpireTimeout > 0) {
// Build in a bottom-up manner
getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, newFilter)
}

@viirya viirya Sep 2, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Why we need to cache all sub filters? Don't we just need to cache the final conjunction?

@dongjoon-hyun dongjoon-hyun Sep 3, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Final conjunction? All sub function results will be cached in the end.

newFilter
}
// Then tries to build a single ORC `SearchArgument` for the conjunction predicate
builder <- buildSearchArgument(dataTypeMap, conjunction, SearchArgumentFactory.newBuilder())
} yield builder.build()
Expand All @@ -77,8 +114,6 @@ private[orc] object OrcFilters extends Logging {
dataTypeMap: Map[String, DataType],
expression: Filter,
builder: Builder): Option[Builder] = {
def newBuilder = SearchArgumentFactory.newBuilder()

def isSearchableType(dataType: DataType): Boolean = dataType match {
// Only the values in the Spark types below can be recognized by
// the `SearchArgumentImpl.BuilderImpl.boxLiteral()` method.
Expand All @@ -98,23 +133,23 @@ private[orc] object OrcFilters extends Logging {
// Pushing one side of AND down is only safe to do at the top level.
// You can see ParquetRelation's initializeLocalJobFunc method as an example.
for {
_ <- buildSearchArgument(dataTypeMap, left, newBuilder)
_ <- buildSearchArgument(dataTypeMap, right, newBuilder)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, left)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, right)
lhs <- buildSearchArgument(dataTypeMap, left, builder.startAnd())
rhs <- buildSearchArgument(dataTypeMap, right, lhs)
} yield rhs.end()

case Or(left, right) =>
for {
_ <- buildSearchArgument(dataTypeMap, left, newBuilder)
_ <- buildSearchArgument(dataTypeMap, right, newBuilder)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, left)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, right)
lhs <- buildSearchArgument(dataTypeMap, left, builder.startOr())
rhs <- buildSearchArgument(dataTypeMap, right, lhs)
} yield rhs.end()

case Not(child) =>
for {
_ <- buildSearchArgument(dataTypeMap, child, newBuilder)
_ <- getOrBuildSearchArgumentWithNewBuilder(dataTypeMap, child)
negate <- buildSearchArgument(dataTypeMap, child, builder.startNot())
} yield negate.end()

Expand Down
16 changes: 13 additions & 3 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcFilterSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,10 @@ package org.apache.spark.sql.hive.orc
import java.nio.charset.StandardCharsets
import java.sql.{Date, Timestamp}

import scala.collection.JavaConverters._

import org.apache.hadoop.hive.ql.io.sarg.{PredicateLeaf, SearchArgument}
import org.scalatest.concurrent.TimeLimits
import org.scalatest.time.SpanSugar._
import scala.collection.JavaConverters._

import org.apache.spark.sql.{Column, DataFrame}
import org.apache.spark.sql.catalyst.dsl.expressions._
Expand All @@ -36,7 +37,7 @@ import org.apache.spark.sql.types._
/**
* A test suite that tests Hive ORC filter API based filter pushdown optimization.
*/
class HiveOrcFilterSuite extends OrcTest with TestHiveSingleton {
class HiveOrcFilterSuite extends OrcTest with TestHiveSingleton with TimeLimits {

override val orcImp: String = "hive"

Expand Down Expand Up @@ -384,4 +385,13 @@ class HiveOrcFilterSuite extends OrcTest with TestHiveSingleton {
)).get.toString
}
}

test("createFilter should not hang") {
import org.apache.spark.sql.sources._
val schema = new StructType(Array(StructField("a", IntegerType, nullable = true)))
val filters = (1 to 500).map(LessThan("a", _)).toArray[Filter]
failAfter(2 seconds) {
OrcFilters.createFilter(schema, filters)
}
}
}