[SPARK-58523][SQL] Add Catalyst runtime filtering interface for DSv2 scans #57727
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * 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.internal.connector | ||
|
|
||
| import org.apache.spark.sql.catalyst.expressions.Expression | ||
| import org.apache.spark.sql.connector.expressions.NamedReference | ||
| import org.apache.spark.sql.connector.read.Scan | ||
|
|
||
| /** | ||
| * A mix-in interface for [[Scan]]. Data sources can implement this interface if they can | ||
| * filter initially planned [[org.apache.spark.sql.connector.read.InputPartition]]s using | ||
| * Catalyst [[Expression]]s Spark infers at runtime. | ||
| * Only one runtime filtering interface should be implemented by a data source. | ||
| * | ||
| * Spark considers a runtime predicate fully pushed when all attributes referenced by the | ||
| * predicate are returned by [[fullyPushedFilterAttributes]]. Fully pushed predicates are not | ||
| * evaluated again after the scan. | ||
| * | ||
| * Note that Spark will push runtime filters only if they are beneficial. | ||
| */ | ||
| trait SupportsRuntimeCatalystFiltering extends Scan { | ||
|
|
||
| /** | ||
| * Returns attributes this scan can be filtered by at runtime. | ||
| * | ||
| * Spark will call [[filter]] if it can derive a runtime filter for any of these attributes. | ||
| */ | ||
| def filterAttributes(): Array[NamedReference] | ||
|
|
||
| /** | ||
| * Returns attributes for which this scan fully evaluates runtime predicates. | ||
| * | ||
| * Any runtime predicate that references only attributes in this set is considered fully pushed | ||
| * and will not be evaluated again after the scan. These attributes must also be returned by | ||
| * [[filterAttributes]]. | ||
| */ | ||
| def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty | ||
|
Contributor
There was a problem hiding this comment. Finding 2. No objection to the attribute-level shape — v2 file sources already work this way.
Something along these lines: /**
* Returns attributes for which this scan fully evaluates runtime predicates.
*
* Any runtime predicate that references only attributes in this set is considered fully pushed
* and will not be evaluated again after the scan. These attributes must also be returned by
* [[filterAttributes]].
*
* Only declare an attribute here if this scan evaluates an arbitrary deterministic Catalyst
* predicate over it exactly, for every row it returns -- e.g. an identity partition column,
* whose value is known for every row of a surviving partition. Do not declare an attribute
* whose predicates only guide approximate pruning, such as file or row-group statistics.
* Spark may push any expression that references only these attributes, so do not assume a
* fixed set of operators: bind and evaluate the expression (see
* [[PartitionPredicateImpl]]) instead of pattern matching it.
*/While you're here, it would help to name the intended implementor in the PR description — it makes the contract judgeable and tells a reader why the interface is internal.
Contributor
There was a problem hiding this comment. One correction here in light of #57760: that PR adds |
||
|
|
||
| /** | ||
| * Filters this scan using runtime Catalyst expressions. | ||
| * | ||
| * The provided expressions must be interpreted as a set of predicates that are ANDed together. | ||
| * Implementations may use the expressions to prune initially planned | ||
| * [[org.apache.spark.sql.connector.read.InputPartition]]s. | ||
| * | ||
| * Note that Spark will call [[Scan.toBatch]] again after filtering the scan at runtime. | ||
| */ | ||
| def filter(expressions: Array[Expression]): Unit | ||
|
Contributor
There was a problem hiding this comment. Finding 4. |
||
|
|
||
| /** | ||
| * Returns the predicates that are pushed to the data source via [[filter]]. | ||
| * | ||
| * This method does not indicate whether a predicate is fully pushed. Spark infers that from | ||
| * [[fullyPushedFilterAttributes]]. The returned predicates may fully or partially help the data | ||
| * source prune initially planned | ||
| * [[org.apache.spark.sql.connector.read.InputPartition]]s. | ||
| * | ||
| * It's possible that there are no runtime predicates and [[filter]] is never called; | ||
| * an empty array should be returned for this case. | ||
| */ | ||
| def pushedPredicates(): Array[Expression] = Array.empty | ||
|
Contributor
There was a problem hiding this comment. Finding 5. Nothing in Spark reads this. Either drop it and let the fixture expose its own accessor ( |
||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| /* | ||
| * 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.connector.catalog | ||
|
|
||
| import java.util | ||
|
|
||
| import scala.collection.mutable.ArrayBuffer | ||
|
|
||
| import InMemoryCatalystRuntimeFilterTable._ | ||
|
|
||
| import org.apache.spark.sql.catalyst.expressions.Expression | ||
| import org.apache.spark.sql.connector.expressions.{NamedReference, Transform} | ||
| import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder} | ||
| import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering | ||
| import org.apache.spark.sql.types.StructType | ||
| import org.apache.spark.sql.util.CaseInsensitiveStringMap | ||
| import org.apache.spark.util.ArrayImplicits._ | ||
|
|
||
| /** | ||
| * In-memory table whose batch scan implements | ||
| * [[SupportsRuntimeCatalystFiltering]], so runtime filters arrive as Catalyst | ||
| * [[Expression]]s rather than connector predicates. | ||
| * | ||
| * Table properties: | ||
| * - `filter-attributes` (default: all partition cols): comma-separated list of | ||
| * column names to expose from `filterAttributes`. | ||
| * - `fully-pushed-filter-attributes` (default: none): comma-separated list of | ||
| * column names to expose from `fullyPushedFilterAttributes`. | ||
| */ | ||
| class InMemoryCatalystRuntimeFilterTable( | ||
| name: String, | ||
| columns: Array[Column], | ||
| partitioning: Array[Transform], | ||
| properties: util.Map[String, String]) | ||
| extends InMemoryTableWithV2Filter(name, columns, partitioning, properties) { | ||
|
|
||
| override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = { | ||
| new InMemoryCatalystRuntimeFilterScanBuilder(schema, options) | ||
| } | ||
|
|
||
| class InMemoryCatalystRuntimeFilterScanBuilder( | ||
| tableSchema: StructType, | ||
| options: CaseInsensitiveStringMap) | ||
| extends InMemoryScanBuilder(tableSchema, options) { | ||
| override def build: Scan = InMemoryCatalystRuntimeFilterBatchScan( | ||
| data.map(_.asInstanceOf[InputPartition]).toImmutableArraySeq, | ||
| schema, tableSchema, options) | ||
| } | ||
|
|
||
| /** | ||
| * Scan that receives runtime filters as Catalyst expressions. | ||
| * Records what was pushed; pruning is left to the | ||
| * [[org.apache.spark.sql.execution.FilterExec]] above the scan, so the recorded | ||
| * expressions are the only observable effect. | ||
| */ | ||
| case class InMemoryCatalystRuntimeFilterBatchScan( | ||
| var _data: Seq[InputPartition], | ||
| readSchema: StructType, | ||
| tableSchema: StructType, | ||
| options: CaseInsensitiveStringMap) | ||
| extends BatchScanBaseClass(_data, readSchema, tableSchema) | ||
| with SupportsRuntimeCatalystFiltering { | ||
|
|
||
| private val _catalystPredicates = ArrayBuffer.empty[Expression] | ||
|
|
||
| private val restrictedFilterAttrs: Option[Set[String]] = | ||
| Option(InMemoryCatalystRuntimeFilterTable.this.properties.get(FilterAttributesKey)) | ||
| .map(_.split(",").map(_.trim).toSet) | ||
|
|
||
| override def filterAttributes(): Array[NamedReference] = { | ||
| val scanFields = readSchema.fields.map(_.name).toSet | ||
| partitioning.flatMap(_.references()).filter { ref => | ||
| val name = ref.fieldNames.mkString(".") | ||
| scanFields.contains(name) && | ||
| restrictedFilterAttrs.forall(_.contains(name)) | ||
| } | ||
| } | ||
|
|
||
| override def fullyPushedFilterAttributes(): Array[NamedReference] = { | ||
| val fullyPushedFilterAttrs = Option( | ||
| InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey)) | ||
| .map(_.split(",").map(_.trim).toSet) | ||
| .getOrElse(Set.empty) | ||
| filterAttributes().filter { ref => | ||
| fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")) | ||
| } | ||
| } | ||
|
|
||
| override def filter(expressions: Array[Expression]): Unit = | ||
| _catalystPredicates ++= expressions | ||
|
Contributor
There was a problem hiding this comment. A scan that declares a filter fully pushed must ensure its returned partitions satisfy that predicate. This implementation only records the expression, while the fully-pushed test uses rows that all happen to match, so |
||
|
|
||
| override def pushedPredicates(): Array[Expression] = | ||
| _catalystPredicates.toArray | ||
| } | ||
| } | ||
|
|
||
| object InMemoryCatalystRuntimeFilterTable { | ||
| /** | ||
| * Table property: comma-separated column names to expose from | ||
| * filterAttributes. Default: all partition columns. | ||
| */ | ||
| private[catalog] val FilterAttributesKey = "filter-attributes" | ||
|
|
||
| /** | ||
| * Table property: comma-separated column names to expose from | ||
| * fullyPushedFilterAttributes. Default: none. | ||
| */ | ||
| private[catalog] val FullyPushedFilterAttributesKey = "fully-pushed-filter-attributes" | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| /* | ||
| * 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.connector.catalog | ||
|
|
||
| import java.util | ||
|
|
||
| import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException | ||
| import org.apache.spark.sql.connector.expressions.Transform | ||
|
|
||
| class InMemoryTableCatalystRuntimeFilterCatalog extends InMemoryTableCatalog { | ||
| import CatalogV2Implicits._ | ||
|
|
||
| override def createTable( | ||
| ident: Identifier, | ||
| columns: Array[Column], | ||
| partitions: Array[Transform], | ||
| properties: util.Map[String, String]): Table = { | ||
| if (tables.containsKey(ident)) { | ||
| throw new TableAlreadyExistsException(ident.asMultipartIdentifier) | ||
| } | ||
|
|
||
| InMemoryTableCatalog.maybeSimulateFailedTableCreation(properties) | ||
|
|
||
| val tableName = s"$name.${ident.quoted}" | ||
| val table = new InMemoryCatalystRuntimeFilterTable( | ||
| tableName, columns, partitions, properties) | ||
| tables.put(ident, table) | ||
| namespaces.putIfAbsent(ident.namespace.toList, Map()) | ||
| table | ||
| } | ||
|
|
||
| override def createTable(ident: Identifier, tableInfo: TableInfo): Table = { | ||
| createTable(ident, tableInfo.columns(), tableInfo.partitions(), tableInfo.properties) | ||
| } | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Finding 6.
resolveRefs→V2ExpressionUtils.resolveRefthrowscannotResolveAttributeErrorwhen a reference doesn't resolve against the plan's output, and casts the result toAttribute— so a nested reference, whichLogicalPlan.resolvehands back as anAlias(GetStructField(...)), throws aClassCastException.fullyPushedFilterAttributes()therefore has an unwritten requirement: top-level attributes only, and only ones that survived column pruning into the scan'sreadSchema. Break it and the query fails at planning time.filterAttributescarries the same requirement and is equally undocumented, but it's forced for every scan relation, so an adopter trips it on the first query. This one is only forced when a scalar-subquery runtime filter is present (scalarSubqueryFilters.filterdoesn't evaluate its closure on an empty Seq), which makes it a query-shape-dependent failure. Worth a line on the trait alongside finding 2; the new fixture quietly depends on it via thescanFields.contains(name)guard atInMemoryCatalystRuntimeFilterTable.scala:267.