dongjoon-hyun commented on code in PR #59943: URL: https://github.com/apache/spark/pull/59943#discussion_r4476491679
########## 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: Following up on the [thread](https://github.com/apache/spark/pull/59943#discussion_r4474774587): `castsTimeOnly` only replaces a cast whose child is foldable in the plan before the optimization, but the optimizer can make the child of a cast foldable afterwards. For example, with `readStream.table(t).withColumn("cutoff", lit("12:00:00")).where("ts < CAST(cutoff AS TIMESTAMP)")`, the child of the cast is the attribute `cutoff`, so `forStartup` keeps the cast. `PushPredicateThroughNonJoin` then pushes the filter through the `Project`, which puts `'12:00:00'` in place of `cutoff`, and `ConstantFolding` folds the cast into `TIMESTAMP'2026-11-05 12:00:00'` when the query starts on 2026-11-05. The filter is deterministic and can be translated, so it is pushed down, and `withPushedFilters` applies it to every micro-batch: from 2026-11-06 on, it drops the rows between the two times with no error, as in that thread. `ConstantPropagation` does the same with `tag = '12:00:00' AND ts < CAST(t ag AS TIMESTAMP)`, and a `CASE WHEN` on such a cast can make a scan miss a column in the same way. The new tests only cast literals. Could `castsTimeOnly` replace the casts of a string to `TIMESTAMP` or `TIMESTAMP_LTZ(p)` whatever their child is (see the comment on L256)? ########## 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: Correction to my suggestion in the [thread](https://github.com/apache/spark/pull/59943#discussion_r4474774614): with the `Filter` holding only the filters that the builder gets, a filter of the prefix can be pushed down while a filter written before it is not, so each micro-batch evaluates them out of order, which the `takeWhile` of L217 is meant to prevent. For a `SupportsPushDownV2Filters` builder with ANSI on, both filters of `payload RLIKE '^[0-9]+$' AND CAST(payload AS INT) > 0` are in the prefix. `V2ExpressionBuilder` translates the cast (`V2ExpressionBuilder.scala:175-176`) but not `RLIKE`, so `pushedFilters` is `[CAST(payload AS INT) > 0]`, and `CombineFilters` puts it before `RLIKE` in each micro-batch (`Optimizer.scala:2430`). A row with the payload `'abc'` then fails the query with `CAST_INVALID_INPUT`, while it passed before this PR. A batch scan orders its post-scan filters this way too, but the doc of `pushableFilters` (L211-214) says that a pushed filter does not se e a row that a filter written before it drops, and that of `STREAMING_FILTER_PUSHDOWN` that the pushed filters are those that the query applies first. The filters that `SupportsPushDownFilters` translates cannot fail, so the suite does not see it (see the comment on `StreamingV2ScanPushDownSuite.scala:747`). Could the prefix end at the first filter that the builder does not get, e.g., `normalized.takeWhile(f => PushDownUtils.filtersOfScanBuilder(builder, Seq(f)).nonEmpty)`? The test on `StreamingV2ScanPushDownSuite.scala:969` would then push nothing for `id + extra > 10 AND extra > 15`, which it could write as `extra > 15 AND id + extra > 10`. ########## 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: Besides the case of the comment on L272, `castsTimeOnly` only looks at a cast to a timestamp type itself, while a cast to an `ARRAY`, `STRUCT` or `MAP` of timestamps casts each string element with `DateTimeUtils.stringToTimestamp` (`Cast.scala:922`), which takes `LocalDate.now(zoneId)` for a string with only a time too (`SparkDateTimeUtils.scala:907-908`). For example, `ts < element_at(CAST(array('12:00:00') AS ARRAY<TIMESTAMP>), 1)` and `ts < CAST(named_struct('t', '12:00:00') AS STRUCT<t: TIMESTAMP>).t` keep their casts, which `ConstantFolding` folds with the date of the start, so the filter is pushed down and drops rows from the next day on, as in the comment on L272. Could the check also cover the casts whose types have a string and a timestamp with a local time zone at the same place, at any depth? ########## 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: `pushed` returns the filters of `StreamingV2ScanPushDown.reads`, before `filtersOfScanBuilder` picks those that the builder gets, and all the scan builders of the suite, that of `InMemoryBaseTable` and those of L1106 and L1279, implement `SupportsPushDownFilters`. So neither the `SupportsPushDownV2Filters` and `SupportsPushDownCatalystFilters` branches of `filtersOfScanBuilder` nor the order in which a micro-batch evaluates the filters are tested. For example, L765-766 checks that the prefix of `$raise AND id > 0 AND owner = current_user() AND qty > 0` has `qty` and `id`, but the builder only gets `id > 0`, which each micro-batch evaluates before the `raise_error` filter: that passes only because `id > 0` cannot fail, and the case of the comment on `StreamingV2ScanPushDown.scala:326` is not caught. Could a test use a builder that implements `SupportsPushDownV2Filters`, and check with ANSI on that `payload RLIKE '^[0-9]+$' AND CAST(payload AS INT) > 0` reads a source with a row whose payload is `'abc'`, and which condition the `Filter` over the relation has, as the test on L969 does? ########## 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: Following up on the [thread](https://github.com/apache/spark/pull/59943#discussion_r4474774587): this case picks the placeholders from a list of the shapes that take the current date, the pattern `CURRENT_LIKE`, the rewrites of `ComputeCurrentTime` and `SpecialDatetimeValues`, and now `castsTimeOnly`, and each round has added one to it. The comments on L268 and L272 show shapes that it still misses, some of which only the optimization produces, so no check of the plan before the optimization sees them. All of them are casts of strings to timestamps with a local time zone, which take the date of when they are evaluated. Could this case replace every cast of a string to `TIMESTAMP` or `TIMESTAMP_LTZ(p)`, at any depth of its types and whatever its child is? A filter on `CAST(s AS TIMESTAMP)` of a string column `s` would then not be pushed down, nor the filters after it, while `SupportsPushDownFilters` cannot take it anyway, and the placeholder keeps the columns that the cast reads. Another way would be for `ComputeCurrentTime` to give these casts the current date, as it does for the casts of a `TIME`, which `ComputeCurrentTime.applyForExpression(cast) != cast` above would then cover, but that changes every query with such a cast, so it would be a separate change. ########## 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: Minor: following up on the [thread](https://github.com/apache/spark/pull/59943#discussion_r4474774655), `filtersOfScanBuilder` repeats the dispatch of `pushFilters` on the builder interfaces and its choice of the filters that each one gets: `translateFilter` (L64) for the loop of L102-111, `deterministic` and `translateFilterV2` (L67) for L134-143, and `deterministic` (L68) for L167. With `supportsFilters`, the builder interfaces are listed three times, and a change to one of these choices, e.g., skipping the nondeterministic filters for `SupportsPushDownFilters` as the V2 branch does, has to be made twice, or the `Filter` of the micro-batches and the filters that the builder gets differ. Could the branches of `pushFilters` and `filtersOfScanBuilder` share one check of whether a builder gets a filter? ########## sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java: ########## @@ -170,5 +170,44 @@ public enum TableCapability { * * @since 4.4.0 */ - EXTERNAL_LOOKUP + EXTERNAL_LOOKUP, + + /** + * Signals that the scans of the streaming reads of the table may only read the columns that the + * streaming query reads, which Spark asks a scan builder that implements + * {@link org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns} for, as it does + * for the scans of batch reads. + * <p> + * Spark builds the scan of a streaming read once, when the query starts, from the plan of the + * query optimized then, and creates the micro-batch or continuous stream of the read from that + * scan. By returning this capability, a table declares that the partition readers of such a + * stream return the columns of {@link org.apache.spark.sql.connector.read.Scan#readSchema()}, + * as those of the batches of the scan do. Spark asks the scan for its read schema before it Review Comment: Minor: `buildScan` only asks a scan for its read schema before it creates the stream when the scan is pruned (`StreamingV2ScanPushDown.scala:385-391`). A scan of all the columns, with or without pushed filters, goes through `scanOfAllColumns` (L353-356) or `Right(None)` (L386), which call `createStream` without `readSchema()`, although `build` calls `pruneColumns` on a `SupportsPushDownRequiredColumns` builder. `SupportsPushDownRequiredColumns.java:46-47` says the same. Could the docs say that Spark asks a pruned scan for its read schema, or could `buildScan` ask every scan whose builder got `pruneColumns`? -- 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]
