spark-test commented on code in PR #59929:
URL: https://github.com/apache/spark/pull/59929#discussion_r4469485193
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala:
##########
@@ -83,6 +85,7 @@ class ColumnarScanBuilder(
schema: StructType,
createDataSource: () => ColumnarDataSource)
extends ScanBuilder
+ with SupportsPushDownV2Filters
Review Comment:
**Non-blocking (P2):** Because this mixes in `SupportsPushDownV2Filters`
unconditionally, every filtered native batch scan now takes the V2 branch of
`PushDownUtils.pushFilters`. That branch rebuilds the conjuncts that Spark
evaluates after the scan as the translatable deterministic ones, then the
untranslatable ones, then the nondeterministic ones (`prioritizeFilters`). On
the base branch this builder implemented no filter interface, so `pushFilters`
kept the conjuncts in their order. `FilterExec` evaluates them in that order
and drops a row at the first false one.
Take `rand() < 0.01 AND lower(name) LIKE '%n1%'`. `LikeSimplification` turns
the `LIKE` into `Contains(Lower(name), 'n1')`, which V2 translates. So `LOWER`
and `CONTAINS` now run on every scanned row instead of on the 1% of rows that
`rand()` keeps, and the seeded `rand()` now runs only on their survivors, so it
samples other rows. `array_contains(tags, 'x') AND upper(name) = 'X'` behaves
the same, since `array_contains` is not translated. Such expensive conjuncts
derive no domain, so skipping offsets the cost only when another conjunct gives
the library a domain that it uses.
Returning every predicate from `pushPredicates` does not keep the order.
With `spark.sql.dataSource.native.filterPushdown.enabled=false` the builder
still implements the interface, so turning the pushdown off does not restore
the base branch's plan either. The reordering is Spark's heuristic for other V2
sources and helps the opposite shape, but here it also applies to filters that
gain nothing from the pushdown.
**Recommended change:** Keep the written conjunct order of the post-scan
filter for native scans. Add an internal capability, declared by the columnar
scan builder, meaning that every pushed predicate is also returned for
evaluation after the scan; when a builder has it, the filter pushdown helper
uses the conjuncts in their written order, nondeterministic ones included, as
the post-scan filters instead of reordering them. Other Data Source V2 builders
keep the current order.
**Why this works:** The native builder still receives the translated
deterministic predicates and derives domains from them, but the post-scan
filter is the original conjunct list, so the base branch's evaluation order,
and with it the work FilterExec does for each conjunct, stays the same whether
or not domains are pushed and whatever the flag is set to.
**Scope:** Post-scan filter construction for native batch scans
**Compatibility:** Every pushed predicate is still evaluated after the scan,
and the domains passed to libraries are unchanged.
**Risks:** Native scans that push domains no longer get the
translatable-first order either, so a filter that writes an expensive
untranslatable conjunct first keeps running it first, as on the base branch;
this returns to the base branch's behavior rather than adding a regression, and
it is limited to native scans.
**Constraints:** The capability stays internal, and only the columnar scan
builder declares it, so the post-scan order of other V2 sources does not
change. Nondeterministic conjuncts are still never pushed to the builder.
**Exclusions:**
- The PushedFilters entry with an empty list stays in the scan metadata when
nothing is pushed, since formatted explain already hides it.
**Success:** For a native batch scan, the post-scan Filter keeps its
conjuncts in the order they had above the scan, nondeterministic ones included,
except that the conjuncts with subqueries still come last, as on the base
branch, whether or not domains are pushed and whatever
spark.sql.dataSource.native.filterPushdown.enabled is set to. With the flag
off, the optimized plan of a native scan matches the base branch. Domains are
still derived, pushed and reported as before, and results do not change. Other
Data Source V2 sources keep their current post-scan order.
**Verification:**
- **Regression:** A native scan filtered by a selective nondeterministic
conjunct written before an expensive translatable one, such as rand() < 0.01
AND lower(name) LIKE '%n1%', keeps the written conjunct order in its optimized
post-scan Filter, with the flag on and off and for libraries with and without
push_domains.
- **Behavior:** Filters that constrain columns still push the same domains,
report the same pushed filters and return the same rows.
- **Compatibility:** The post-scan filter order of a Data Source V2 source
with filter pushdown other than the native one is unchanged.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -881,6 +994,16 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
check("plan a scan of", "injected failure in plan") { session =>
range(session, "fail" -> "plan").collect()
}
+ // Spark releases the ranges of the domains that it pushes down, after a
failure of push_domains
Review Comment:
**Nit (P3):** These checks cover a failing `push_domains` and a successful
one, but in both cases Spark releases the ranges itself, since the test library
only reads `domain->ranges` in place. The header also lets a library move the
ranges and their schema to take ownership, and both examples in the description
do so (arrow-rs `from_ffi`, Arrow C++ `ImportArray`).
`NativeArrow.exportDomains` relies on that when it calls `release` and `close`
on both structs and then closes the root. If that cleanup ever freed the
exported buffers whatever the library did, a library that keeps the moved
ranges in its reader for `partitions` would read freed memory, and every test
here would still pass.
A test library option could cover that path: move the ranges in
`push_domains`, read them when planning the partitions, and release them in
`close_reader`, with allocator checks before and after the reader is closed.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59929#discussion_r4425737882).
**Verification:**
- **Regression:** With a library that moves the ranges and releases them
when the reader is closed, the query returns the right rows, the library reads
valid ranges after the call, the Arrow allocator still holds them until the
reader is closed, and it is back to its baseline afterwards.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnDomains.scala:
##########
@@ -0,0 +1,547 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.v2.columnar
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.types.PhysicalDataType
+import org.apache.spark.sql.catalyst.util.quoteIfNeeded
+import org.apache.spark.sql.connector.expressions.{Expression, LiteralValue,
NamedReference}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The values that a column can have in the rows that match the filters of a
query: null if
+ * `nullAllowed`, NaN if `nanAllowed`, for a float or double column, and the
other values in one
+ * of the `ranges`, or any of them without ranges. A data source can skip the
data whose statistics
+ * show that it has no value in the domain of a column, but never a row whose
values are in their
+ * domains: Spark evaluates the filters after reading anyway.
+ *
+ * @param path the index of the column in the read schema, and then the index
of the field in each
+ * struct, for a field of a struct
+ * @param name the names of the column and of its fields
+ * @param ranges sorted ranges, which neither overlap nor touch, of the values
other than null and
+ * NaN, in the internal representation of Spark. None if the
column can have any of
+ * them, and empty if it can have none.
+ */
+case class ColumnDomain(
+ path: Seq[Int],
+ name: Seq[String],
+ dataType: DataType,
+ nullAllowed: Boolean,
+ nanAllowed: Boolean,
+ ranges: Option[Seq[ValueRange]]) {
+
+ private def ordering: Ordering[Any] = PhysicalDataType.ordering(dataType)
+
+ /** Whether the domain has the value, in the internal representation of
Spark. */
+ def contains(value: Any): Boolean = value match {
+ case null => nullAllowed
+ case f: Float if f.isNaN => nanAllowed
+ case d: Double if d.isNaN => nanAllowed
+ case _ => ranges.forall(_.exists(_.contains(value, ordering)))
+ }
+
+ override def toString: String = {
+ def literal(value: Any): String = Literal(value, dataType).sql
+ val values = ranges match {
+ case None => Seq("ANY")
+ case Some(ranges) => ranges.map {
+ case ValueRange(Some(low), true, Some(high), true) if
ordering.equiv(low, high) =>
+ literal(low)
+ case range =>
+ (if (range.lowInclusive) "[" else "(") +
range.low.map(literal).getOrElse("-inf") +
+ ", " + range.high.map(literal).getOrElse("+inf") +
+ (if (range.highInclusive) "]" else ")")
+ }
+ }
+ val special = (if (nullAllowed) Seq("NULL") else Nil) ++ (if (nanAllowed)
Seq("NaN") else Nil)
+ val all = special ++ values
+ s"${name.map(quoteIfNeeded).mkString(".")}: ${if (all.isEmpty) "NONE" else
all.mkString(", ")}"
+ }
+}
+
+/**
+ * A range of values in the internal representation of Spark: the values above
`low`, or equal to
+ * it if `lowInclusive`, and below `high`, or equal to it if `highInclusive`.
A bound that is None
+ * is unbounded, and is not inclusive.
+ */
+case class ValueRange(low: Option[Any], lowInclusive: Boolean, high:
Option[Any],
+ highInclusive: Boolean) {
+
+ def contains(value: Any, ordering: Ordering[Any]): Boolean = {
+ low.forall { l =>
+ val c = ordering.compare(value, l)
+ c > 0 || (c == 0 && lowInclusive)
+ } && high.forall { h =>
+ val c = ordering.compare(value, h)
+ c < 0 || (c == 0 && highInclusive)
+ }
+ }
+}
+
+/**
+ * The domains that a query passes to a data source, and the predicates that
each of them comes
+ * from.
+ *
+ * @param domains the domains, at most one for each column, sorted by path
+ * @param columns the paths of the columns that each predicate constrains
+ */
+case class ColumnDomains(domains: Seq[ColumnDomain], columns:
Seq[Set[Seq[Int]]]) {
+
+ /**
+ * Returns the predicates that constrain only the columns whose domains are
used, given whether
+ * the data source uses each domain.
+ */
+ def pushedPredicates(predicates: Array[Predicate], used: Seq[Boolean]):
Array[Predicate] = {
+ val usedPaths = domains.zip(used).collect { case (domain, true) =>
domain.path }.toSet
+ predicates.zip(columns).collect {
+ case (predicate, paths) if paths.nonEmpty && paths.subsetOf(usedPaths)
=> predicate
+ }
+ }
+}
+
+/**
+ * Derives the domains of the columns that the predicates of a query
constrain, the ones that Data
+ * Source V2 pushes to a scan, combined with AND. A domain is a superset of
the values that a column
+ * has in the rows for which all the predicates are true: Spark derives it
from the comparisons of
+ * a column with literals, IN, IS NULL, IS NOT NULL and STARTS_WITH, under
AND, OR and NOT, with the
+ * semantics of Spark SQL. Comparisons with null are null, NaN is equal to NaN
and greater than any
+ * other value, -0.0 is equal to 0.0, and strings and binary values are
compared by their bytes, as
+ * unsigned bytes. A predicate that cannot be expressed this way does not
constrain any column. Only
+ * the columns and the fields of structs of the types that a data source can
compare get a domain:
+ * not the strings of a non-binary collation, nor arrays, maps and structs. A
field of a struct only
+ * gets a domain that does not have null.
+ */
+object ColumnDomains {
+
+ // A column that a predicate constrains, with its type.
+ private case class Column(path: Seq[Int], name: Seq[String], dataType:
DataType) {
+ val ordering: Ordering[Any] = PhysicalDataType.ordering(dataType)
+ val isFloating: Boolean = dataType == FloatType || dataType == DoubleType
+ }
+
+ // The values that a column can have: see ColumnDomain.
+ private case class Values(nullAllowed: Boolean, nanAllowed: Boolean,
+ ranges: Option[Seq[ValueRange]])
+
+ private val NoValue = Values(nullAllowed = false, nanAllowed = false,
Some(Nil))
+ private val OnlyNull = Values(nullAllowed = true, nanAllowed = false,
Some(Nil))
+
+ // The columns that the predicates constrain, with their values: a row can
only make the
+ // predicates true if each of these columns has one of the values. None when
no row can.
+ private type Constraint = Option[Map[Column, Values]]
+ private val All: Constraint = Some(Map.empty)
+ private val NoRow: Constraint = None
+
+ /** Derives the domains of the predicates for a scan of the schema. */
+ def derive(predicates: Seq[Predicate], schema: StructType): ColumnDomains = {
+ val constraints = predicates.map(constraint(_, negated = false, schema))
+ val columns =
constraints.map(_.map(_.keySet.map(_.path)).getOrElse(Set.empty[Seq[Int]]))
+ val domains = constraints.foldLeft(All)(and) match {
+ case Some(values) =>
+ values.toSeq.collect {
+ // A field of a struct is null in the rows where the struct, or a
struct that contains it,
+ // is null, which the statistics of its Arrow child array may not
count as nulls. So a
+ // field only gets a domain without null, which these rows never
match.
+ case (column, v) if column.path.length == 1 || !v.nullAllowed =>
+ ColumnDomain(column.path, column.name, column.dataType,
v.nullAllowed, v.nanAllowed,
+ v.ranges)
+ }.sortBy(_.path)(Ordering.Implicits.seqOrdering[Seq, Int])
+ // A predicate that is always false, which Spark usually removes before
it scans the data.
+ case None => Nil
+ }
+ ColumnDomains(domains, columns)
+ }
+
+ // The rows for which the predicate is true, or false if negated: NOT moves
to the leaves.
+ private def constraint(predicate: Predicate, negated: Boolean, schema:
StructType): Constraint = {
+ def child(i: Int): Option[Predicate] =
predicate.children().lift(i).collect {
+ case p: Predicate => p
+ }
+ def both(combine: (Constraint, Constraint) => Constraint, negated:
Boolean): Constraint = {
+ (child(0), child(1)) match {
+ case (Some(left), Some(right)) =>
+ combine(constraint(left, negated, schema), constraint(right,
negated, schema))
+ case _ => All
+ }
+ }
+ predicate.name() match {
+ case "AND" => if (negated) both(or, negated = true) else both(and,
negated = false)
+ case "OR" => if (negated) both(and, negated = true) else both(or,
negated = false)
+ case "NOT" => child(0).map(constraint(_, !negated,
schema)).getOrElse(All)
+ case "ALWAYS_TRUE" => if (negated) NoRow else All
+ case "ALWAYS_FALSE" => if (negated) All else NoRow
+ case _ => leaf(predicate, negated, schema).getOrElse(All)
+ }
+ }
+
+ private def leaf(
+ predicate: Predicate,
+ negated: Boolean,
+ schema: StructType): Option[Constraint] = {
+ val children = predicate.children()
+ predicate.name() match {
+ case "IS_NULL" | "IS_NOT_NULL" if children.length == 1 =>
+ column(children(0), schema).map { col =>
+ val isNull = (predicate.name() == "IS_NULL") != negated
+ single(col, if (isNull) OnlyNull else anyValue(col))
+ }
+ case op @ ("=" | "<>" | "<=>" | "<" | "<=" | ">" | ">=") if
children.length == 2 =>
+ val operands = (column(children(0), schema), column(children(1),
schema)) match {
+ case (Some(col), _) => literal(children(1), col).map((col, _, op))
+ case (_, Some(col)) => literal(children(0), col).map((col, _,
flip(op)))
+ case _ => None
+ }
+ operands.map { case (col, value, op) => compare(col, value, op,
negated) }
+ case "IN" if children.nonEmpty =>
+ column(children(0), schema).flatMap { col =>
+ val values = children.tail.map(literal(_, col))
+ if (values.forall(_.isDefined)) Some(in(col,
values.map(_.get).toSeq, negated)) else None
+ }
+ case "STARTS_WITH" if children.length == 2 =>
+ column(children(0),
schema).filter(_.dataType.isInstanceOf[StringType]).flatMap { col =>
+ literal(children(1), col).map(prefix => startsWith(col, prefix,
negated))
+ }
+ case _ => None
+ }
+ }
+
+ private def flip(op: String): String = op match {
+ case "<" => ">"
+ case "<=" => ">="
+ case ">" => "<"
+ case ">=" => "<="
+ case other => other
+ }
+
+ private def compare(column: Column, value: Any, op: String, negated:
Boolean): Constraint = {
+ // NOT (x <=> v) is true for null, unless v is null.
+ val nullAllowed = op == "<=>" && negated && value != null
+ if (value == null) {
+ // A comparison with null is null, and so is its negation, but x <=>
NULL is x IS NULL.
+ if (op == "<=>") single(column, if (negated) anyValue(column) else
OnlyNull)
+ else single(column, NoValue)
+ } else if (isNaN(value)) {
+ // NaN is equal to NaN and greater than any other value: (NaN, other
values).
+ val (nan, others) = (op, negated) match {
+ case ("=" | "<=>" | ">=", false) | ("<>" | "<", true) => (true,
Some(Nil))
+ case ("=" | "<=>" | ">=", true) | ("<>" | "<", false) => (false, None)
+ case ("<=", false) | (">", true) => (true, None)
+ case ("<=", true) | (">", false) => (false, Some(Nil))
+ case _ => throw SparkException.internalError(s"Unexpected comparison
$op.")
+ }
+ single(column, Values(nullAllowed, nan, others))
+ } else {
+ val v = normalize(value)
+ // NaN is greater than v, and not equal to it.
+ val (nan, ranges) = (op, negated) match {
+ case ("=" | "<=>", false) | ("<>", true) => (false, Seq(point(v)))
+ case ("=" | "<=>", true) | ("<>", false) =>
+ (true, Seq(below(v, inclusive = false), above(v, inclusive = false)))
+ case ("<", false) | (">=", true) => (false, Seq(below(v, inclusive =
false)))
+ case ("<=", false) | (">", true) => (false, Seq(below(v, inclusive =
true)))
+ case (">", false) | ("<=", true) => (true, Seq(above(v, inclusive =
false)))
+ case (">=", false) | ("<", true) => (true, Seq(above(v, inclusive =
true)))
+ case _ => throw SparkException.internalError(s"Unexpected comparison
$op.")
+ }
+ single(column, Values(nullAllowed, nan, Some(ranges)))
+ }
+ }
+
+ private def in(column: Column, values: Seq[Any], negated: Boolean):
Constraint = {
+ val nonNull = values.filter(_ != null)
+ val hasNaN = nonNull.exists(isNaN)
+ val points =
nonNull.filterNot(isNaN).map(normalize).sorted(column.ordering)
+ .foldLeft(List.empty[Any]) {
+ case (last :: rest, v) if column.ordering.equiv(last, v) => last ::
rest
+ case (acc, v) => v :: acc
+ }.reverse
+ if (values.isEmpty) {
+ // x IN () is false, also for null, unless
spark.sql.legacy.nullInEmptyListBehavior makes it
+ // null for null. So it is never true, and NOT (x IN ()) is true for any
value, and by
+ // default for null.
+ if (negated) All else single(column, NoValue)
+ } else if (!negated) {
+ single(column, Values(nullAllowed = false, hasNaN,
Some(points.map(point))))
+ } else if (nonNull.length < values.length) {
+ // NOT (x IN (..., NULL)) is never true: it is false or null.
+ single(column, NoValue)
+ } else if (points.isEmpty) {
+ // NOT (x IN (NaN)) is true for the values other than NaN.
+ single(column, Values(nullAllowed = false, !hasNaN, None))
+ } else {
+ // The values between the points.
+ val bounds = None +: points.map(Some(_)) :+ None
+ val gaps = bounds.zip(bounds.tail).map { case (low, high) =>
+ ValueRange(low, lowInclusive = false, high = high, highInclusive =
false)
+ }
+ single(column, Values(nullAllowed = false, !hasNaN, Some(gaps)))
+ }
+ }
+
+ private def startsWith(column: Column, prefix: Any, negated: Boolean):
Constraint = {
+ (prefix, negated) match {
+ case (null, _) => single(column, NoValue)
+ // The strings that do not start with the prefix are not a range. A
value whose bytes are not
+ // valid UTF-8 could even be in the range of the strings that do.
+ case (_, true) => single(column, anyValue(column))
+ case (prefix: UTF8String, false) if prefix.numBytes() == 0 =>
single(column, anyValue(column))
+ case (prefix: UTF8String, false) =>
+ val range = ValueRange(Some(prefix), lowInclusive = true, high =
nextPrefix(prefix),
+ highInclusive = false)
+ single(column, Values(nullAllowed = false, nanAllowed = false,
Some(Seq(range))))
+ case _ => All
+ }
+ }
+
+ /**
+ * Returns the smallest string that is greater than all the strings that
start with the prefix,
+ * in the order of their UTF-8 bytes, which is the order of their code
points, or None if there
+ * is none: the prefix with its last code point incremented, after dropping
the trailing
+ * U+10FFFF, the largest code point.
+ */
+ private[columnar] def nextPrefix(prefix: UTF8String): Option[UTF8String] = {
+ val codePoints = prefix.toString.codePoints().toArray.toBuffer
+ while (codePoints.nonEmpty && codePoints.last == Character.MAX_CODE_POINT)
{
+ codePoints.remove(codePoints.length - 1)
+ }
+ if (codePoints.isEmpty) {
+ None
+ } else {
+ val last = codePoints.last + 1
+ // Skips the surrogates, which are not code points of characters.
+ codePoints(codePoints.length - 1) =
+ if (last >= Character.MIN_SURROGATE && last <=
Character.MAX_SURROGATE) {
+ Character.MAX_SURROGATE + 1
+ } else {
+ last
+ }
+ Some(UTF8String.fromString(new String(codePoints.toArray, 0,
codePoints.length)))
+ }
+ }
+
+ private def single(column: Column, values: Values): Constraint = {
+ // Only floating-point columns have NaN.
+ val normalized = values.copy(nanAllowed = values.nanAllowed &&
column.isFloating)
+ if (isAll(column, normalized)) All else Some(Map(column -> normalized))
+ }
+
+ private def point(v: Any): ValueRange =
+ ValueRange(Some(v), lowInclusive = true, high = Some(v), highInclusive =
true)
+
+ private def below(v: Any, inclusive: Boolean): ValueRange =
+ ValueRange(None, lowInclusive = false, high = Some(v), highInclusive =
inclusive)
+
+ private def above(v: Any, inclusive: Boolean): ValueRange =
+ ValueRange(Some(v), lowInclusive = inclusive, high = None, highInclusive =
false)
+
+ private def anyValue(column: Column): Values = {
+ Values(nullAllowed = false, nanAllowed = column.isFloating, None)
+ }
+
+ private def isAll(column: Column, values: Values): Boolean = {
+ values.nullAllowed && values.ranges.isEmpty && (values.nanAllowed ||
!column.isFloating)
+ }
+
+ private def isEmpty(values: Values): Boolean = {
+ !values.nullAllowed && !values.nanAllowed &&
values.ranges.exists(_.isEmpty)
+ }
+
+ // Whether no row can make the predicates true.
+ private def matchesNoRow(constraint: Constraint): Boolean = {
+ constraint.forall(_.values.exists(isEmpty))
+ }
+
+ private def and(left: Constraint, right: Constraint): Constraint = (left,
right) match {
+ case (Some(l), Some(r)) =>
+ Some((l.keySet ++ r.keySet).map { column =>
Review Comment:
**Nit (P3):** `(l.keySet ++ r.keySet).map { ... }` builds a `Set` of
`(Column, Values)` pairs before `.toMap`. From the fifth element on, Scala
2.13's `Set` builder switches to a `HashSet` and hashes every pair, and the
case-class hash of `Values` walks its whole list of ranges; a `Decimal` bound
even builds a `BigDecimal` to hash. `derive` folds every pushed predicate
through `and`, so once five columns are constrained, each further predicate
rehashes all the ranges of an earlier `IN` list. An example is `d IN (<10,000
decimals>) AND a > 1 AND b > 1 AND c > 1 AND e > 1` with its inferred `IS NOT
NULL` filters.
Mapping through an iterator, `(l.keySet ++ r.keySet).iterator.map { ...
}.toMap`, or folding `r` into `l`, gives the same map and hashes only the
columns. `or` builds its map the same way.
**Verification:**
- **Regression:** Deriving the domains of an IN list of thousands of decimal
values combined with predicates on five or more other columns takes time
proportional to the number of values, not to the number of values times the
number of predicates.
- **Behavior:** The domains derived in the existing rule cases and in the
randomized check against Spark's evaluation are unchanged.
##########
docs/sql-native-data-sources.md:
##########
@@ -169,12 +170,67 @@ a write that runs for long, or blocks, for example on
network I/O, should check
regularly and use timeouts. The task reaper, `spark.task.reaper.enabled` and
`spark.task.reaper.killTimeout`, kills an executor whose killed task does not
end.
+### Filter Pushdown
+
+Spark does not pass the filters of a query to a library as expressions, which
a library would have
+to evaluate with the exact semantics of Spark SQL. It passes the *domains* of
the columns that the
+filters constrain instead: for each column, the values that it can have in the
rows that match,
+which a library compares with the statistics of its data to skip what cannot
match, such as files,
+row groups or blocks. With `push_domains`, a reader gets a `SparkDsDomain` for
each column, with:
+
+- `path`: the index of the column in the schema to read, and then the index of
the field in each
+ struct, for a field of a struct;
+- `null_allowed` and `nan_allowed`: whether null and NaN are in the domain;
+- `ranges`: the ranges of the other values in the domain, as a struct array
whose fields are
+ `low` and `high`, of the Arrow type of the column, with a null bound for no
bound, and
+ `low_inclusive` and `high_inclusive`. The ranges are sorted and do not
overlap. A single value
+ is a range whose bounds are equal and inclusive, and `ranges` is `NULL` when
all the other
+ values are in the domain.
+
+For example:
+
+| Filter | Domain |
+|--------|--------|
+| `id > 5 AND id <= 10` | `id`: `(5, 10]` |
+| `id IN (1, 3) OR id IS NULL` | `id`: null, `[1, 1]`, `[3, 3]` |
+| `NOT (id = 5)` | `id`: `(-inf, 5)`, `(5, +inf)` |
+| `value > 1.5`, for a `double` column | `value`: NaN, `(1.5, +inf)` |
+| `name LIKE 'ab%'`, which is `STARTS_WITH(name, 'ab')` | `name`: `['ab',
'ac')` |
+| `s.x = 1 AND name IS NOT NULL` | `s.x`: `[1, 1]`; `name`: all the values but
null |
+| `id > 5 OR name = 'a'` | none: each side only constrains one of the columns |
+
+Spark derives the domains from the comparisons of a column with literals (`=`,
`<>`, `<=>`, `<`,
+`<=`, `>`, `>=`), `IN`, `IS NULL`, `IS NOT NULL` and `STARTS_WITH`, combined
with `AND`, `OR` and
+`NOT`, with the semantics of Spark SQL: comparisons with null are null, so that
+`NOT (id IN (1, NULL))` matches no row, NaN is equal to NaN and greater than
any other value, and
+-0.0 is equal to 0.0. A domain has all the values of its column in the rows
for which the filters
+are true, and may have others: a column that only one side of an `OR`
constrains gets no domain
+from it, and the other filters, such as `id + 1 > 5`, `a = b` or
`ENDS_WITH(name, 'x')`, constrain
+no column. Columns of strings with a non-binary collation, such as
`UTF8_LCASE`, and arrays, maps
+and structs, get no domain. A field of a struct is null in the rows where the
struct, or a struct
+that contains it, is null, whatever the Arrow child array of the field holds
there, so statistics
+computed from that array may not count these rows as nulls: Spark only passes
domains without null
+for the fields of structs.
+
+A library compares values with the order of Spark SQL: numbers, dates and
timestamps by their
+values, with -0.0 equal to 0.0 (use IEEE 754 comparisons, not a total order
such as
+`f64::total_cmp`), strings by their UTF-8 bytes, binary values as unsigned
bytes, and false before
+true. It must never skip a row whose values are all in their domains, which
Spark cannot detect,
+but it can return rows that are not: Spark evaluates all the filters again
after reading. So a
+library only uses the domains to read less data, and Spark does not push a
`LIMIT` down with a
+filter. The library sets `used` to whether it uses each domain, and the plan
of the query shows the
Review Comment:
**Nit (P3):** `ColumnarScan` exposes `PushedFilters` and `PushedDomains`
only through `getMetaData`, which only
`DataSourceV2ScanExecBase.verboseStringWithOperatorId` prints.
`BatchScanExec.simpleString`, which `df.explain()`, `EXPLAIN` and `EXPLAIN
EXTENDED` print, shows the output, the scan's description and `RuntimeFilters`,
and here the description is just the short name of the data source. So a
library author who checks with `df.explain()` whether Spark pushed a domain
sees neither entry and would conclude that nothing was pushed. They only appear
with `EXPLAIN FORMATTED`, `df.explain("formatted")` and in the SQL UI.
The `push_domains` comment in `spark_datasource.h` makes the same claim.
Naming the formatted plan and the SQL UI in both places fixes it.
**Verification:**
- **Inspection:** The guide's and the header's statements about where
PushedFilters and PushedDomains appear match a native scan's metadata, which
only formatted explain and the SQL UI print.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnDomains.scala:
##########
@@ -0,0 +1,547 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.v2.columnar
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.types.PhysicalDataType
+import org.apache.spark.sql.catalyst.util.quoteIfNeeded
+import org.apache.spark.sql.connector.expressions.{Expression, LiteralValue,
NamedReference}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The values that a column can have in the rows that match the filters of a
query: null if
+ * `nullAllowed`, NaN if `nanAllowed`, for a float or double column, and the
other values in one
+ * of the `ranges`, or any of them without ranges. A data source can skip the
data whose statistics
+ * show that it has no value in the domain of a column, but never a row whose
values are in their
+ * domains: Spark evaluates the filters after reading anyway.
+ *
+ * @param path the index of the column in the read schema, and then the index
of the field in each
+ * struct, for a field of a struct
+ * @param name the names of the column and of its fields
+ * @param ranges sorted ranges, which neither overlap nor touch, of the values
other than null and
+ * NaN, in the internal representation of Spark. None if the
column can have any of
+ * them, and empty if it can have none.
+ */
+case class ColumnDomain(
+ path: Seq[Int],
+ name: Seq[String],
+ dataType: DataType,
+ nullAllowed: Boolean,
+ nanAllowed: Boolean,
+ ranges: Option[Seq[ValueRange]]) {
+
+ private def ordering: Ordering[Any] = PhysicalDataType.ordering(dataType)
+
+ /** Whether the domain has the value, in the internal representation of
Spark. */
+ def contains(value: Any): Boolean = value match {
+ case null => nullAllowed
+ case f: Float if f.isNaN => nanAllowed
+ case d: Double if d.isNaN => nanAllowed
+ case _ => ranges.forall(_.exists(_.contains(value, ordering)))
+ }
+
+ override def toString: String = {
+ def literal(value: Any): String = Literal(value, dataType).sql
+ val values = ranges match {
+ case None => Seq("ANY")
+ case Some(ranges) => ranges.map {
+ case ValueRange(Some(low), true, Some(high), true) if
ordering.equiv(low, high) =>
+ literal(low)
+ case range =>
+ (if (range.lowInclusive) "[" else "(") +
range.low.map(literal).getOrElse("-inf") +
+ ", " + range.high.map(literal).getOrElse("+inf") +
+ (if (range.highInclusive) "]" else ")")
+ }
+ }
+ val special = (if (nullAllowed) Seq("NULL") else Nil) ++ (if (nanAllowed)
Seq("NaN") else Nil)
+ val all = special ++ values
+ s"${name.map(quoteIfNeeded).mkString(".")}: ${if (all.isEmpty) "NONE" else
all.mkString(", ")}"
+ }
+}
+
+/**
+ * A range of values in the internal representation of Spark: the values above
`low`, or equal to
+ * it if `lowInclusive`, and below `high`, or equal to it if `highInclusive`.
A bound that is None
+ * is unbounded, and is not inclusive.
+ */
+case class ValueRange(low: Option[Any], lowInclusive: Boolean, high:
Option[Any],
+ highInclusive: Boolean) {
+
+ def contains(value: Any, ordering: Ordering[Any]): Boolean = {
+ low.forall { l =>
+ val c = ordering.compare(value, l)
+ c > 0 || (c == 0 && lowInclusive)
+ } && high.forall { h =>
+ val c = ordering.compare(value, h)
+ c < 0 || (c == 0 && highInclusive)
+ }
+ }
+}
+
+/**
+ * The domains that a query passes to a data source, and the predicates that
each of them comes
+ * from.
+ *
+ * @param domains the domains, at most one for each column, sorted by path
+ * @param columns the paths of the columns that each predicate constrains
+ */
+case class ColumnDomains(domains: Seq[ColumnDomain], columns:
Seq[Set[Seq[Int]]]) {
+
+ /**
+ * Returns the predicates that constrain only the columns whose domains are
used, given whether
+ * the data source uses each domain.
+ */
+ def pushedPredicates(predicates: Array[Predicate], used: Seq[Boolean]):
Array[Predicate] = {
+ val usedPaths = domains.zip(used).collect { case (domain, true) =>
domain.path }.toSet
+ predicates.zip(columns).collect {
+ case (predicate, paths) if paths.nonEmpty && paths.subsetOf(usedPaths)
=> predicate
+ }
+ }
+}
+
+/**
+ * Derives the domains of the columns that the predicates of a query
constrain, the ones that Data
+ * Source V2 pushes to a scan, combined with AND. A domain is a superset of
the values that a column
+ * has in the rows for which all the predicates are true: Spark derives it
from the comparisons of
+ * a column with literals, IN, IS NULL, IS NOT NULL and STARTS_WITH, under
AND, OR and NOT, with the
+ * semantics of Spark SQL. Comparisons with null are null, NaN is equal to NaN
and greater than any
+ * other value, -0.0 is equal to 0.0, and strings and binary values are
compared by their bytes, as
+ * unsigned bytes. A predicate that cannot be expressed this way does not
constrain any column. Only
+ * the columns and the fields of structs of the types that a data source can
compare get a domain:
+ * not the strings of a non-binary collation, nor arrays, maps and structs. A
field of a struct only
+ * gets a domain that does not have null.
+ */
+object ColumnDomains {
+
+ // A column that a predicate constrains, with its type.
+ private case class Column(path: Seq[Int], name: Seq[String], dataType:
DataType) {
+ val ordering: Ordering[Any] = PhysicalDataType.ordering(dataType)
+ val isFloating: Boolean = dataType == FloatType || dataType == DoubleType
+ }
+
+ // The values that a column can have: see ColumnDomain.
+ private case class Values(nullAllowed: Boolean, nanAllowed: Boolean,
+ ranges: Option[Seq[ValueRange]])
+
+ private val NoValue = Values(nullAllowed = false, nanAllowed = false,
Some(Nil))
+ private val OnlyNull = Values(nullAllowed = true, nanAllowed = false,
Some(Nil))
+
+ // The columns that the predicates constrain, with their values: a row can
only make the
+ // predicates true if each of these columns has one of the values. None when
no row can.
+ private type Constraint = Option[Map[Column, Values]]
+ private val All: Constraint = Some(Map.empty)
+ private val NoRow: Constraint = None
+
+ /** Derives the domains of the predicates for a scan of the schema. */
+ def derive(predicates: Seq[Predicate], schema: StructType): ColumnDomains = {
+ val constraints = predicates.map(constraint(_, negated = false, schema))
+ val columns =
constraints.map(_.map(_.keySet.map(_.path)).getOrElse(Set.empty[Seq[Int]]))
+ val domains = constraints.foldLeft(All)(and) match {
+ case Some(values) =>
+ values.toSeq.collect {
+ // A field of a struct is null in the rows where the struct, or a
struct that contains it,
+ // is null, which the statistics of its Arrow child array may not
count as nulls. So a
+ // field only gets a domain without null, which these rows never
match.
+ case (column, v) if column.path.length == 1 || !v.nullAllowed =>
+ ColumnDomain(column.path, column.name, column.dataType,
v.nullAllowed, v.nanAllowed,
+ v.ranges)
+ }.sortBy(_.path)(Ordering.Implicits.seqOrdering[Seq, Int])
+ // A predicate that is always false, which Spark usually removes before
it scans the data.
+ case None => Nil
+ }
+ ColumnDomains(domains, columns)
+ }
+
+ // The rows for which the predicate is true, or false if negated: NOT moves
to the leaves.
+ private def constraint(predicate: Predicate, negated: Boolean, schema:
StructType): Constraint = {
+ def child(i: Int): Option[Predicate] =
predicate.children().lift(i).collect {
+ case p: Predicate => p
+ }
+ def both(combine: (Constraint, Constraint) => Constraint, negated:
Boolean): Constraint = {
+ (child(0), child(1)) match {
+ case (Some(left), Some(right)) =>
+ combine(constraint(left, negated, schema), constraint(right,
negated, schema))
+ case _ => All
+ }
+ }
+ predicate.name() match {
+ case "AND" => if (negated) both(or, negated = true) else both(and,
negated = false)
+ case "OR" => if (negated) both(and, negated = true) else both(or,
negated = false)
+ case "NOT" => child(0).map(constraint(_, !negated,
schema)).getOrElse(All)
+ case "ALWAYS_TRUE" => if (negated) NoRow else All
+ case "ALWAYS_FALSE" => if (negated) All else NoRow
+ case _ => leaf(predicate, negated, schema).getOrElse(All)
+ }
+ }
+
+ private def leaf(
+ predicate: Predicate,
+ negated: Boolean,
+ schema: StructType): Option[Constraint] = {
+ val children = predicate.children()
+ predicate.name() match {
+ case "IS_NULL" | "IS_NOT_NULL" if children.length == 1 =>
+ column(children(0), schema).map { col =>
+ val isNull = (predicate.name() == "IS_NULL") != negated
+ single(col, if (isNull) OnlyNull else anyValue(col))
+ }
+ case op @ ("=" | "<>" | "<=>" | "<" | "<=" | ">" | ">=") if
children.length == 2 =>
+ val operands = (column(children(0), schema), column(children(1),
schema)) match {
+ case (Some(col), _) => literal(children(1), col).map((col, _, op))
+ case (_, Some(col)) => literal(children(0), col).map((col, _,
flip(op)))
+ case _ => None
+ }
+ operands.map { case (col, value, op) => compare(col, value, op,
negated) }
+ case "IN" if children.nonEmpty =>
+ column(children(0), schema).flatMap { col =>
+ val values = children.tail.map(literal(_, col))
+ if (values.forall(_.isDefined)) Some(in(col,
values.map(_.get).toSeq, negated)) else None
+ }
+ case "STARTS_WITH" if children.length == 2 =>
+ column(children(0),
schema).filter(_.dataType.isInstanceOf[StringType]).flatMap { col =>
+ literal(children(1), col).map(prefix => startsWith(col, prefix,
negated))
+ }
+ case _ => None
+ }
+ }
+
+ private def flip(op: String): String = op match {
+ case "<" => ">"
+ case "<=" => ">="
+ case ">" => "<"
+ case ">=" => "<="
+ case other => other
+ }
+
+ private def compare(column: Column, value: Any, op: String, negated:
Boolean): Constraint = {
+ // NOT (x <=> v) is true for null, unless v is null.
+ val nullAllowed = op == "<=>" && negated && value != null
+ if (value == null) {
+ // A comparison with null is null, and so is its negation, but x <=>
NULL is x IS NULL.
+ if (op == "<=>") single(column, if (negated) anyValue(column) else
OnlyNull)
+ else single(column, NoValue)
+ } else if (isNaN(value)) {
+ // NaN is equal to NaN and greater than any other value: (NaN, other
values).
+ val (nan, others) = (op, negated) match {
+ case ("=" | "<=>" | ">=", false) | ("<>" | "<", true) => (true,
Some(Nil))
+ case ("=" | "<=>" | ">=", true) | ("<>" | "<", false) => (false, None)
+ case ("<=", false) | (">", true) => (true, None)
+ case ("<=", true) | (">", false) => (false, Some(Nil))
+ case _ => throw SparkException.internalError(s"Unexpected comparison
$op.")
+ }
+ single(column, Values(nullAllowed, nan, others))
+ } else {
+ val v = normalize(value)
+ // NaN is greater than v, and not equal to it.
+ val (nan, ranges) = (op, negated) match {
+ case ("=" | "<=>", false) | ("<>", true) => (false, Seq(point(v)))
+ case ("=" | "<=>", true) | ("<>", false) =>
+ (true, Seq(below(v, inclusive = false), above(v, inclusive = false)))
+ case ("<", false) | (">=", true) => (false, Seq(below(v, inclusive =
false)))
+ case ("<=", false) | (">", true) => (false, Seq(below(v, inclusive =
true)))
+ case (">", false) | ("<=", true) => (true, Seq(above(v, inclusive =
false)))
+ case (">=", false) | ("<", true) => (true, Seq(above(v, inclusive =
true)))
+ case _ => throw SparkException.internalError(s"Unexpected comparison
$op.")
+ }
+ single(column, Values(nullAllowed, nan, Some(ranges)))
+ }
+ }
+
+ private def in(column: Column, values: Seq[Any], negated: Boolean):
Constraint = {
+ val nonNull = values.filter(_ != null)
+ val hasNaN = nonNull.exists(isNaN)
+ val points =
nonNull.filterNot(isNaN).map(normalize).sorted(column.ordering)
+ .foldLeft(List.empty[Any]) {
+ case (last :: rest, v) if column.ordering.equiv(last, v) => last ::
rest
+ case (acc, v) => v :: acc
+ }.reverse
+ if (values.isEmpty) {
+ // x IN () is false, also for null, unless
spark.sql.legacy.nullInEmptyListBehavior makes it
+ // null for null. So it is never true, and NOT (x IN ()) is true for any
value, and by
+ // default for null.
+ if (negated) All else single(column, NoValue)
+ } else if (!negated) {
+ single(column, Values(nullAllowed = false, hasNaN,
Some(points.map(point))))
+ } else if (nonNull.length < values.length) {
+ // NOT (x IN (..., NULL)) is never true: it is false or null.
+ single(column, NoValue)
+ } else if (points.isEmpty) {
+ // NOT (x IN (NaN)) is true for the values other than NaN.
+ single(column, Values(nullAllowed = false, !hasNaN, None))
+ } else {
+ // The values between the points.
+ val bounds = None +: points.map(Some(_)) :+ None
+ val gaps = bounds.zip(bounds.tail).map { case (low, high) =>
+ ValueRange(low, lowInclusive = false, high = high, highInclusive =
false)
+ }
+ single(column, Values(nullAllowed = false, !hasNaN, Some(gaps)))
+ }
+ }
+
+ private def startsWith(column: Column, prefix: Any, negated: Boolean):
Constraint = {
+ (prefix, negated) match {
+ case (null, _) => single(column, NoValue)
+ // The strings that do not start with the prefix are not a range. A
value whose bytes are not
+ // valid UTF-8 could even be in the range of the strings that do.
+ case (_, true) => single(column, anyValue(column))
+ case (prefix: UTF8String, false) if prefix.numBytes() == 0 =>
single(column, anyValue(column))
+ case (prefix: UTF8String, false) =>
+ val range = ValueRange(Some(prefix), lowInclusive = true, high =
nextPrefix(prefix),
+ highInclusive = false)
+ single(column, Values(nullAllowed = false, nanAllowed = false,
Some(Seq(range))))
+ case _ => All
+ }
+ }
+
+ /**
+ * Returns the smallest string that is greater than all the strings that
start with the prefix,
+ * in the order of their UTF-8 bytes, which is the order of their code
points, or None if there
+ * is none: the prefix with its last code point incremented, after dropping
the trailing
+ * U+10FFFF, the largest code point.
+ */
+ private[columnar] def nextPrefix(prefix: UTF8String): Option[UTF8String] = {
+ val codePoints = prefix.toString.codePoints().toArray.toBuffer
+ while (codePoints.nonEmpty && codePoints.last == Character.MAX_CODE_POINT)
{
+ codePoints.remove(codePoints.length - 1)
+ }
+ if (codePoints.isEmpty) {
+ None
+ } else {
+ val last = codePoints.last + 1
+ // Skips the surrogates, which are not code points of characters.
+ codePoints(codePoints.length - 1) =
+ if (last >= Character.MIN_SURROGATE && last <=
Character.MAX_SURROGATE) {
+ Character.MAX_SURROGATE + 1
+ } else {
+ last
+ }
+ Some(UTF8String.fromString(new String(codePoints.toArray, 0,
codePoints.length)))
+ }
+ }
+
+ private def single(column: Column, values: Values): Constraint = {
+ // Only floating-point columns have NaN.
+ val normalized = values.copy(nanAllowed = values.nanAllowed &&
column.isFloating)
+ if (isAll(column, normalized)) All else Some(Map(column -> normalized))
+ }
+
+ private def point(v: Any): ValueRange =
+ ValueRange(Some(v), lowInclusive = true, high = Some(v), highInclusive =
true)
+
+ private def below(v: Any, inclusive: Boolean): ValueRange =
+ ValueRange(None, lowInclusive = false, high = Some(v), highInclusive =
inclusive)
+
+ private def above(v: Any, inclusive: Boolean): ValueRange =
+ ValueRange(Some(v), lowInclusive = inclusive, high = None, highInclusive =
false)
+
+ private def anyValue(column: Column): Values = {
+ Values(nullAllowed = false, nanAllowed = column.isFloating, None)
+ }
+
+ private def isAll(column: Column, values: Values): Boolean = {
+ values.nullAllowed && values.ranges.isEmpty && (values.nanAllowed ||
!column.isFloating)
+ }
+
+ private def isEmpty(values: Values): Boolean = {
+ !values.nullAllowed && !values.nanAllowed &&
values.ranges.exists(_.isEmpty)
+ }
+
+ // Whether no row can make the predicates true.
+ private def matchesNoRow(constraint: Constraint): Boolean = {
+ constraint.forall(_.values.exists(isEmpty))
+ }
+
+ private def and(left: Constraint, right: Constraint): Constraint = (left,
right) match {
+ case (Some(l), Some(r)) =>
+ Some((l.keySet ++ r.keySet).map { column =>
+ column -> ((l.get(column), r.get(column)) match {
+ case (Some(a), Some(b)) => intersect(column, a, b)
+ case (a, b) => a.orElse(b).get
+ })
+ }.toMap)
+ case _ => NoRow
+ }
+
+ // A column that only one side constrains can have any value in the rows of
the other side, so
+ // only the columns that both sides constrain are constrained.
+ private def or(left: Constraint, right: Constraint): Constraint = {
+ if (matchesNoRow(left)) {
+ right
+ } else if (matchesNoRow(right)) {
+ left
+ } else {
+ val (l, r) = (left.get, right.get)
+ Some(l.keySet.intersect(r.keySet).map(column => column -> union(column,
l(column), r(column)))
+ .filterNot { case (column, values) => isAll(column, values) }.toMap)
+ }
+ }
+
+ private def intersect(column: Column, a: Values, b: Values): Values = Values(
+ a.nullAllowed && b.nullAllowed,
+ a.nanAllowed && b.nanAllowed,
+ (a.ranges, b.ranges) match {
+ case (Some(x), Some(y)) => Some(intersectRanges(x, y, column.ordering))
+ case (x, y) => x.orElse(y)
+ })
+
+ private def union(column: Column, a: Values, b: Values): Values = Values(
+ a.nullAllowed || b.nullAllowed,
+ a.nanAllowed || b.nanAllowed,
+ (a.ranges, b.ranges) match {
+ case (Some(x), Some(y)) =>
+ unionRanges(x, y, column.ordering) match {
+ // All the values.
+ case Seq(ValueRange(None, _, None, _)) => None
+ case ranges => Some(ranges)
+ }
+ case _ => None
+ })
+
+ // A negative value when `a` starts before `b`.
+ private def compareLows(a: ValueRange, b: ValueRange, ordering:
Ordering[Any]): Int = {
+ (a.low, b.low) match {
+ case (None, None) => 0
+ case (None, _) => -1
+ case (_, None) => 1
+ case (Some(x), Some(y)) =>
+ val c = ordering.compare(x, y)
+ if (c != 0 || a.lowInclusive == b.lowInclusive) c else if
(a.lowInclusive) -1 else 1
+ }
+ }
+
+ // A negative value when `a` ends before `b`.
+ private def compareHighs(a: ValueRange, b: ValueRange, ordering:
Ordering[Any]): Int = {
+ (a.high, b.high) match {
+ case (None, None) => 0
+ case (None, _) => 1
+ case (_, None) => -1
+ case (Some(x), Some(y)) =>
+ val c = ordering.compare(x, y)
+ if (c != 0 || a.highInclusive == b.highInclusive) c else if
(a.highInclusive) 1 else -1
+ }
+ }
+
+ private def isEmptyRange(range: ValueRange, ordering: Ordering[Any]):
Boolean = {
+ (range.low, range.high) match {
+ case (Some(low), Some(high)) =>
+ val c = ordering.compare(low, high)
+ c > 0 || (c == 0 && !(range.lowInclusive && range.highInclusive))
+ case _ => false
+ }
+ }
+
+ // Intersects two lists of sorted ranges that neither overlap nor touch, by
sweeping them.
+ private def intersectRanges(
+ left: Seq[ValueRange],
+ right: Seq[ValueRange],
+ ordering: Ordering[Any]): Seq[ValueRange] = {
+ // The ranges are usually lists, such as those of IN, so the sweep indexes
copies of them.
+ val (a, b) = (left.toIndexedSeq, right.toIndexedSeq)
+ val result = ArrayBuffer.empty[ValueRange]
+ var i = 0
+ var j = 0
+ while (i < a.length && j < b.length) {
+ val (x, y) = (a(i), b(j))
+ val low = if (compareLows(x, y, ordering) >= 0) x else y
+ val high = if (compareHighs(x, y, ordering) <= 0) x else y
+ val range = ValueRange(low.low, low.lowInclusive, high.high,
high.highInclusive)
+ if (!isEmptyRange(range, ordering)) {
+ result += range
+ }
+ // Moves past the range that ends first.
+ if (compareHighs(x, y, ordering) <= 0) i += 1 else j += 1
+ }
+ result.toSeq
+ }
+
+ // Merges two lists of sorted ranges, and the ranges that overlap or touch,
such as [1, 2) and
+ // [2, 3].
+ private def unionRanges(
+ a: Seq[ValueRange],
+ b: Seq[ValueRange],
+ ordering: Ordering[Any]): Seq[ValueRange] = {
+ def touches(first: ValueRange, next: ValueRange): Boolean = (first.high,
next.low) match {
+ case (Some(high), Some(low)) =>
+ val c = ordering.compare(low, high)
+ c < 0 || (c == 0 && (first.highInclusive || next.lowInclusive))
+ case _ => true
+ }
+ (a ++ b).sortWith(compareLows(_, _, ordering) <
0).foldLeft(List.empty[ValueRange]) {
Review Comment:
**Nit (P3):** Each `OR` node calls `unionRanges` with the ranges of its two
sides, and this line concatenates them, sorts the result and folds it into a
new list. A disjunction built with the DataFrame API, such as `ids.map(v =>
col("id") === v).reduce(_ || _)`, is a left-deep tree. So the i-th `OR` node
copies, sorts and rebuilds the i ranges accumulated so far, and no optimizer
rule turns such a chain of equalities into `IN`. With 10,000 ids that is on the
order of 10^8 range operations on the driver while planning, about a second by
estimate, where one sort of the whole chain would do.
Collecting the disjuncts of nested `OR`s, and of `AND`s under `NOT`, before
a single union per column would make it O(k log k). It needs to keep the
current rules for disjuncts that match no row and for columns that only some
disjuncts constrain.
**Verification:**
- **Regression:** Deriving the domains of a left-deep OR chain of tens of
thousands of equalities on one column, as reduce(_ || _) builds it, takes time
proportional to k log k.
- **Behavior:** The domains of OR and negated AND combinations in the
existing rule cases and in the randomized check are unchanged.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnDomainsSuite.scala:
##########
@@ -0,0 +1,383 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.v2.columnar
+
+import java.sql.Date
+
+import scala.jdk.CollectionConverters._
+import scala.util.Random
+
+import org.scalatest.concurrent.TimeLimits
+import org.scalatest.time.SpanSugar._
+
+import org.apache.spark.sql.{Column, QueryTest, Row}
+import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}
+import org.apache.spark.sql.catalyst.expressions.{Literal, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.connector.expressions.{Expression => V2Expression,
FieldReference, LiteralValue}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy
+import org.apache.spark.sql.functions.{col, expr, not}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/** Tests how [[ColumnDomains]] derives the domains of columns from the
predicates of a query. */
+class ColumnDomainsSuite extends QueryTest with SharedSparkSession with
PredicateHelper
+ with TimeLimits {
+
+ private val schema = StructType.fromDDL("id BIGINT, i INT, d DOUBLE, f
FLOAT, s STRING, " +
Review Comment:
**Nit (P3):** The header promises domains for `tinyint`, `smallint` and
`timestamp_ntz` columns, and `isComparable` accepts `ByteType`, `ShortType` and
`TimestampNTZType`. But none of these schemas has such a column, and neither
does `ColumnarDataSourceSuite` or the native test library. The gap is wider for
`tinyint` and `smallint`: `t = 1` compares `CAST(t AS INT)` in the analyzed
plan that `predicates` reads, so the column only gets a domain after the
optimizer's `UnwrapCastInBinaryComparison`. Dropping one of these types from
`isComparable`, or a change that keeps the cast, would silently stop data
skipping on such columns, and every suite would still pass.
Cases for these types here would cover the derivation. An end-to-end case in
which `t = 1` on a `tinyint` column gets its domain after optimization would
cover the cast unwrapping. For that case, the JVM test planner would have to
report the column's domain, since today it only reads `bigint` bounds of the
first column.
**Verification:**
- **Regression:** Filters such as t = 1 on a tinyint column and s < 10 on a
smallint column of a pushdown scan push a domain for the column, and a
timestamp_ntz comparison derives a domain with timestamp_ntz bounds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
