HyukjinKwon commented on code in PR #59943: URL: https://github.com/apache/spark/pull/59943#discussion_r4479845003
########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,475 @@ +/* + * 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.streaming.runtime + +import scala.collection.mutable +import scala.util.{Failure, Success, Try} +import scala.util.control.NonFatal + +import org.apache.spark.internal.{Logging, LogKeys} +import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, ResolvedInlineTable} +import org.apache.spark.sql.catalyst.catalog.HiveTableRelation +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeReference, AttributeSet, Cast, Expression, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime, InferFiltersFromConstraints, ReplaceExpressions, SpecialDatetimeValues} +import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST, CURRENT_LIKE, RUNTIME_REPLACEABLE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.classic.SparkSession +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.connector.read.streaming.SparkDataStream +import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, LogicalRelation} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{DataType, StringType, StructType, TimestampLTZNanosType, TimestampType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap +import org.apache.spark.unsafe.types.UTF8String + +/** + * Pushes the top-level columns and the filters that the optimized plan of a streaming query reads + * from each Data Source V2 streaming source down to the scan of the source, as + * `V2ScanRelationPushDown` does for the scans of batch queries, if the table of the source has the + * capability `STREAMING_COLUMN_PRUNING` or `STREAMING_FILTER_PUSHDOWN`. + * + * The stream of a source is created from its scan, and has to exist before any micro-batch is + * planned, so the scan is built when the query starts. The query is optimized once at that point, + * by the optimizer of its micro-batches and with their configuration, and the scan of each source + * gets what the planner takes from the optimized plan for the relation of the source (see + * `DataSourceV2Strategy`): the projections and the filters over each occurrence of the relation. + * + * The plan of each micro-batch is the one without these pushdowns, but for a filter of the rows + * of the relation with the filters that the scan builder got, which the query applies anyway, so + * that the scan cannot drop a row that the plan keeps (see `withPushedFilters`). The optimizer + * keeps deciding which columns every operator gets, and in which order, including the stateful + * ones, and Spark still evaluates all the filters over the relation. Only the scan node of the + * physical plan outputs the columns that the scan reads, and the planner checks that the + * micro-batch reads no other column of the relation. + * + * The other operations that the scans of batch queries can apply, such as aggregates and limits, + * apply to the data of all the micro-batches of a streaming query, which the scan of a micro-batch + * does not read, so they are not pushed down. + */ +private[sql] object StreamingV2ScanPushDown extends Logging { + + /** + * What the optimized plan of a streaming query reads from a streaming relation. + * + * @param requiredColumns the columns of the relation that the plan reads + * @param filters the filters that the scan of the relation may get, in the order in which the + * plan evaluates them: those that every occurrence of the relation applies first, + * see `pushableFilters` + */ + case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression]) + + /** + * The scan of a streaming source, and its stream. + * + * @param prunedOutput the columns of the relation that the scan reads, in the order of its read + * schema, unless these are all its columns + * @param pushedFilters the filters that the scan builder got, which the scan may apply in any + * form, so the plan of each micro-batch applies them over the relation + */ + case class StreamingScan( + scan: Scan, + stream: SparkDataStream, + prunedOutput: Option[Seq[AttributeReference]], + pushedFilters: Seq[Expression]) { + + /** Returns the scan relation of the source whose relation is `relation`. */ + def toRelation( + relation: StreamingDataSourceV2Relation): StreamingDataSourceV2ScanRelation = { + StreamingDataSourceV2ScanRelation(relation, scan, relation.output, stream, + prunedOutput = prunedOutput, pushedFilters = pushedFilters) + } + } + + /** + * Returns what the optimized `plan` reads from each of its streaming relations of Data Source + * V2: what the planner takes from the projections and the filters over each occurrence of the + * relation, or all its columns and no filter where nothing is projected or filtered over it. A + * relation that the optimized plan does not have has no entry, so that its scan reads all its + * columns and gets no filter, as no relation has if the optimizer fails, or if the optimized + * plan has a streaming relation that `plan` does not have. + * + * The plan is optimized as the micro-batches are: with the optimizer and the configuration of + * the stream, the values of the configurations that the offset log of a restarted query pins, and + * without the cost-based reordering of joins, which `StreamExecution` disables for its queries. + * The plan does not depend on when it is optimized, and the scan of a relation only gets what the + * query reads from that relation: + * - The expressions whose values depend on when they are evaluated or on the state of the + * session, such as the current time, the casts of special strings such as 'today' or 'now' + * to dates and timestamps, the casts of strings with only a time to timestamps, or the + * current database, which each micro-batch gets the values of, become placeholders that the + * optimizer does not take as deterministic, as the current time of a micro-batch, so neither + * its filters nor its columns depend on them, and no filter on them is pushed down. + * - The filters that the optimizer infers from the constraints of the plan, such as those of the + * filters over another relation joined with the relation, are not inferred, so that the scan + * only gets the filters that the query applies to the relation itself. + * - The relations of batch scans become placeholders, so that their scans are neither built + * nor planned: they do not change what the plan reads from its streaming relations. + * + * @param session the session of the stream + * @param offsetLogMetadata the metadata of the last entry of the offset log of the query, if any + */ + def reads( + session: SparkSession, + plan: LogicalPlan, + offsetLogMetadata: Option[OffsetSeqMetadataBase]): Map[StreamingRelationV2, Reads] = { + val conf = session.sessionState.conf.clone() + offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf, logChanges = false)) + conf.setConf(SQLConf.CBO_ENABLED, false) + conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, + (conf.optimizerExcludedRules.toSeq :+ InferFiltersFromConstraints.ruleName).mkString(",")) + session.withActive { + SQLConf.withExistingConf(conf) { + try { + val optimized = session.sessionState.optimizer.execute(forStartup(plan)) + // The planner of a micro-batch gets the values of the expressions that depend on when + // they are evaluated, which are deterministic. + val planned = optimized.transformUpWithSubqueries { + case node => node.transformExpressions { + case value: StartupValue => value.copy(planned = true) + } + } + val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2, Reads)] + // Visits the nodes as the planner plans them: the projections and the filters over a + // relation are planned with its scan, and the other nodes on their own. + def visit(node: LogicalPlan): Unit = node match { + case PhysicalOperation(project, filters, relation: StreamingRelationV2) => + occurrences += relation -> Reads( + AttributeSet((project ++ filters).flatMap(_.references)), + pushableFilters(filters)) + case _ => + node.children.foreach(visit) + } + visit(planned) + val relations = plan.collect { case relation: StreamingRelationV2 => relation }.toSet + val unknown = occurrences.map(_._1).filterNot(relations.contains) + if (unknown.nonEmpty) { + // The optimizer does not copy streaming relations: `DeduplicateRelations` keeps + // streaming plans as they are, so all the references to a common table expression, + // for example, have the same relation. A copy would share the scan of its relation in + // the micro-batches, so the scan would have to read what the copy reads too. + logWarning(log"The optimized streaming query reads the streaming relations " + + log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(", "))} that its " + + log"plan does not have, so the scans of its sources read all their columns with " + + log"no filter.") + Map.empty + } else { + combine(occurrences.toSeq) + } + } catch { + case NonFatal(e) => + logWarning(log"Could not optimize the streaming query to push down what it reads " + + log"to the scans of its sources, which read all their columns with no filter.", e) + Map.empty + } + } + } + } + + /** + * Returns what the occurrences of each relation read, combined: the scan of a relation reads the + * rows of all its occurrences, so it reads the columns that any of them reads, and can only + * apply the filters that all of them apply. Each occurrence evaluates the pushed filters before + * its own, so the pushed filters are those that all the occurrences evaluate first, in the same + * order: the longest common prefix of their filters, see `pushableFilters`. + */ + private def combine( + occurrences: Seq[(StreamingRelationV2, Reads)]): Map[StreamingRelationV2, Reads] = { + occurrences.groupBy(_._1).map { case (relation, reads) => + val filters = reads.map(_._2.filters).reduce { (left, right) => + left.zip(right).takeWhile { case (l, r) => l.semanticEquals(r) }.map(_._1) + } + relation -> Reads(AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters) + } + } + + /** + * Returns the filters that the scan of a relation may get of `filters`, the filters over an + * occurrence of the relation in the order in which the plan evaluates them: the deterministic + * ones without subqueries or placeholders, up to the first filter that is not, which is not + * pushed down, and neither are those after it. The plan of each micro-batch evaluates the pushed + * filters first, over the relation (see `withPushedFilters`), so a pushed filter does not see a + * row that a filter written before it drops: it could fail on such a row, as a filter with a + * cast or a division does under ANSI, or with `raise_error`. + */ + private def pushableFilters(filters: Seq[Expression]): Seq[Expression] = { + filters.takeWhile { filter => + filter.deterministic && !SubqueryExpression.hasSubquery(filter) && + !filter.exists(_.isInstanceOf[StartupValue]) + } + } + + /** + * Returns `plan` for the optimization when the query starts, see `reads`, with its subqueries. + * The expressions that are replaced keep their children, so that the plan still reads the same + * columns. Visible for tests. + */ + private[sql] def forStartup(plan: LogicalPlan): LogicalPlan = plan.transformUpWithSubqueries { + case relation: DataSourceV2Relation => BatchRelation(relation.output) + case relation: LogicalRelation if !relation.isStreaming => BatchRelation(relation.output) + case relation: HiveTableRelation => BatchRelation(relation.output) + // The rows of an inline table may have values of the current time, which the optimizer + // evaluates. + case table: ResolvedInlineTable => BatchRelation(table.output) + case node => + // The expressions that ReplaceExpressions replaces with others when the optimization starts + // are replaced first, as the optimizer does, so that the casts that they stand for, such as + // the one of to_date('today'), are replaced below. + val replaced = if (node.containsPattern(RUNTIME_REPLACEABLE)) { + node.mapExpressions(ReplaceExpressions.replace) + } else { + node + } + replaced.transformExpressionsUpWithPruning( + bits => bits.containsPattern(CURRENT_LIKE) || bits.containsPattern(CAST)) { + // The children of an expression are replaced before it, as are the subqueries, so an + // expression that still has the pattern CURRENT_LIKE has it itself: its value is the + // current time, or comes from the state of the session, as the current database does. + case e if e.containsPattern(CURRENT_LIKE) => StartupValue(e, e.children) + // So does a cast that ComputeCurrentTime or SpecialDatetimeValues rewrites: a cast of a + // time to a timestamp, which takes the current date, or a cast of a special string, such + // as 'today' or 'now', to a date or a timestamp. These rules only rewrite casts besides + // the expressions with the pattern CURRENT_LIKE. A cast of a string with only a time, + // such as '12:00:00', to a timestamp takes the current date too, which the optimizer + // folds in. + case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast || + SpecialDatetimeValues.applyForExpression(cast) != cast || castsTimeOnly(cast) => + StartupValue(cast, cast.children) + } + } + + /** + * Returns whether `cast` casts a foldable string with only a time to a timestamp with local time + * zone, of any precision, which takes the date of when it is evaluated. A cast to a timestamp + * without time zone or to a date does not take the current date: it rejects such a string. + */ + private def castsTimeOnly(cast: Cast): Boolean = { + val toTimestampLTZ = cast.dataType match { + case TimestampType | _: TimestampLTZNanosType => true + case _ => false + } + toTimestampLTZ && cast.child.dataType.isInstanceOf[StringType] && cast.child.foldable && Review Comment: Done in 89114e3: `forStartup` now replaces every cast of a string to `TIMESTAMP` or `TIMESTAMP_LTZ(p)`, whatever its child is, so the cast of `cutoff` becomes a placeholder before the optimizer propagates the string. With the filter pushdown moved to SPARK-60926 (#60129), what these placeholders decide here is the columns of a scan. The new test "read the columns of a condition on a cast whose string is a literal only once optimized" checks your `withColumn("cutoff", lit("12:00:00"))` case in a `CASE WHEN`: the scan reads the columns of both branches. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,475 @@ +/* + * 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.streaming.runtime + +import scala.collection.mutable +import scala.util.{Failure, Success, Try} +import scala.util.control.NonFatal + +import org.apache.spark.internal.{Logging, LogKeys} +import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, ResolvedInlineTable} +import org.apache.spark.sql.catalyst.catalog.HiveTableRelation +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeReference, AttributeSet, Cast, Expression, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime, InferFiltersFromConstraints, ReplaceExpressions, SpecialDatetimeValues} +import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST, CURRENT_LIKE, RUNTIME_REPLACEABLE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.classic.SparkSession +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.connector.read.streaming.SparkDataStream +import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, LogicalRelation} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{DataType, StringType, StructType, TimestampLTZNanosType, TimestampType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap +import org.apache.spark.unsafe.types.UTF8String + +/** + * Pushes the top-level columns and the filters that the optimized plan of a streaming query reads + * from each Data Source V2 streaming source down to the scan of the source, as + * `V2ScanRelationPushDown` does for the scans of batch queries, if the table of the source has the + * capability `STREAMING_COLUMN_PRUNING` or `STREAMING_FILTER_PUSHDOWN`. + * + * The stream of a source is created from its scan, and has to exist before any micro-batch is + * planned, so the scan is built when the query starts. The query is optimized once at that point, + * by the optimizer of its micro-batches and with their configuration, and the scan of each source + * gets what the planner takes from the optimized plan for the relation of the source (see + * `DataSourceV2Strategy`): the projections and the filters over each occurrence of the relation. + * + * The plan of each micro-batch is the one without these pushdowns, but for a filter of the rows + * of the relation with the filters that the scan builder got, which the query applies anyway, so + * that the scan cannot drop a row that the plan keeps (see `withPushedFilters`). The optimizer + * keeps deciding which columns every operator gets, and in which order, including the stateful + * ones, and Spark still evaluates all the filters over the relation. Only the scan node of the + * physical plan outputs the columns that the scan reads, and the planner checks that the + * micro-batch reads no other column of the relation. + * + * The other operations that the scans of batch queries can apply, such as aggregates and limits, + * apply to the data of all the micro-batches of a streaming query, which the scan of a micro-batch + * does not read, so they are not pushed down. + */ +private[sql] object StreamingV2ScanPushDown extends Logging { + + /** + * What the optimized plan of a streaming query reads from a streaming relation. + * + * @param requiredColumns the columns of the relation that the plan reads + * @param filters the filters that the scan of the relation may get, in the order in which the + * plan evaluates them: those that every occurrence of the relation applies first, + * see `pushableFilters` + */ + case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression]) + + /** + * The scan of a streaming source, and its stream. + * + * @param prunedOutput the columns of the relation that the scan reads, in the order of its read + * schema, unless these are all its columns + * @param pushedFilters the filters that the scan builder got, which the scan may apply in any + * form, so the plan of each micro-batch applies them over the relation + */ + case class StreamingScan( + scan: Scan, + stream: SparkDataStream, + prunedOutput: Option[Seq[AttributeReference]], + pushedFilters: Seq[Expression]) { + + /** Returns the scan relation of the source whose relation is `relation`. */ + def toRelation( + relation: StreamingDataSourceV2Relation): StreamingDataSourceV2ScanRelation = { + StreamingDataSourceV2ScanRelation(relation, scan, relation.output, stream, + prunedOutput = prunedOutput, pushedFilters = pushedFilters) + } + } + + /** + * Returns what the optimized `plan` reads from each of its streaming relations of Data Source + * V2: what the planner takes from the projections and the filters over each occurrence of the + * relation, or all its columns and no filter where nothing is projected or filtered over it. A + * relation that the optimized plan does not have has no entry, so that its scan reads all its + * columns and gets no filter, as no relation has if the optimizer fails, or if the optimized + * plan has a streaming relation that `plan` does not have. + * + * The plan is optimized as the micro-batches are: with the optimizer and the configuration of + * the stream, the values of the configurations that the offset log of a restarted query pins, and + * without the cost-based reordering of joins, which `StreamExecution` disables for its queries. + * The plan does not depend on when it is optimized, and the scan of a relation only gets what the + * query reads from that relation: + * - The expressions whose values depend on when they are evaluated or on the state of the + * session, such as the current time, the casts of special strings such as 'today' or 'now' + * to dates and timestamps, the casts of strings with only a time to timestamps, or the + * current database, which each micro-batch gets the values of, become placeholders that the + * optimizer does not take as deterministic, as the current time of a micro-batch, so neither + * its filters nor its columns depend on them, and no filter on them is pushed down. + * - The filters that the optimizer infers from the constraints of the plan, such as those of the + * filters over another relation joined with the relation, are not inferred, so that the scan + * only gets the filters that the query applies to the relation itself. + * - The relations of batch scans become placeholders, so that their scans are neither built + * nor planned: they do not change what the plan reads from its streaming relations. + * + * @param session the session of the stream + * @param offsetLogMetadata the metadata of the last entry of the offset log of the query, if any + */ + def reads( + session: SparkSession, + plan: LogicalPlan, + offsetLogMetadata: Option[OffsetSeqMetadataBase]): Map[StreamingRelationV2, Reads] = { + val conf = session.sessionState.conf.clone() + offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf, logChanges = false)) + conf.setConf(SQLConf.CBO_ENABLED, false) + conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, + (conf.optimizerExcludedRules.toSeq :+ InferFiltersFromConstraints.ruleName).mkString(",")) + session.withActive { + SQLConf.withExistingConf(conf) { + try { + val optimized = session.sessionState.optimizer.execute(forStartup(plan)) + // The planner of a micro-batch gets the values of the expressions that depend on when + // they are evaluated, which are deterministic. + val planned = optimized.transformUpWithSubqueries { + case node => node.transformExpressions { + case value: StartupValue => value.copy(planned = true) + } + } + val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2, Reads)] + // Visits the nodes as the planner plans them: the projections and the filters over a + // relation are planned with its scan, and the other nodes on their own. + def visit(node: LogicalPlan): Unit = node match { + case PhysicalOperation(project, filters, relation: StreamingRelationV2) => + occurrences += relation -> Reads( + AttributeSet((project ++ filters).flatMap(_.references)), + pushableFilters(filters)) + case _ => + node.children.foreach(visit) + } + visit(planned) + val relations = plan.collect { case relation: StreamingRelationV2 => relation }.toSet + val unknown = occurrences.map(_._1).filterNot(relations.contains) + if (unknown.nonEmpty) { + // The optimizer does not copy streaming relations: `DeduplicateRelations` keeps + // streaming plans as they are, so all the references to a common table expression, + // for example, have the same relation. A copy would share the scan of its relation in + // the micro-batches, so the scan would have to read what the copy reads too. + logWarning(log"The optimized streaming query reads the streaming relations " + + log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(", "))} that its " + + log"plan does not have, so the scans of its sources read all their columns with " + + log"no filter.") + Map.empty + } else { + combine(occurrences.toSeq) + } + } catch { + case NonFatal(e) => + logWarning(log"Could not optimize the streaming query to push down what it reads " + + log"to the scans of its sources, which read all their columns with no filter.", e) + Map.empty + } + } + } + } + + /** + * Returns what the occurrences of each relation read, combined: the scan of a relation reads the + * rows of all its occurrences, so it reads the columns that any of them reads, and can only + * apply the filters that all of them apply. Each occurrence evaluates the pushed filters before + * its own, so the pushed filters are those that all the occurrences evaluate first, in the same + * order: the longest common prefix of their filters, see `pushableFilters`. + */ + private def combine( + occurrences: Seq[(StreamingRelationV2, Reads)]): Map[StreamingRelationV2, Reads] = { + occurrences.groupBy(_._1).map { case (relation, reads) => + val filters = reads.map(_._2.filters).reduce { (left, right) => + left.zip(right).takeWhile { case (l, r) => l.semanticEquals(r) }.map(_._1) + } + relation -> Reads(AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters) + } + } + + /** + * Returns the filters that the scan of a relation may get of `filters`, the filters over an + * occurrence of the relation in the order in which the plan evaluates them: the deterministic + * ones without subqueries or placeholders, up to the first filter that is not, which is not + * pushed down, and neither are those after it. The plan of each micro-batch evaluates the pushed + * filters first, over the relation (see `withPushedFilters`), so a pushed filter does not see a + * row that a filter written before it drops: it could fail on such a row, as a filter with a + * cast or a division does under ANSI, or with `raise_error`. + */ + private def pushableFilters(filters: Seq[Expression]): Seq[Expression] = { + filters.takeWhile { filter => + filter.deterministic && !SubqueryExpression.hasSubquery(filter) && + !filter.exists(_.isInstanceOf[StartupValue]) + } + } + + /** + * Returns `plan` for the optimization when the query starts, see `reads`, with its subqueries. + * The expressions that are replaced keep their children, so that the plan still reads the same + * columns. Visible for tests. + */ + private[sql] def forStartup(plan: LogicalPlan): LogicalPlan = plan.transformUpWithSubqueries { + case relation: DataSourceV2Relation => BatchRelation(relation.output) + case relation: LogicalRelation if !relation.isStreaming => BatchRelation(relation.output) + case relation: HiveTableRelation => BatchRelation(relation.output) + // The rows of an inline table may have values of the current time, which the optimizer + // evaluates. + case table: ResolvedInlineTable => BatchRelation(table.output) + case node => + // The expressions that ReplaceExpressions replaces with others when the optimization starts + // are replaced first, as the optimizer does, so that the casts that they stand for, such as + // the one of to_date('today'), are replaced below. + val replaced = if (node.containsPattern(RUNTIME_REPLACEABLE)) { + node.mapExpressions(ReplaceExpressions.replace) + } else { + node + } + replaced.transformExpressionsUpWithPruning( + bits => bits.containsPattern(CURRENT_LIKE) || bits.containsPattern(CAST)) { + // The children of an expression are replaced before it, as are the subqueries, so an + // expression that still has the pattern CURRENT_LIKE has it itself: its value is the + // current time, or comes from the state of the session, as the current database does. + case e if e.containsPattern(CURRENT_LIKE) => StartupValue(e, e.children) + // So does a cast that ComputeCurrentTime or SpecialDatetimeValues rewrites: a cast of a + // time to a timestamp, which takes the current date, or a cast of a special string, such + // as 'today' or 'now', to a date or a timestamp. These rules only rewrite casts besides + // the expressions with the pattern CURRENT_LIKE. A cast of a string with only a time, + // such as '12:00:00', to a timestamp takes the current date too, which the optimizer + // folds in. + case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast || + SpecialDatetimeValues.applyForExpression(cast) != cast || castsTimeOnly(cast) => + StartupValue(cast, cast.children) + } + } + + /** + * Returns whether `cast` casts a foldable string with only a time to a timestamp with local time + * zone, of any precision, which takes the date of when it is evaluated. A cast to a timestamp + * without time zone or to a date does not take the current date: it rejects such a string. + */ + private def castsTimeOnly(cast: Cast): Boolean = { + val toTimestampLTZ = cast.dataType match { + case TimestampType | _: TimestampLTZNanosType => true + case _ => false + } + toTimestampLTZ && cast.child.dataType.isInstanceOf[StringType] && cast.child.foldable && + Try(cast.child.eval()).toOption.exists { + case string: UTF8String => DateTimeUtils.parseTimestampString(string)._3 + case _ => false + } + } + + /** + * Builds the scan of a streaming source, and creates its stream with `createStream`. If the + * table of the source has the capability `STREAMING_COLUMN_PRUNING` or + * `STREAMING_FILTER_PUSHDOWN`, its scan builder supports the pushdown, and `conf` enables it, + * the scan only reads the columns of `output` that the query reads, or gets the filters that + * the query applies to all the rows that it reads from the source (see `reads`). `reads` is only + * evaluated for such a scan. Otherwise, the scan reads all the columns of `output` with no + * filter, as the scans of streaming queries did before these pushdowns. + * + * A pruned scan must read the required columns and only columns of `output`, each once, as + * `SupportsPushDownRequiredColumns` requires. Spark asks a scan for its read schema before it + * creates its stream, as for a batch. If the scan builder, the scan or its stream fails with the + * pushdowns, or the read schema of a pruned scan cannot be read or does not follow + * `SupportsPushDownRequiredColumns`, a scan of all the columns with no filter is built instead, + * from a new scan builder, as without these pushdowns. + * + * @param output the output of the relation of the source, with the metadata columns that the + * query reads, which the analyzer added to it, so that connectors implementing + * `SupportsMetadataColumns` include them in `readSchema()` + * @param conf the configuration of the query + * @param reads what the query reads from the relation, or None to read all its columns + */ + def buildScan( + table: SupportsRead, + options: CaseInsensitiveStringMap, + output: Seq[AttributeReference], + conf: SQLConf, + reads: => Option[Reads])( + createStream: Scan => SparkDataStream): StreamingScan = { + val builder = table.newScanBuilder(options) + val capabilities = table.capabilities() + val prunesColumns = conf.streamingV2ColumnPruningEnabled && + capabilities.contains(TableCapability.STREAMING_COLUMN_PRUNING) && + builder.isInstanceOf[SupportsPushDownRequiredColumns] + val pushesFilters = conf.streamingV2FilterPushdownEnabled && + capabilities.contains(TableCapability.STREAMING_FILTER_PUSHDOWN) && + PushDownUtils.supportsFilters(builder) + val known = if (prunesColumns || pushesFilters) reads else None + val requiredOutput = known.filter(_ => prunesColumns) + .map(r => output.filter(r.requiredColumns.contains)) + .getOrElse(output) + // The filters that the scan builder gets in some form, which each micro-batch applies first: + // the others, such as a filter with a UDF that cannot be translated to a filter of the source, + // are evaluated after them, as for the scans of batch queries, rather than on all the rows of + // the source. + val filters = known.filter(_ => pushesFilters) + .map { r => + PushDownUtils.filtersOfScanBuilder(builder, Review Comment: The filter pushdown moved to SPARK-60926 (#60129), as zikangh suggested, so this PR no longer pushes filters down. This case is the first open item there: the pushed filters should end at the first filter that the builder does not get, with your `RLIKE` test. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,475 @@ +/* + * 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.streaming.runtime + +import scala.collection.mutable +import scala.util.{Failure, Success, Try} +import scala.util.control.NonFatal + +import org.apache.spark.internal.{Logging, LogKeys} +import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, ResolvedInlineTable} +import org.apache.spark.sql.catalyst.catalog.HiveTableRelation +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeReference, AttributeSet, Cast, Expression, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime, InferFiltersFromConstraints, ReplaceExpressions, SpecialDatetimeValues} +import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST, CURRENT_LIKE, RUNTIME_REPLACEABLE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.classic.SparkSession +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.connector.read.streaming.SparkDataStream +import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, LogicalRelation} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{DataType, StringType, StructType, TimestampLTZNanosType, TimestampType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap +import org.apache.spark.unsafe.types.UTF8String + +/** + * Pushes the top-level columns and the filters that the optimized plan of a streaming query reads + * from each Data Source V2 streaming source down to the scan of the source, as + * `V2ScanRelationPushDown` does for the scans of batch queries, if the table of the source has the + * capability `STREAMING_COLUMN_PRUNING` or `STREAMING_FILTER_PUSHDOWN`. + * + * The stream of a source is created from its scan, and has to exist before any micro-batch is + * planned, so the scan is built when the query starts. The query is optimized once at that point, + * by the optimizer of its micro-batches and with their configuration, and the scan of each source + * gets what the planner takes from the optimized plan for the relation of the source (see + * `DataSourceV2Strategy`): the projections and the filters over each occurrence of the relation. + * + * The plan of each micro-batch is the one without these pushdowns, but for a filter of the rows + * of the relation with the filters that the scan builder got, which the query applies anyway, so + * that the scan cannot drop a row that the plan keeps (see `withPushedFilters`). The optimizer + * keeps deciding which columns every operator gets, and in which order, including the stateful + * ones, and Spark still evaluates all the filters over the relation. Only the scan node of the + * physical plan outputs the columns that the scan reads, and the planner checks that the + * micro-batch reads no other column of the relation. + * + * The other operations that the scans of batch queries can apply, such as aggregates and limits, + * apply to the data of all the micro-batches of a streaming query, which the scan of a micro-batch + * does not read, so they are not pushed down. + */ +private[sql] object StreamingV2ScanPushDown extends Logging { + + /** + * What the optimized plan of a streaming query reads from a streaming relation. + * + * @param requiredColumns the columns of the relation that the plan reads + * @param filters the filters that the scan of the relation may get, in the order in which the + * plan evaluates them: those that every occurrence of the relation applies first, + * see `pushableFilters` + */ + case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression]) + + /** + * The scan of a streaming source, and its stream. + * + * @param prunedOutput the columns of the relation that the scan reads, in the order of its read + * schema, unless these are all its columns + * @param pushedFilters the filters that the scan builder got, which the scan may apply in any + * form, so the plan of each micro-batch applies them over the relation + */ + case class StreamingScan( + scan: Scan, + stream: SparkDataStream, + prunedOutput: Option[Seq[AttributeReference]], + pushedFilters: Seq[Expression]) { + + /** Returns the scan relation of the source whose relation is `relation`. */ + def toRelation( + relation: StreamingDataSourceV2Relation): StreamingDataSourceV2ScanRelation = { + StreamingDataSourceV2ScanRelation(relation, scan, relation.output, stream, + prunedOutput = prunedOutput, pushedFilters = pushedFilters) + } + } + + /** + * Returns what the optimized `plan` reads from each of its streaming relations of Data Source + * V2: what the planner takes from the projections and the filters over each occurrence of the + * relation, or all its columns and no filter where nothing is projected or filtered over it. A + * relation that the optimized plan does not have has no entry, so that its scan reads all its + * columns and gets no filter, as no relation has if the optimizer fails, or if the optimized + * plan has a streaming relation that `plan` does not have. + * + * The plan is optimized as the micro-batches are: with the optimizer and the configuration of + * the stream, the values of the configurations that the offset log of a restarted query pins, and + * without the cost-based reordering of joins, which `StreamExecution` disables for its queries. + * The plan does not depend on when it is optimized, and the scan of a relation only gets what the + * query reads from that relation: + * - The expressions whose values depend on when they are evaluated or on the state of the + * session, such as the current time, the casts of special strings such as 'today' or 'now' + * to dates and timestamps, the casts of strings with only a time to timestamps, or the + * current database, which each micro-batch gets the values of, become placeholders that the + * optimizer does not take as deterministic, as the current time of a micro-batch, so neither + * its filters nor its columns depend on them, and no filter on them is pushed down. + * - The filters that the optimizer infers from the constraints of the plan, such as those of the + * filters over another relation joined with the relation, are not inferred, so that the scan + * only gets the filters that the query applies to the relation itself. + * - The relations of batch scans become placeholders, so that their scans are neither built + * nor planned: they do not change what the plan reads from its streaming relations. + * + * @param session the session of the stream + * @param offsetLogMetadata the metadata of the last entry of the offset log of the query, if any + */ + def reads( + session: SparkSession, + plan: LogicalPlan, + offsetLogMetadata: Option[OffsetSeqMetadataBase]): Map[StreamingRelationV2, Reads] = { + val conf = session.sessionState.conf.clone() + offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf, logChanges = false)) + conf.setConf(SQLConf.CBO_ENABLED, false) + conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, + (conf.optimizerExcludedRules.toSeq :+ InferFiltersFromConstraints.ruleName).mkString(",")) + session.withActive { + SQLConf.withExistingConf(conf) { + try { + val optimized = session.sessionState.optimizer.execute(forStartup(plan)) + // The planner of a micro-batch gets the values of the expressions that depend on when + // they are evaluated, which are deterministic. + val planned = optimized.transformUpWithSubqueries { + case node => node.transformExpressions { + case value: StartupValue => value.copy(planned = true) + } + } + val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2, Reads)] + // Visits the nodes as the planner plans them: the projections and the filters over a + // relation are planned with its scan, and the other nodes on their own. + def visit(node: LogicalPlan): Unit = node match { + case PhysicalOperation(project, filters, relation: StreamingRelationV2) => + occurrences += relation -> Reads( + AttributeSet((project ++ filters).flatMap(_.references)), + pushableFilters(filters)) + case _ => + node.children.foreach(visit) + } + visit(planned) + val relations = plan.collect { case relation: StreamingRelationV2 => relation }.toSet + val unknown = occurrences.map(_._1).filterNot(relations.contains) + if (unknown.nonEmpty) { + // The optimizer does not copy streaming relations: `DeduplicateRelations` keeps + // streaming plans as they are, so all the references to a common table expression, + // for example, have the same relation. A copy would share the scan of its relation in + // the micro-batches, so the scan would have to read what the copy reads too. + logWarning(log"The optimized streaming query reads the streaming relations " + + log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(", "))} that its " + + log"plan does not have, so the scans of its sources read all their columns with " + + log"no filter.") + Map.empty + } else { + combine(occurrences.toSeq) + } + } catch { + case NonFatal(e) => + logWarning(log"Could not optimize the streaming query to push down what it reads " + + log"to the scans of its sources, which read all their columns with no filter.", e) + Map.empty + } + } + } + } + + /** + * Returns what the occurrences of each relation read, combined: the scan of a relation reads the + * rows of all its occurrences, so it reads the columns that any of them reads, and can only + * apply the filters that all of them apply. Each occurrence evaluates the pushed filters before + * its own, so the pushed filters are those that all the occurrences evaluate first, in the same + * order: the longest common prefix of their filters, see `pushableFilters`. + */ + private def combine( + occurrences: Seq[(StreamingRelationV2, Reads)]): Map[StreamingRelationV2, Reads] = { + occurrences.groupBy(_._1).map { case (relation, reads) => + val filters = reads.map(_._2.filters).reduce { (left, right) => + left.zip(right).takeWhile { case (l, r) => l.semanticEquals(r) }.map(_._1) + } + relation -> Reads(AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters) + } + } + + /** + * Returns the filters that the scan of a relation may get of `filters`, the filters over an + * occurrence of the relation in the order in which the plan evaluates them: the deterministic + * ones without subqueries or placeholders, up to the first filter that is not, which is not + * pushed down, and neither are those after it. The plan of each micro-batch evaluates the pushed + * filters first, over the relation (see `withPushedFilters`), so a pushed filter does not see a + * row that a filter written before it drops: it could fail on such a row, as a filter with a + * cast or a division does under ANSI, or with `raise_error`. + */ + private def pushableFilters(filters: Seq[Expression]): Seq[Expression] = { + filters.takeWhile { filter => + filter.deterministic && !SubqueryExpression.hasSubquery(filter) && + !filter.exists(_.isInstanceOf[StartupValue]) + } + } + + /** + * Returns `plan` for the optimization when the query starts, see `reads`, with its subqueries. + * The expressions that are replaced keep their children, so that the plan still reads the same + * columns. Visible for tests. + */ + private[sql] def forStartup(plan: LogicalPlan): LogicalPlan = plan.transformUpWithSubqueries { + case relation: DataSourceV2Relation => BatchRelation(relation.output) + case relation: LogicalRelation if !relation.isStreaming => BatchRelation(relation.output) + case relation: HiveTableRelation => BatchRelation(relation.output) + // The rows of an inline table may have values of the current time, which the optimizer + // evaluates. + case table: ResolvedInlineTable => BatchRelation(table.output) + case node => + // The expressions that ReplaceExpressions replaces with others when the optimization starts + // are replaced first, as the optimizer does, so that the casts that they stand for, such as + // the one of to_date('today'), are replaced below. + val replaced = if (node.containsPattern(RUNTIME_REPLACEABLE)) { + node.mapExpressions(ReplaceExpressions.replace) + } else { + node + } + replaced.transformExpressionsUpWithPruning( + bits => bits.containsPattern(CURRENT_LIKE) || bits.containsPattern(CAST)) { + // The children of an expression are replaced before it, as are the subqueries, so an + // expression that still has the pattern CURRENT_LIKE has it itself: its value is the + // current time, or comes from the state of the session, as the current database does. + case e if e.containsPattern(CURRENT_LIKE) => StartupValue(e, e.children) + // So does a cast that ComputeCurrentTime or SpecialDatetimeValues rewrites: a cast of a + // time to a timestamp, which takes the current date, or a cast of a special string, such + // as 'today' or 'now', to a date or a timestamp. These rules only rewrite casts besides + // the expressions with the pattern CURRENT_LIKE. A cast of a string with only a time, + // such as '12:00:00', to a timestamp takes the current date too, which the optimizer + // folds in. + case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast || + SpecialDatetimeValues.applyForExpression(cast) != cast || castsTimeOnly(cast) => + StartupValue(cast, cast.children) + } + } + + /** + * Returns whether `cast` casts a foldable string with only a time to a timestamp with local time + * zone, of any precision, which takes the date of when it is evaluated. A cast to a timestamp + * without time zone or to a date does not take the current date: it rejects such a string. + */ + private def castsTimeOnly(cast: Cast): Boolean = { + val toTimestampLTZ = cast.dataType match { Review Comment: Done in 89114e3: the check covers the casts whose types have a string and a timestamp with local time zone at the same place, at any depth: the elements of arrays, the keys and the values of maps, and the fields of structs. "read the columns of a condition on a string that depends on when it is cast" now has your `element_at(CAST(array('12:00:00') AS ARRAY<TIMESTAMP>), 1)` and `CAST(named_struct('t', '12:00:00') AS STRUCT<t: TIMESTAMP>).t` cases, and one with a map. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,475 @@ +/* + * 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.streaming.runtime + +import scala.collection.mutable +import scala.util.{Failure, Success, Try} +import scala.util.control.NonFatal + +import org.apache.spark.internal.{Logging, LogKeys} +import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, ResolvedInlineTable} +import org.apache.spark.sql.catalyst.catalog.HiveTableRelation +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeReference, AttributeSet, Cast, Expression, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime, InferFiltersFromConstraints, ReplaceExpressions, SpecialDatetimeValues} +import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST, CURRENT_LIKE, RUNTIME_REPLACEABLE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.classic.SparkSession +import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability} +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.connector.read.streaming.SparkDataStream +import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, LogicalRelation} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{DataType, StringType, StructType, TimestampLTZNanosType, TimestampType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap +import org.apache.spark.unsafe.types.UTF8String + +/** + * Pushes the top-level columns and the filters that the optimized plan of a streaming query reads + * from each Data Source V2 streaming source down to the scan of the source, as + * `V2ScanRelationPushDown` does for the scans of batch queries, if the table of the source has the + * capability `STREAMING_COLUMN_PRUNING` or `STREAMING_FILTER_PUSHDOWN`. + * + * The stream of a source is created from its scan, and has to exist before any micro-batch is + * planned, so the scan is built when the query starts. The query is optimized once at that point, + * by the optimizer of its micro-batches and with their configuration, and the scan of each source + * gets what the planner takes from the optimized plan for the relation of the source (see + * `DataSourceV2Strategy`): the projections and the filters over each occurrence of the relation. + * + * The plan of each micro-batch is the one without these pushdowns, but for a filter of the rows + * of the relation with the filters that the scan builder got, which the query applies anyway, so + * that the scan cannot drop a row that the plan keeps (see `withPushedFilters`). The optimizer + * keeps deciding which columns every operator gets, and in which order, including the stateful + * ones, and Spark still evaluates all the filters over the relation. Only the scan node of the + * physical plan outputs the columns that the scan reads, and the planner checks that the + * micro-batch reads no other column of the relation. + * + * The other operations that the scans of batch queries can apply, such as aggregates and limits, + * apply to the data of all the micro-batches of a streaming query, which the scan of a micro-batch + * does not read, so they are not pushed down. + */ +private[sql] object StreamingV2ScanPushDown extends Logging { + + /** + * What the optimized plan of a streaming query reads from a streaming relation. + * + * @param requiredColumns the columns of the relation that the plan reads + * @param filters the filters that the scan of the relation may get, in the order in which the + * plan evaluates them: those that every occurrence of the relation applies first, + * see `pushableFilters` + */ + case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression]) + + /** + * The scan of a streaming source, and its stream. + * + * @param prunedOutput the columns of the relation that the scan reads, in the order of its read + * schema, unless these are all its columns + * @param pushedFilters the filters that the scan builder got, which the scan may apply in any + * form, so the plan of each micro-batch applies them over the relation + */ + case class StreamingScan( + scan: Scan, + stream: SparkDataStream, + prunedOutput: Option[Seq[AttributeReference]], + pushedFilters: Seq[Expression]) { + + /** Returns the scan relation of the source whose relation is `relation`. */ + def toRelation( + relation: StreamingDataSourceV2Relation): StreamingDataSourceV2ScanRelation = { + StreamingDataSourceV2ScanRelation(relation, scan, relation.output, stream, + prunedOutput = prunedOutput, pushedFilters = pushedFilters) + } + } + + /** + * Returns what the optimized `plan` reads from each of its streaming relations of Data Source + * V2: what the planner takes from the projections and the filters over each occurrence of the + * relation, or all its columns and no filter where nothing is projected or filtered over it. A + * relation that the optimized plan does not have has no entry, so that its scan reads all its + * columns and gets no filter, as no relation has if the optimizer fails, or if the optimized + * plan has a streaming relation that `plan` does not have. + * + * The plan is optimized as the micro-batches are: with the optimizer and the configuration of + * the stream, the values of the configurations that the offset log of a restarted query pins, and + * without the cost-based reordering of joins, which `StreamExecution` disables for its queries. + * The plan does not depend on when it is optimized, and the scan of a relation only gets what the + * query reads from that relation: + * - The expressions whose values depend on when they are evaluated or on the state of the + * session, such as the current time, the casts of special strings such as 'today' or 'now' + * to dates and timestamps, the casts of strings with only a time to timestamps, or the + * current database, which each micro-batch gets the values of, become placeholders that the + * optimizer does not take as deterministic, as the current time of a micro-batch, so neither + * its filters nor its columns depend on them, and no filter on them is pushed down. + * - The filters that the optimizer infers from the constraints of the plan, such as those of the + * filters over another relation joined with the relation, are not inferred, so that the scan + * only gets the filters that the query applies to the relation itself. + * - The relations of batch scans become placeholders, so that their scans are neither built + * nor planned: they do not change what the plan reads from its streaming relations. + * + * @param session the session of the stream + * @param offsetLogMetadata the metadata of the last entry of the offset log of the query, if any + */ + def reads( + session: SparkSession, + plan: LogicalPlan, + offsetLogMetadata: Option[OffsetSeqMetadataBase]): Map[StreamingRelationV2, Reads] = { + val conf = session.sessionState.conf.clone() + offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf, logChanges = false)) + conf.setConf(SQLConf.CBO_ENABLED, false) + conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, + (conf.optimizerExcludedRules.toSeq :+ InferFiltersFromConstraints.ruleName).mkString(",")) + session.withActive { + SQLConf.withExistingConf(conf) { + try { + val optimized = session.sessionState.optimizer.execute(forStartup(plan)) + // The planner of a micro-batch gets the values of the expressions that depend on when + // they are evaluated, which are deterministic. + val planned = optimized.transformUpWithSubqueries { + case node => node.transformExpressions { + case value: StartupValue => value.copy(planned = true) + } + } + val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2, Reads)] + // Visits the nodes as the planner plans them: the projections and the filters over a + // relation are planned with its scan, and the other nodes on their own. + def visit(node: LogicalPlan): Unit = node match { + case PhysicalOperation(project, filters, relation: StreamingRelationV2) => + occurrences += relation -> Reads( + AttributeSet((project ++ filters).flatMap(_.references)), + pushableFilters(filters)) + case _ => + node.children.foreach(visit) + } + visit(planned) + val relations = plan.collect { case relation: StreamingRelationV2 => relation }.toSet + val unknown = occurrences.map(_._1).filterNot(relations.contains) + if (unknown.nonEmpty) { + // The optimizer does not copy streaming relations: `DeduplicateRelations` keeps + // streaming plans as they are, so all the references to a common table expression, + // for example, have the same relation. A copy would share the scan of its relation in + // the micro-batches, so the scan would have to read what the copy reads too. + logWarning(log"The optimized streaming query reads the streaming relations " + + log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(", "))} that its " + + log"plan does not have, so the scans of its sources read all their columns with " + + log"no filter.") + Map.empty + } else { + combine(occurrences.toSeq) + } + } catch { + case NonFatal(e) => + logWarning(log"Could not optimize the streaming query to push down what it reads " + + log"to the scans of its sources, which read all their columns with no filter.", e) + Map.empty + } + } + } + } + + /** + * Returns what the occurrences of each relation read, combined: the scan of a relation reads the + * rows of all its occurrences, so it reads the columns that any of them reads, and can only + * apply the filters that all of them apply. Each occurrence evaluates the pushed filters before + * its own, so the pushed filters are those that all the occurrences evaluate first, in the same + * order: the longest common prefix of their filters, see `pushableFilters`. + */ + private def combine( + occurrences: Seq[(StreamingRelationV2, Reads)]): Map[StreamingRelationV2, Reads] = { + occurrences.groupBy(_._1).map { case (relation, reads) => + val filters = reads.map(_._2.filters).reduce { (left, right) => + left.zip(right).takeWhile { case (l, r) => l.semanticEquals(r) }.map(_._1) + } + relation -> Reads(AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters) + } + } + + /** + * Returns the filters that the scan of a relation may get of `filters`, the filters over an + * occurrence of the relation in the order in which the plan evaluates them: the deterministic + * ones without subqueries or placeholders, up to the first filter that is not, which is not + * pushed down, and neither are those after it. The plan of each micro-batch evaluates the pushed + * filters first, over the relation (see `withPushedFilters`), so a pushed filter does not see a + * row that a filter written before it drops: it could fail on such a row, as a filter with a + * cast or a division does under ANSI, or with `raise_error`. + */ + private def pushableFilters(filters: Seq[Expression]): Seq[Expression] = { + filters.takeWhile { filter => + filter.deterministic && !SubqueryExpression.hasSubquery(filter) && + !filter.exists(_.isInstanceOf[StartupValue]) + } + } + + /** + * Returns `plan` for the optimization when the query starts, see `reads`, with its subqueries. + * The expressions that are replaced keep their children, so that the plan still reads the same + * columns. Visible for tests. + */ + private[sql] def forStartup(plan: LogicalPlan): LogicalPlan = plan.transformUpWithSubqueries { + case relation: DataSourceV2Relation => BatchRelation(relation.output) + case relation: LogicalRelation if !relation.isStreaming => BatchRelation(relation.output) + case relation: HiveTableRelation => BatchRelation(relation.output) + // The rows of an inline table may have values of the current time, which the optimizer + // evaluates. + case table: ResolvedInlineTable => BatchRelation(table.output) + case node => + // The expressions that ReplaceExpressions replaces with others when the optimization starts + // are replaced first, as the optimizer does, so that the casts that they stand for, such as + // the one of to_date('today'), are replaced below. + val replaced = if (node.containsPattern(RUNTIME_REPLACEABLE)) { + node.mapExpressions(ReplaceExpressions.replace) + } else { + node + } + replaced.transformExpressionsUpWithPruning( + bits => bits.containsPattern(CURRENT_LIKE) || bits.containsPattern(CAST)) { + // The children of an expression are replaced before it, as are the subqueries, so an + // expression that still has the pattern CURRENT_LIKE has it itself: its value is the + // current time, or comes from the state of the session, as the current database does. + case e if e.containsPattern(CURRENT_LIKE) => StartupValue(e, e.children) + // So does a cast that ComputeCurrentTime or SpecialDatetimeValues rewrites: a cast of a + // time to a timestamp, which takes the current date, or a cast of a special string, such + // as 'today' or 'now', to a date or a timestamp. These rules only rewrite casts besides + // the expressions with the pattern CURRENT_LIKE. A cast of a string with only a time, + // such as '12:00:00', to a timestamp takes the current date too, which the optimizer + // folds in. + case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast || Review Comment: Done in 89114e3, as you suggested: every cast of a string to `TIMESTAMP` or `TIMESTAMP_LTZ(p)`, at any depth of its types and whatever its child is, becomes a placeholder, which keeps the columns that the cast reads. I kept `ComputeCurrentTime` as it is, since giving these casts the current date there would change every query with such a cast, as you said. The cost that you mention, that a filter on such a cast is not pushed down, now belongs to SPARK-60926 (#60129), where I listed it. ########## sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingV2ScanPushDownSuite.scala: ########## @@ -0,0 +1,1313 @@ +/* + * 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.streaming + +import java.util +import java.util.concurrent.ConcurrentLinkedQueue +import java.util.concurrent.atomic.{AtomicInteger, AtomicLong} + +import scala.jdk.CollectionConverters._ + +import org.apache.logging.log4j.Level +import org.scalatest.BeforeAndAfter + +import org.apache.spark.SparkException +import org.apache.spark.sql.{DataFrame, Row} +import org.apache.spark.sql.catalyst.{InternalRow, ProjectingInternalRow} +import org.apache.spark.sql.catalyst.expressions.{Alias, Cast} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, Project} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.classic.Dataset +import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryBaseTable, InMemoryTableCatalog, SupportsRead, Table, TableCapability} +import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.connector.read.streaming.{ContinuousPartitionReader, ContinuousPartitionReaderFactory, ContinuousStream, MicroBatchStream, Offset, PartitionOffset} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, MicroBatchScanExec, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.runtime.{MemoryStream, StreamExecution, StreamingQueryWrapper, StreamingV2ScanPushDown} +import org.apache.spark.sql.execution.streaming.sources.ContinuousMemoryStream +import org.apache.spark.sql.functions._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.sources.{Filter => SourceFilter} +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +class StreamingV2ScanPushDownSuite extends StreamTest with BeforeAndAfter { + import testImplicits._ + + private val tbl = "testcat.t" + + before { + spark.conf.set("spark.sql.catalog.testcat", classOf[InMemoryTableCatalog].getName) + } + + after { + spark.sessionState.catalogManager.reset() + spark.conf.unset("spark.sql.catalog.testcat") + } + + private def withSourceTable(properties: String = "")(f: => Unit): Unit = { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, data string, extra int) $properties") + sql(s"INSERT INTO $tbl VALUES (1, 'a', 10), (2, 'b', 20)") + f + } + } + + private def insert(values: String, table: String = tbl): StreamAction = Execute { _ => + sql(s"INSERT INTO $table VALUES $values") + } + + private def pruningEnabled(enabled: Boolean): Map[String, String] = + Map(SQLConf.STREAMING_V2_COLUMN_PRUNING_ENABLED.key -> enabled.toString) + + /** + * Runs `f`, which restarts streams with other values of the confs of these pushdowns, and then + * restores their values: `testStream` restores the value that a conf had before the last + * `StartStream` that set it, rather than before the first. + */ + private def withPushDownConfsRestored(f: => Unit): Unit = withSQLConf( + SQLConf.STREAMING_V2_COLUMN_PRUNING_ENABLED.key -> "true", + SQLConf.STREAMING_V2_FILTER_PUSHDOWN_ENABLED.key -> "true")(f) + + private def filterPushdownEnabled(enabled: Boolean): Map[String, String] = + Map(SQLConf.STREAMING_V2_FILTER_PUSHDOWN_ENABLED.key -> enabled.toString) + + /** Returns the filters pushed down to the scan of each streaming source of the query. */ + private def pushedFilters(query: StreamExecution): Seq[Seq[String]] = { + query.logicalPlan.collect { + case relation: StreamingDataSourceV2ScanRelation => + relation.scan match { + case scan: InMemoryBaseTable#InMemoryBatchScan => + scan.pushedFilters.map(_.toString).toSeq.sorted + case scan => + fail(s"The scan $scan of ${relation.name} is not a scan of an in-memory table.") + } + } + } + + /** Checks that the filters pushed down to the scan of each streaming source are `filters`. */ + private def assertPushedFilters(filters: String*): StreamAction = AssertOnQuery { query => + val pushed = pushedFilters(query) + assert(pushed.nonEmpty) + pushed.foreach(p => assert(p === filters.sorted)) + true + } + + /** Checks the read schemas of the scans of the streaming sources of the query. */ + private def assertReadSchemas(columns: Seq[String]*): StreamAction = AssertOnQuery { query => + val readSchemas = query.logicalPlan.collect { + case relation: StreamingDataSourceV2ScanRelation => + relation.scan.readSchema().fieldNames.toSeq + } + assert(readSchemas.sortBy(_.mkString(",")) === columns.sortBy(_.mkString(","))) + true + } + + /** Checks that the scan of each streaming source of the query reads `columns`. */ + private def assertReadSchema(columns: String*): StreamAction = AssertOnQuery { query => + val readSchemas = query.logicalPlan.collect { + case relation: StreamingDataSourceV2ScanRelation => + relation.scan.readSchema().fieldNames.toSeq + } + assert(readSchemas.nonEmpty) + readSchemas.foreach(readSchema => assert(readSchema === columns)) + true + } + + /** Checks the output of the scan nodes of the last micro-batch. */ + private def assertScanOutputs(columns: Seq[String]*): StreamAction = AssertOnQuery { query => + val outputs = query.lastExecution.executedPlan.collect { + case scan: MicroBatchScanExec => scan.output.map(_.name) + } + assert(outputs.sortBy(_.mkString(",")) === columns.sortBy(_.mkString(","))) + true + } + + test("read only the columns that a streaming query needs, also after a restart") { + withSourceTable() { + val df = spark.readStream.table(tbl).where("extra > 10").select("data") + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("data", "extra"), + assertScanOutputs(Seq("data", "extra")), + CheckAnswer(Row("b")), + StopStream, + insert("(3, 'c', 30), (4, 'd', 5)"), + StartStream(), + ProcessAllAvailable(), + assertReadSchema("data", "extra"), + CheckAnswer(Row("b"), Row("c"))) + } + } + + // Each case is a query over the source, and the columns that the optimizer reads for it, which + // fewer than those that the operators of its analyzed plan reference. + Seq[(String, DataFrame => DataFrame, Seq[String])]( + ("a column that a later projection drops", + _.withColumn("x", lit(1)).select("data"), Seq("data")), + ("chained projections", _.select("id", "data").select("data"), Seq("data")), + ("expressions and functions that a later projection drops", + _.select(upper($"data").as("u"), ($"id" * 2).as("i")).select("u"), Seq("data")), + ("several aliases of a column", + _.select($"id".as("x"), $"data".as("y")).select($"x".as("z")).where("z > 1"), Seq("id")), + ("a field of a struct of columns", + _.select(struct($"id", $"data").as("s")).select("s.data"), Seq("data")), + ("the fields of a struct of columns", + _.select(struct($"id", $"extra").as("s")).select("s.*"), Seq("id", "extra")), + ("a filter between projections", + _.select("id", "data").where("id > 1").select("data"), Seq("id", "data")), + ("a projection of all the columns", _.select("*").select("data"), Seq("data")) + ).foreach { case (name, query, columns) => + test(s"read only the columns that the optimized query reads: $name") { + withSourceTable() { + val expected = query(spark.read.table(tbl)).collect().toSeq + testStream(query(spark.readStream.table(tbl)))( + ProcessAllAvailable(), + assertReadSchema(columns: _*), + assertScanOutputs(columns), + CheckAnswer(expected: _*)) + } + } + } + + Seq[(String, DataFrame => DataFrame)]( + ("from_json", _.select(from_json($"json", new StructType().add("a", "int")).as("p")) + .select("p.a")), + ("get_json_object", _.select(get_json_object($"json", "$.a").as("a"))), + ("parse_json", _.select(parse_json($"json").as("v")) + .select(variant_get($"v", "$.a", "int").as("a"))) + ).foreach { case (name, query) => + test(s"read only the parsed column of $name") { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, json string, extra int)") + sql(s"""INSERT INTO $tbl VALUES (1, '{"a": 1}', 10), (2, '{"a": 2}', 20)""") + val expected = query(spark.read.table(tbl)).collect().toSeq + testStream(query(spark.readStream.table(tbl)))( + ProcessAllAvailable(), + assertReadSchema("json"), + CheckAnswer(expected: _*)) + } + } + } + + test("read only the columns of a streaming aggregation, also after a restart") { + withSourceTable() { + val df = spark.readStream.table(tbl).groupBy("data").count() + testStream(df, OutputMode.Complete())( + ProcessAllAvailable(), + assertReadSchema("data"), + CheckAnswer(Row("a", 1L), Row("b", 1L)), + StopStream, + insert("(3, 'a', 30)"), + StartStream(), + ProcessAllAvailable(), + assertReadSchema("data"), + CheckAnswer(Row("a", 2L), Row("b", 1L))) + } + } + + test("read only the event-time column of a time window") { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, data string, ts timestamp)") + sql(s"INSERT INTO $tbl VALUES " + + "(1, 'a', timestamp_seconds(0)), (2, 'b', timestamp_seconds(15))") + val df = spark.readStream.table(tbl).groupBy(window($"ts", "10 seconds")).count() + .select($"window.start".cast("long"), $"count") + testStream(df, OutputMode.Complete())( + ProcessAllAvailable(), + assertReadSchema("ts"), + CheckAnswer(Row(0L, 1L), Row(10L, 1L))) + } + } + + test("read no column of the source when the query needs none, also after a restart") { + withSourceTable() { + testStream(spark.readStream.table(tbl).groupBy().count(), OutputMode.Complete())( + ProcessAllAvailable(), + assertReadSchema(), + assertScanOutputs(Nil), + CheckAnswer(Row(2L)), + StopStream, + insert("(3, 'c', 30)"), + StartStream(), + ProcessAllAvailable(), + assertReadSchema(), + CheckAnswer(Row(3L))) + testStream(spark.readStream.table(tbl).select(lit(1).as("one")))( + ProcessAllAvailable(), + assertReadSchema(), + CheckAnswer(Row(1), Row(1), Row(1))) + } + } + + test("read only the columns of a streaming deduplication, also after a restart") { + withSourceTable() { + val df = spark.readStream.table(tbl).dropDuplicates("data").select("id") + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("id", "data"), + CheckAnswer(Row(1L), Row(2L)), + StopStream, + insert("(3, 'a', 30), (4, 'c', 40)"), + StartStream(), + ProcessAllAvailable(), + assertReadSchema("id", "data"), + CheckAnswer(Row(1L), Row(2L), Row(4L))) + } + } + + test("read the metadata columns that a streaming query needs") { + withSourceTable() { + val df = spark.readStream.table(tbl).select("data", "index") + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("data", "index"), + CheckAnswerRowsByFunc(rows => { + assert(rows.map(_.getString(0)).sorted === Seq("a", "b")) + assert(rows.forall(!_.isNullAt(1))) + }, lastOnly = false)) + } + } + + test("read only the columns of each child of a union") { + withSourceTable() { + // Each read of the table has its own relation and scan, and the optimizer prunes the + // columns of each child of the union by position. + val twoReads = spark.readStream.table(tbl).union(spark.readStream.table(tbl)).select("data") + testStream(twoReads)( + ProcessAllAvailable(), + assertReadSchemas(Seq("data"), Seq("data")), + CheckAnswer(Row("a"), Row("b"), Row("a"), Row("b"))) + + val differentColumns = spark.readStream.table(tbl).select("id") + .union(spark.readStream.table(tbl).select($"extra".cast("long"))) + testStream(differentColumns)( + ProcessAllAvailable(), + assertReadSchemas(Seq("id"), Seq("extra")), + CheckAnswer(Row(1L), Row(2L), Row(10L), Row(20L))) + + // The children of a self-union read the same relation, whose scan reads the columns that + // both of them read. + val source = spark.readStream.table(tbl) + testStream(source.select("id", "data").union(source.select("id", "data")).select("data"))( + ProcessAllAvailable(), + AssertOnQuery { query => + val relations = query.logicalPlan.collect { + case relation: StreamingDataSourceV2ScanRelation => relation + } + assert(relations.length == 2 && relations.map(_.stream).distinct.length == 1) + assert(relations.head.scan.readSchema().fieldNames.toSeq === Seq("data")) + true + }, + CheckAnswer(Row("a"), Row("b"), Row("a"), Row("b"))) + } + } + + test("read the columns that a scan returns when it reads more than the required ones") { + withSourceTable( + s"TBLPROPERTIES ('${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')") { + val df = spark.readStream.table(tbl).select("data", "index") + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("id", "data", "extra", "index"), + CheckAnswerRowsByFunc(rows => { + assert(rows.map(_.getString(0)).sorted === Seq("a", "b")) + assert(rows.forall(!_.isNullAt(1))) + }, lastOnly = false)) + } + } + + test("read all the columns when a scan does not read the required ones") { + withSourceTable( + s"TBLPROPERTIES ('${InMemoryBaseTable.SIMULATE_INCOMPLETE_COLUMN_PRUNING}' = 'true')") { + val table = spark.sessionState.catalogManager.catalog("testcat") + .asInstanceOf[InMemoryTableCatalog].liveTable(Identifier.of(Array.empty, "t")) + .asInstanceOf[InMemoryBaseTable] + val logAppender = new LogAppender("the fallback of streaming column pruning") + withLogAppender(logAppender, + loggerNames = Seq(StreamingV2ScanPushDown.getClass.getName.stripSuffix("$")), + level = Some(Level.WARN)) { + testStream(spark.readStream.table(tbl).select("data"))( + ProcessAllAvailable(), + // The scan that does not read "data" is replaced by one of all the columns before its + // stream is created, so the source only has the stream of the scan of all the columns. + assertReadSchema("id", "data", "extra"), + AssertOnQuery(_ => table.createdMicroBatchStreams.get() == 1), + CheckAnswer(Row("a"), Row("b"))) + } + val warnings = logAppender.loggingEvents.map(_.getMessage.getFormattedMessage) + assert(warnings.exists { message => + message.contains("could not be built with the pushdowns (columns: data, filters: none)") && + message.contains("does not have the columns data") && + message.contains("It reads all the columns with no filter instead") + }, warnings) + } + } + + test("read the event-time column of a watermark that the query does not output") { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, data string, ts timestamp)") + sql(s"INSERT INTO $tbl VALUES " + + "(1, 'a', timestamp_seconds(0)), (2, 'b', timestamp_seconds(10))") + val df = spark.readStream.table(tbl).withWatermark("ts", "5 seconds").select("data") + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("data", "ts"), + CheckAnswer(Row("a"), Row("b")), + // The watermark gets the event times of the rows. + AssertOnQuery { query => + query.recentProgress.filter(_.numInputRows > 0).map(_.eventTime.get("max")).toSeq == + Seq("1970-01-01T00:00:10.000Z") + }) + } + } + + test("read only the columns of a stream-static join") { + withSourceTable() { + val static = spark.range(2, 4).toDF("sid") + val source = spark.readStream.table(tbl) + val df = source.join(static, source("id") === static("sid")).select("data") + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("id", "data"), + CheckAnswer(Row("b")), + insert("(3, 'c', 30), (4, 'd', 40)"), + ProcessAllAvailable(), + CheckAnswer(Row("b"), Row("c"))) + } + } + + test("read only the columns of each side of a stream-stream join, also after a restart") { + withSourceTable() { + val left = spark.readStream.table(tbl).select("id", "data") + val right = spark.readStream.table(tbl).selectExpr("id AS rid", "extra") + val df = left.join(right, left("id") === right("rid")).select("data", "extra") + testStream(df)( + ProcessAllAvailable(), + assertReadSchemas(Seq("id", "data"), Seq("id", "extra")), + CheckAnswer(Row("a", 10), Row("b", 20)), + StopStream, + insert("(3, 'c', 30)"), + StartStream(), + ProcessAllAvailable(), + assertReadSchemas(Seq("id", "data"), Seq("id", "extra")), + CheckAnswer(Row("a", 10), Row("b", 20), Row("c", 30))) + } + } + + test("keep the layout of the state of a stream-stream join across the values of the conf " + + "when a scan returns the columns in another order") { + withPushDownConfsRestored { + val (leftTable, rightTable) = ("testcat.l", "testcat.r") + withTable(leftTable, rightTable) { + sql(s"CREATE TABLE $leftTable (k bigint, v bigint, unused int) TBLPROPERTIES " + + s"('${InMemoryBaseTable.SIMULATE_REORDERED_COLUMN_PRUNING}' = 'true')") + sql(s"CREATE TABLE $rightTable (rk bigint, w int)") + // The left row is kept in the state of the join, which is written without pruning first. + sql(s"INSERT INTO $leftTable VALUES (1, 100, 0)") + val left = spark.readStream.table(leftTable) + val right = spark.readStream.table(rightTable) + val df = left.join(right, $"k" === $"rk").select("v", "w") + testStream(df)( + StartStream(additionalConfs = pruningEnabled(false)), + ProcessAllAvailable(), + assertReadSchemas(Seq("k", "v", "unused"), Seq("rk", "w")), + CheckAnswer(), + StopStream, + // The pruned scan returns (v, k), and the join still gets (k, v). + insert("(1, 7)", rightTable), + StartStream(additionalConfs = pruningEnabled(true)), + ProcessAllAvailable(), + assertReadSchemas(Seq("v", "k"), Seq("rk", "w")), + CheckAnswer(Row(100L, 7)), + // And the other way around. + insert("(2, 200, 0)", leftTable), + ProcessAllAvailable(), + StopStream, + insert("(2, 8)", rightTable), + StartStream(additionalConfs = pruningEnabled(false)), + ProcessAllAvailable(), + CheckAnswer(Row(100L, 7), Row(200L, 8))) + } + } + } + + test("keep the state of a streaming aggregation across the values of the conf when a scan " + + "returns the columns in another order") { + withPushDownConfsRestored { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (k bigint, v bigint, unused int) TBLPROPERTIES " + + s"('${InMemoryBaseTable.SIMULATE_REORDERED_COLUMN_PRUNING}' = 'true')") + sql(s"INSERT INTO $tbl VALUES (1, 10, 0), (2, 20, 0)") + val df = spark.readStream.table(tbl).groupBy("k").agg(sum("v")) + testStream(df, OutputMode.Complete())( + StartStream(additionalConfs = pruningEnabled(false)), + ProcessAllAvailable(), + assertReadSchema("k", "v", "unused"), + CheckAnswer(Row(1L, 10L), Row(2L, 20L)), + StopStream, + insert("(1, 1, 0)"), + StartStream(additionalConfs = pruningEnabled(true)), + ProcessAllAvailable(), + assertReadSchema("v", "k"), + CheckAnswer(Row(1L, 11L), Row(2L, 20L)), + StopStream, + insert("(2, 2, 0)"), + StartStream(additionalConfs = pruningEnabled(false)), + ProcessAllAvailable(), + CheckAnswer(Row(1L, 11L), Row(2L, 22L))) + } + } + } + + test("fail to plan a micro-batch that reads a column that its scan does not read") { + withSourceTable() { + testStream(spark.readStream.table(tbl).select("data"))( + ProcessAllAvailable(), + AssertOnQuery { query => + val relation = query.logicalPlan.collectFirst { + case relation: StreamingDataSourceV2ScanRelation => relation + }.get + assert(relation.output.map(_.name) === Seq("id", "data", "extra")) + assert(relation.prunedOutput.map(_.map(_.name)) === Some(Seq("data"))) + assert(relation.simpleString(10).contains("PrunedOutput: [data#")) + val offset = relation.stream.initialOffset() + val batch = relation.copy(startOffset = Some(offset), endOffset = Some(offset)) + // Unlike the query, this plan reads all the columns of the relation. + val e = intercept[SparkException] { + spark.sessionState.planner.plan(Project(relation.output, batch)).next() + } + assert(e.getCondition === "INTERNAL_ERROR") + assert(e.getMessage.contains("does not read the columns")) + assert(e.getMessage.contains(SQLConf.STREAMING_V2_COLUMN_PRUNING_ENABLED.key)) + true + }) + } + } + + test("read only the columns that a continuous query needs") { + val input = ContinuousMemoryStream.singlePartition[(Long, String, Int)] + val table = new PruningContinuousTable(input, takesFilters = false) + val df = Dataset.ofRows(spark, StreamingRelationV2(None, "pruning_continuous", table, + CaseInsensitiveStringMap.empty(), DataTypeUtils.toAttributes(table.schema()), None, None, + None)).toDF("id", "data", "extra").where("extra > 10").select("data") + withTempView("continuous_pruning") { + val query = df.writeStream.format("memory").queryName("continuous_pruning") + .trigger(Trigger.Continuous(100)).start() + try { + input.addData((1L, "a", 10), (2L, "b", 20), (3L, "c", 30)) + eventually(timeout(streamingTimeout)) { + checkAnswer(spark.table("continuous_pruning"), Seq(Row("b"), Row("c"))) + } + val readSchemas = query.asInstanceOf[StreamingQueryWrapper].streamingQuery.logicalPlan + .collect { case r: StreamingDataSourceV2ScanRelation => r.scan.readSchema().fieldNames } + assert(readSchemas.map(_.toSeq) === Seq(Seq("_2", "_3"))) + } finally { + query.stop() + } + } + } + + test("push the filters of a streaming query down to the scan of its source") { + withPushDownConfsRestored { + withSourceTable() { + val df = spark.readStream.table(tbl).where("extra > 10").select("data") + testStream(df)( + StartStream(additionalConfs = filterPushdownEnabled(true)), + ProcessAllAvailable(), + assertPushedFilters("GreaterThan(extra,10)"), + // The scan of the in-memory table does not apply the filters, which Spark evaluates. + CheckAnswer(Row("b")), + StopStream, + insert("(3, 'c', 30), (4, 'd', 5)"), + StartStream(additionalConfs = filterPushdownEnabled(false)), + ProcessAllAvailable(), + assertPushedFilters(), + CheckAnswer(Row("b"), Row("c")), + StopStream, + insert("(5, 'e', 50), (6, 'f', 6)"), + StartStream(additionalConfs = filterPushdownEnabled(true)), + ProcessAllAvailable(), + assertPushedFilters("GreaterThan(extra,10)"), + CheckAnswer(Row("b"), Row("c"), Row("e"))) + } + } + } + + test("push down only the filters that every read of a relation applies") { + withSourceTable() { + val source = spark.readStream.table(tbl) + val df = source.where("extra > 10").union(source.where("extra > 10 AND id > 1")) + testStream(df)( + ProcessAllAvailable(), + assertPushedFilters("GreaterThan(extra,10)"), + CheckAnswer(Row(2L, "b", 20), Row(2L, "b", 20))) + } + } + + // Each case is a condition on a value that the optimizer of each micro-batch folds into its own + // value, such as its current time, and the column that it reads. + Seq( + "ts < current_timestamp()" -> "ts", + "ts < now()" -> "ts", + "CAST(ts AS DATE) < current_date()" -> "ts", + "CAST(ts AS TIMESTAMP_NTZ) < localtimestamp()" -> "ts", + "tod < current_time()" -> "tod", + "CAST(tod AS TIMESTAMP_NTZ) < TIMESTAMP_NTZ'2100-01-01 00:00:00'" -> "tod", + "CAST(tod AS TIMESTAMP) < TIMESTAMP'2100-01-01 00:00:00'" -> "tod", + "data = current_database()" -> "data", + "data = current_timezone()" -> "data", + // The casts of the special strings, which SpecialDatetimeValues folds. + "d = 'today'" -> "d", + "ts < 'now'" -> "ts", + "d = to_date('today')" -> "d", + "d < CAST('tomorrow' AS DATE)" -> "d", + "CAST(ts AS TIMESTAMP_NTZ) > CAST('yesterday' AS TIMESTAMP_NTZ)" -> "ts", + // The casts of strings with only a time to timestamps, which take the current date. + "ts < '12:00:00'" -> "ts", + "ts > CAST('T09:30:00' AS TIMESTAMP)" -> "ts" + ).foreach { case (condition, column) => + test(s"do not push down a filter on a value that depends on when it is evaluated: $condition") { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, data string, ts timestamp, tod time, d date)") + val df = spark.readStream.table(tbl).where(s"id > 1 AND $condition").select("id") + val reads = StreamingV2ScanPushDown.reads(spark, df.queryExecution.analyzed, None) + assert(reads.size === 1) + val read = reads.values.head + // The column of the condition is still read, but only the filter on the id is pushed down. + assert(read.requiredColumns.map(_.name).toSet === Set("id", column)) + assert(read.filters.nonEmpty) + assert(read.filters.forall(_.references.map(_.name).toSet == Set("id")), read.filters) + } + } + } + + test("do not push down a filter on the current time of a micro-batch") { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, data string, ts timestamp)") + sql(s"INSERT INTO $tbl VALUES " + + "(1, 'a', timestamp_seconds(0)), (2, 'b', timestamp_seconds(10))") + // Each micro-batch has its own current time, so the time when the query started cannot be + // pushed down in its place. + Seq("current_timestamp()", "now()").foreach { time => + val df = spark.readStream.table(tbl).where(s"id > 0 AND ts < $time").select("data") + testStream(df)( + ProcessAllAvailable(), + assertPushedFilters("GreaterThan(id,0)"), + CheckAnswer(Row("a"), Row("b"))) + } + } + } + + test("read the columns and apply the filters of every reference to a common table expression") { + withSourceTable() { + withTempView("s") { + spark.readStream.table(tbl).createOrReplaceTempView("s") + // Both references to the common table expression have the relation of the source, so its + // scan reads the columns of both, and applies no filter, as only one of them filters. + val df = sql("WITH c AS (SELECT * FROM s) " + + "SELECT a.id, b.extra FROM c a JOIN c b ON a.id = b.id WHERE a.data = 'a'") + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("id", "data", "extra"), + assertPushedFilters(), + CheckAnswer(Row(1L, 10)), + insert("(3, 'a', 30)"), + ProcessAllAvailable(), + CheckAnswer(Row(1L, 10), Row(3L, 30))) + + // Each reference applies its own filter. + val pruned = sql("WITH c AS (SELECT * FROM s) " + + "SELECT data FROM c WHERE extra > 15 UNION ALL SELECT data FROM c WHERE extra < 15") + testStream(pruned)( + ProcessAllAvailable(), + assertReadSchema("data", "extra"), + assertPushedFilters(), + CheckAnswer(Row("a"), Row("a"), Row("b"))) + } + } + } + + test("push down no filter that the optimizer infers from the filters of another source") { + val otherTable = "testcat.other" + withSourceTable() { + withTable(otherTable) { + sql(s"CREATE TABLE $otherTable (oid bigint, value string)") + val source = spark.readStream.table(tbl) + val other = spark.readStream.table(otherTable).where("oid < 3") + val df = source.join(other, source("id") === other("oid")).select("data", "value") + testStream(df)( + ProcessAllAvailable(), + // The optimizer infers a filter on the id of the source from the one on the other + // source, which is not pushed down: a scan only gets the filters that the query applies + // to its own source. + AssertOnQuery { query => + assert(pushedFilters(query).sortBy(_.mkString) === Seq(Nil, Seq("LessThan(oid,3)"))) + true + }, + CheckAnswer(), + insert("(1, 'x'), (2, 'y'), (3, 'z')", otherTable), + ProcessAllAvailable(), + CheckAnswer(Row("a", "x"), Row("b", "y")), + insert("(3, 'c', 30)"), + ProcessAllAvailable(), + CheckAnswer(Row("a", "x"), Row("b", "y"))) + } + } + } + + test("plan the micro-batches whose filters imply the pushed filters in another form") { + val otherTable = "testcat.other" + withSourceTable() { + withTable(otherTable) { + sql(s"CREATE TABLE $otherTable (oid bigint, value string)") + sql(s"INSERT INTO $otherTable VALUES (1, 'x'), (2, 'y')") + // A micro-batch infers id = 2 from the equality of the static side, and puts its value in + // place of the column in the pushed filter id > 0, which it then drops: id = 2 implies it. + val source = spark.readStream.table(tbl) + val static = spark.range(1, 3).toDF("sid").where("sid = 2") + val joined = source.where("id > 0").join(static, source("id") === static("sid")) + testStream(joined.select("data"))( + ProcessAllAvailable(), + assertPushedFilters("GreaterThan(id,0)"), + CheckAnswer(Row("b"))) + // So does a stream-stream join, in the micro-batches where both sources have data. + val other = spark.readStream.table(otherTable).where("oid = 2") + val streams = source.where("id > 0").join(other, source("id") === other("oid")) + testStream(streams.select("data", "value"))( + ProcessAllAvailable(), + CheckAnswer(Row("b", "y")), + insert("(2, 'bb', 22)"), + insert("(2, 'yy')", otherTable), + ProcessAllAvailable(), + CheckAnswer(Row("b", "y"), Row("b", "yy"), Row("bb", "y"), Row("bb", "yy"))) + } + } + } + + test("plan the micro-batches that fold a value of their own into a pushed filter") { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, owner string, d date)") + sql(s"INSERT INTO $tbl VALUES (1, current_user(), current_date()), (2, 'other', NULL)") + // A micro-batch folds current_user() and now() into literals, which it then puts in place + // of the columns in the pushed filters owner <> '' and d >= DATE'2020-01-01'. + testStream(spark.readStream.table(tbl).where("owner <> '' AND owner = current_user()") + .select("id"))( + ProcessAllAvailable(), + assertPushedFilters("Not(EqualTo(owner,))"), + CheckAnswer(Row(1L))) + testStream(spark.readStream.table(tbl) + .where("d >= DATE'2020-01-01' AND d = to_date(now())").select("id"))( + ProcessAllAvailable(), + assertPushedFilters("GreaterThanOrEqual(d,2020-01-01)")) + } + } + + Seq( + "CAST('today' AS DATE) < DATE'2100-01-01'", + "CAST('12:00:00' AS TIMESTAMP) < TIMESTAMP'2100-01-01 00:00:00'" + ).foreach { condition => + test(s"read the columns of a condition on a string that depends on when it is cast: " + + condition) { + withSourceTable() { + // The condition takes the date of each micro-batch, so the scan reads both columns. + val df = spark.readStream.table(tbl).select(expr(s"CASE WHEN $condition THEN data " + + "ELSE CAST(extra AS STRING) END AS v")) + val reads = StreamingV2ScanPushDown.reads(spark, df.queryExecution.analyzed, None) + assert(reads.values.map(_.requiredColumns.map(_.name).toSet).toSeq === + Seq(Set("data", "extra"))) + testStream(df)( + ProcessAllAvailable(), + assertReadSchema("data", "extra"), + CheckAnswer(Row("a"), Row("b"))) + } + } + } + + test("push down no filter after a filter that is not pushed down") { + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id bigint, owner string, qty int, payload string)") + sql(s"INSERT INTO $tbl VALUES (1, current_user(), 1, '1'), (2, 'other', 0, 'abc')") + val raise = "CASE WHEN qty <= 0 THEN raise_error('bad qty') ELSE true END" + def pushed(df: DataFrame): Seq[Set[String]] = { Review Comment: This moved with the filter pushdown to SPARK-60926 (#60129), whose open items list tests with `SupportsPushDownV2Filters` and `SupportsPushDownCatalystFilters` builders, and of the order in which a micro-batch evaluates the filters, with your `RLIKE` case. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala: ########## @@ -43,6 +43,32 @@ import org.apache.spark.util.collection.Utils object PushDownUtils extends Logging { + /** Returns whether [[pushFilters]] may push filters down to `scanBuilder`. */ + def supportsFilters(scanBuilder: ScanBuilder): Boolean = scanBuilder match { + case _: SupportsPushDownFilters | _: SupportsPushDownV2Filters | + _: SupportsPushDownCatalystFilters => true + case _ => false + } + + /** + * Returns the filters of `filters` that [[pushFilters]] gives to `scanBuilder` in some form, in + * their order: those that it translates to data source filters or predicates, or the + * deterministic ones for a [[SupportsPushDownCatalystFilters]] builder. The others are only + * evaluated after the scan. + */ + def filtersOfScanBuilder( Review Comment: `filtersOfScanBuilder` left this PR with the filter pushdown, which moved to SPARK-60926 (#60129). Sharing one check with `pushFilters` is listed there. -- 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]
