peter-toth commented on code in PR #57727: URL: https://github.com/apache/spark/pull/57727#discussion_r3726899153
########## sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala: ########## @@ -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 Review Comment: Taking you up on "happy to revisit if you disagree" — I do, but only on the first of the two points, and with one sentence rather than the block I proposed. Dropping the second point (any shape, not the shapes you recognise). You're right: "fully evaluates runtime predicates" plus "will not be evaluated again after the scan" carries it, and the expanded version really is implementation guidance. The exactness point is the one I don't think the current wording reaches, because the neighbouring Javadoc points the other way. `filter()` says *"Implementations may use the expressions to prune initially planned InputPartitions"* — pruning partitions is the only action it sanctions. An adopter puts the two together and concludes that pruning partitions *is* what fully evaluating means. That holds only when the attribute's value is fixed within each partition. It doesn't hold for a data column a scan prunes by file or row-group min/max statistics, which is a normal thing for the class of source this targets, and the result is extra rows with no error anywhere. So the gap isn't that "fully evaluates" is ambiguous — it's that nothing connects it to the partition granularity `filter()` offers. One sentence closes it: ```scala * 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]]. Each attribute's value must therefore be fixed within every * [[org.apache.spark.sql.connector.read.InputPartition]] the scan returns, since pruning * partitions cannot fully evaluate a predicate on a column that varies within a partition. ``` Non-blocking either way — if you'd rather ship as is, I won't raise it again. -- 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]
