spark-test commented on code in PR #59929: URL: https://github.com/apache/spark/pull/59929#discussion_r4418778137
########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnDomains.scala: ########## @@ -0,0 +1,534 @@ +/* + * 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. + */ +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.map { case (column, v) => + 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 (!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) { + single(column, Values(nullAllowed = false, !hasNaN, None)) Review Comment: **Blocking (P1):** When `values` is empty, the NULL-in-list check is `0 < 0` and `points` is empty, so this branch also handles `NOT (x IN ())`. It returns every non-null value with `nullAllowed = false`. Under the default semantics, though, `In.eval` with an empty list is `false` for every row, null included (`predicates.scala:513-517`; `legacyNullInEmptyBehavior` defaults to `!ansiEnabled`). So `NOT (x IN ())` is true for rows where `x` is null. The DataFrame API builds that predicate from `col("x").isin()` or `isInCollection(Seq.empty)`; SQL rejects an empty IN list. The default optimizer folds it away first. With `spark.sql.optimizer.excludedRules` set to `OptimizeIn` and `NullPropagation`, neither of which is non-excludable, it reaches `pushPredicates`. The library then gets `null_allowed = false` with `ranges = NULL`. A library that skips null-only files or row groups, as the description's `may_match` does, drops rows that match, and Spark's post-scan filter never sees them. Handling an empty list before the other branches fixes it: no value when not negated, and no constraint when negated. That is correct under both the current and the legacy semantics, without reading the legacy flag. `ParquetFilters` likewise builds an IN filter only when `values.nonEmpty`. **Verification:** - **Regression:** Deriving domains for NOT (x IN ()) built with the DataFrame API gives x no domain, and every row with a null x for which Spark evaluates the filter to true is in the derived domains. - **Behavior:** x IN () derives an empty domain, and the existing IN, NOT IN, NULL and NaN cases derive the same domains as before. ########## sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h: ########## @@ -350,6 +350,63 @@ typedef struct SparkDsWriteInfo { const char* query_id; } SparkDsWriteInfo; +/* + * The domain of a column, which push_domains gets: the values that the column can have in the + * rows that match the filters of a query. A value is in the domain when it is: + * - null, if null_allowed is true; + * - NaN, if nan_allowed is true, for a float or double column; + * - another value, if ranges is NULL, or if it is in one of the ranges. + * A reader can skip the data whose statistics show that it has no value in the domain of a column, + * such as a file or a row group whose values are all below the ranges. But it must never skip a + * row whose values are all in their domains, which Spark cannot detect. Spark evaluates the + * filters again after reading, so the rows that a reader returns do not need to match them. + * + * Values are compared with the order of Spark SQL: + * - numbers, dates and timestamps by their values. The bounds of the ranges of a float or double + * column are never NaN, and -0.0 is equal to 0.0: compare them with the comparisons of IEEE 754, + * as the operators of C, C++ and Rust do, not with a total order, such as f64::total_cmp in Rust, + * that puts -0.0 before 0.0. Data that has NaN values, or may have, can only be skipped if + * nan_allowed is false; + * - strings by their UTF-8 bytes, which order the characters outside the Basic Multilingual Plane + * after all the others, unlike comparisons of UTF-16 strings; + * - binary values as unsigned bytes, in lexicographic order: X'80' is greater than X'7F', and a + * value is greater than its prefixes; + * - booleans with false before true. + */ +typedef struct SparkDsDomain { + /* The size of this struct in the header that Spark was built with, as in SparkDsHost. */ + int64_t size; + /* + * The column, by the path_length indices of its fields: the index of the column in the schema to + * read, the one that create_reader got, and then the index of the field in each struct, for a + * field of a struct. {2, 0} is the first field of the third column. + */ + const int32_t* path; + int64_t path_length; + /* Whether null is in the domain. */ Review Comment: **Non-blocking (P2):** This defines `null_allowed` only as whether null is in the domain. For a struct-field path, neither the header nor the Filter Pushdown docs say what the field's value is in a row whose struct, or an enclosing struct, is null. Spark derives the domain as Spark SQL evaluates it, so the value is null there: `st.y IS NULL` gives `st.y: NULL`. Arrow allows a child array to hold valid values in slots where the struct is null (the header already notes a related validity trap for batches). A library that computes statistics from the child array of `y` can then find `null_count = 0` in a block whose rows all have a null `st`. For `st.y IS NULL` it sees no null and no range, skips the block, and loses every row in it. Only null-allowing domains are affected. Readers whose statistics already count those rows as null are fine, as with Parquet's definition levels. One sentence here and in the docs closes it: a field is null in every row where it or an enclosing struct is null, and statistics compared with `null_allowed` must count those rows as null. The alternative is for Spark to withhold null-allowing domains for struct fields. **Verification:** - **Inspection:** The SparkDsDomain documentation and the Filter Pushdown documentation define a struct field's value as null in rows whose enclosing struct is null, and tell readers to count those rows as null when they compare statistics with null_allowed. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnDomains.scala: ########## @@ -0,0 +1,534 @@ +/* + * 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. + */ +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.map { case (column, v) => + 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 (!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) { + 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( + a: Seq[ValueRange], + b: Seq[ValueRange], + ordering: Ordering[Any]): Seq[ValueRange] = { + val result = ArrayBuffer.empty[ValueRange] + var i = 0 + var j = 0 + while (i < a.length && j < b.length) { Review Comment: **Non-blocking (P2):** Every range sequence that reaches this loop is a `List`: `Seq(...)` in `compare` and `startsWith`, the point and gap lists of `in`, the result of `unionRanges`, and `result.toSeq` on the `ArrayBuffer` below. `List.length` and `List.apply` are linear, so each iteration re-walks both inputs, and the sweep is quadratic in the number of ranges. This runs on the driver during optimization, by default for every native batch scan, whenever two pushed predicates constrain the same column with ranges. Take `col("id").isin(ids: _*) && col("id") > 0` with 10^5 ids: OptimizeIn turns the list into an `InSet`, which is pushed as one IN carrying every literal. That makes about 10^10 list-node visits, tens of seconds of planning, and around 3 x 10^4 literals it already takes seconds. A NOT IN list, or two IN lists on one column, behaves the same. Converting both inputs once with `toIndexedSeq`, or walking them with buffered iterators, restores the linear sweep without changing any domain. **Verification:** - **Inspection:** The intersection sweep reads its inputs through constant-time indexing or iteration, so apart from the value comparisons each step does constant work. - **Behavior:** The domains derived for comparisons, IN, NOT IN and STARTS_WITH, and for their AND, OR and NOT combinations, are the same as before, including in the matching-rows property check. - **Regression:** Deriving the domains of an IN list with about 100,000 values combined with a range on the same column takes time proportional to the number of values, not its square. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnDomainsSuite.scala: ########## @@ -0,0 +1,305 @@ +/* + * 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.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.catalyst.CatalystTypeConverters +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.filter.Predicate +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy +import org.apache.spark.sql.functions.expr +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 { + + private val schema = StructType.fromDDL("id BIGINT, i INT, d DOUBLE, f FLOAT, s STRING, " + + "b BINARY, flag BOOLEAN, dt DATE, ts TIMESTAMP, dec DECIMAL(10, 2), " + + "st STRUCT<x: INT, y: STRING>, arr ARRAY<INT>, c STRING COLLATE UTF8_LCASE") + + private def char(codePoint: Int): String = new String(Character.toChars(codePoint)) + + /** + * The predicates that Data Source V2 pushes for the filter, as in the analyzed plan with the + * constants folded: the optimizer would rewrite some of them, such as NOT (x > 1). + */ + private def predicates(filter: String, schema: StructType): Seq[Predicate] = { + val df = spark.createDataFrame(Seq.empty[Row].asJava, schema) + val condition = df.where(filter).queryExecution.analyzed.collectFirst { + case Filter(condition, _) => condition + }.get + splitConjunctivePredicates(condition.transformUp { + case e if e.foldable && !e.isInstanceOf[Literal] => Literal.create(e.eval(), e.dataType) + }).flatMap(DataSourceV2Strategy.translateFilterV2) + } + + private def domains(filter: String): Seq[String] = { + ColumnDomains.derive(predicates(filter, schema), schema).domains.map(_.toString) + } + + test("comparisons, IN and null checks") { + assert(domains("id > 5") == Seq("id: (5L, +inf)")) + assert(domains("5 < id") == Seq("id: (5L, +inf)")) + assert(domains("id >= 5 AND id < 10") == Seq("id: [5L, 10L)")) + assert(domains("id <= 5") == Seq("id: (-inf, 5L]")) + assert(domains("id = 5") == Seq("id: 5L")) + assert(domains("id <> 5") == Seq("id: (-inf, 5L), (5L, +inf)")) + assert(domains("id <=> 5") == Seq("id: 5L")) + // x <=> v is true for null when false, and so is NOT (x <=> v). Review Comment: **Nit (P3):** For a null `x` and a non-null `v`, `x <=> v` is false, not null, so `NOT (x <=> v)` is true. That is why `NULL` is in the expected domain, and it is what `ColumnDomains.compare` says. Read literally, this line says `x <=> v` is true for null. ```suggestion // x <=> v is false, not null, when x is null, so NOT (x <=> v) is true for null. ``` **Verification:** - **Inspection:** The comment above the null-safe equality assertion says that x <=> v is false, not null, for a null x, consistent with the asserted domain and with the corresponding comment in the derivation code. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala: ########## @@ -394,6 +399,88 @@ class NativeDataSourceSuite extends QueryTest with SharedSparkSession { assert(!scan(filtered).getMetaData().contains("PushedLimit")) } + nativeTest("push down the domains of filtered columns") { + val file = new File(Utils.createTempDir(), "domains") + def readIn(session: SparkSession, options: (String, String)*): DataFrame = { + file.delete() + range(session, Seq("end" -> "45", "domains_path" -> file.getPath) ++ options: _*) + } + def read(options: (String, String)*): DataFrame = { + readIn(newSession(defaultPackage), options: _*) + } + // The domains that the library got: for each, the path of the column, whether null and NaN are + // in the domain, and the ranges. + def received(): Seq[String] = Files.readAllLines(file.toPath, UTF_8).asScala.toSeq + + // The library reads the blocks of 10 rows whose ids may match, [10, 20) and [20, 30), with all + // their rows, and Spark filters the rows after reading. + val df = read().filter("id > 13 AND id < 27") + checkRows(df, (14L until 27L).map(expectedRow)) + assert(received() == Seq("0 - - (13,27)")) Review Comment: **Nit (P3):** Every `received()` line in this test has `-` for null and at most one range. No native filter here uses `IS NULL`, `<=>`, `<>`, `NOT IN` or a same-column `OR`. So if `NativeLibrary.pushDomains` lost the `null_allowed` byte, or `NativeArrow.exportDomains` exported only the first range, everything here would still pass. Those are the encodings that tell a library which null or between-value rows it must keep, and the test library already prints `null` and every range. Separately, the collated-column assertion added at line 355 checks `PushedFilters == "[]"`. That holds even if Spark pushed a byte-wise domain for the `UTF8_LCASE` column, because the library marks only the `id` domain used unless `use_domains` is `all`. Separate cases such as `NOT (id <=> 5)` (null plus two ranges), `name IS NULL` and `id NOT IN (5, 20)` would cover the encodings. A collated case with `domains_path` or `use_domains = all` would cover the exclusion. Combining a null check with `NOT IN` on the same column in one filter would not work: `NOT IN` excludes null, so that domain is empty. **Verification:** - **Regression:** Native tests with filters such as NOT (id <=> 5) and id NOT IN (5, 20), each handing the library a domain that allows null or has several ranges, assert the null flag and every range that the library receives, so losing the null flag or any range after the first makes them fail. - **Regression:** The collated-column test records the domains the library receives and fails if one is passed for the collated column. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarDataSourceSuite.scala: ########## @@ -83,6 +83,34 @@ class ColumnarDataSourceSuite extends QueryTest with SharedSparkSession { assert(scan(limited).getMetaData()("PushedLimit") == "LIMIT 2") } + test("push down the domains of filtered columns") { + val df = spark.read.format(format).option("end", "100").load() + .filter("id >= 40 AND id < 50 AND name <> 'x'") + checkAnswer(df, (40L until 50L).map(id => Row(id, s"name$id"))) + // The data source only uses the domain of id, to read the ids in it. + val metadata = scan(df).getMetaData() + assert(metadata("PushedDomains") == "[id: [40L, 50L)]") + val pushed = metadata("PushedFilters") + assert(pushed.contains("id >= 40") && pushed.contains("id < 50"), pushed) + assert(!pushed.contains("name"), pushed) + assert(scan(df).toBatch.planInputPartitions().toSeq == + Seq(TestRangePartition(40, 45), TestRangePartition(45, 50))) + + // A filter that is not a domain is not pushed, and Spark evaluates all the filters after + // reading: here, the data source reads all the ids that are not 3. Review Comment: **Nit (P3):** `TestRangePlanner.pushDomains` keeps only the hull of the ranges. For `(-inf, 3L), (3L, +inf)` it plans `[0, 5)` and `[5, 10)`, so it reads id 3 too, and the unpushed `id % 2 = 0` also rejects that row. In the first case the planner reads exactly `[40, 50)`. No row that the planner reads is rejected only by a pushed predicate, so if `pushPredicates` stopped returning the pushed predicates for post-scan evaluation, both `checkAnswer` calls would still pass. Only the native tests would catch it, and they are skipped without a C++ compiler and Java 22 or later. A filter like `id <> 3 AND id < 8` would make this regression-sensitive. Its hull plans `[0, 4)` and `[4, 8)`, so id 3 is read and only the pushed `id <> 3` rejects it. Assert the partitions and the answer, and fix this comment. **Verification:** - **Regression:** With a filter whose domain hull includes a row that only a pushed predicate rejects, the test checks that the row is read and that the result does not contain it, so dropping pushed predicates from post-scan evaluation fails the test. -- 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]
