HyukjinKwon commented on code in PR #59943: URL: https://github.com/apache/spark/pull/59943#discussion_r4474060160
########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,419 @@ +/* + * 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, ExpressionSet, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime, InferFiltersFromConstraints} +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_TO_TIMESTAMP, CURRENT_LIKE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +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, SupportsPushDownFilters, SupportsPushDownRequiredColumns, SupportsPushDownV2Filters} +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, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters +import org.apache.spark.sql.types.{DataType, StructType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +/** + * 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 deterministic filters without subqueries that the plan applies over every + * occurrence of the relation, which its scan may apply too + */ + 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 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 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, 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)), + filters.filter { filter => + filter.deterministic && !SubqueryExpression.hasSubquery(filter) && + !filter.exists(_.isInstanceOf[StartupValue]) + }) + 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. + */ + private def combine( + occurrences: Seq[(StreamingRelationV2, Reads)]): Map[StreamingRelationV2, Reads] = { + occurrences.groupBy(_._1).map { case (relation, reads) => + val filters = reads.map(r => ExpressionSet(r._2.filters)).reduce { (left, right) => + ExpressionSet(left.filter(right.contains)) + } + relation -> Reads( + AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters.toSeq) + } + } + + /** + * 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 => + node.transformExpressionsUpWithPruning( + bits => bits.containsPattern(CURRENT_LIKE) || bits.containsPattern(CAST_TO_TIMESTAMP)) { + // 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 rewrites, such as a cast of a time to a + // timestamp, which takes the current date. ComputeCurrentTime only rewrites casts besides + // the expressions with the pattern CURRENT_LIKE. + case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast => Review Comment: Fixed in 1873787: `forStartup` also replaces the casts that `SpecialDatetimeValues.applyForExpression` rewrites, next to those of `ComputeCurrentTime`, with the pattern `CAST` in the pruning, so the casts of `'today'`, `'now'`, `'tomorrow'` and `'yesterday'` to dates and timestamps become placeholders, and no filter on them is pushed. `to_date('today')` only stands for such a cast once `ReplaceExpressions` replaced it, which the optimizer does before `SpecialDatetimeValues`, so `forStartup` replaces these expressions first with `ReplaceExpressions.replace`, as the optimizer does; without that, the test of `to_date('today')` pushed `d = DATE'2026-11-04'`. The list of conditions now has `d = 'today'`, `ts < 'now'`, `d = to_date('today')`, `d < CAST('tomorrow' AS DATE)` and a cast of `'yesterday'` to `TIMESTAMP_NTZ`. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,419 @@ +/* + * 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, ExpressionSet, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime, InferFiltersFromConstraints} +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_TO_TIMESTAMP, CURRENT_LIKE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +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, SupportsPushDownFilters, SupportsPushDownRequiredColumns, SupportsPushDownV2Filters} +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, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters +import org.apache.spark.sql.types.{DataType, StructType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +/** + * 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 deterministic filters without subqueries that the plan applies over every + * occurrence of the relation, which its scan may apply too + */ + 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 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 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, 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)), Review Comment: Fixed by the same change in 1873787: the cast keeps its placeholder, so the optimization at the start cannot fold the condition, and the scan reads both columns. The new test "read the columns of a condition on a special string that depends on when it is cast" checks the columns of `CASE WHEN CAST('today' AS DATE) < DATE'2100-01-01' THEN data ELSE CAST(extra AS STRING) END`, and runs the query. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,419 @@ +/* + * 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, ExpressionSet, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime, InferFiltersFromConstraints} +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_TO_TIMESTAMP, CURRENT_LIKE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +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, SupportsPushDownFilters, SupportsPushDownRequiredColumns, SupportsPushDownV2Filters} +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, StreamingDataSourceV2ScanRelation} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters +import org.apache.spark.sql.types.{DataType, StructType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +/** + * 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 deterministic filters without subqueries that the plan applies over every + * occurrence of the relation, which its scan may apply too + */ + 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 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 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, 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)), + filters.filter { filter => + filter.deterministic && !SubqueryExpression.hasSubquery(filter) && + !filter.exists(_.isInstanceOf[StartupValue]) + }) + 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. + */ + private def combine( + occurrences: Seq[(StreamingRelationV2, Reads)]): Map[StreamingRelationV2, Reads] = { + occurrences.groupBy(_._1).map { case (relation, reads) => + val filters = reads.map(r => ExpressionSet(r._2.filters)).reduce { (left, right) => + ExpressionSet(left.filter(right.contains)) + } + relation -> Reads( + AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters.toSeq) + } + } + + /** + * 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 => + node.transformExpressionsUpWithPruning( + bits => bits.containsPattern(CURRENT_LIKE) || bits.containsPattern(CAST_TO_TIMESTAMP)) { + // 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 rewrites, such as a cast of a time to a + // timestamp, which takes the current date. ComputeCurrentTime only rewrites casts besides + // the expressions with the pattern CURRENT_LIKE. + case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast => + StartupValue(cast, cast.children) + } + } + + /** + * 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) && supportsFilters(builder) + val known = if (prunesColumns || pushesFilters) reads else None + val requiredOutput = known.filter(_ => prunesColumns) + .map(r => output.filter(r.requiredColumns.contains)) + .getOrElse(output) + val filters = known.filter(_ => pushesFilters) + .map(r => DataSourceStrategy.normalizeExprs(r.filters, output)) + .getOrElse(Nil) + + // Builds the scan of `columns` with `builder`, once it pushed `filters` down to it. + def build( + builder: ScanBuilder, + columns: Seq[AttributeReference], + filters: Seq[Expression]): Scan = { + if (filters.nonEmpty) { + // The plan of each micro-batch applies all the filters to the rows of the source, see + // `withPushedFilters`, so the scan may apply any of them, in any form. + val (pushed, _) = PushDownUtils.pushFilters(builder, filters, partitionFields = None) + val taken = pushed.fold(_.mkString(", "), _.mkString(", ")) + logInfo(log"The scan builder of the streaming table " + + log"${MDC(LogKeys.STREAMING_TABLE, table)} took the filters " + + log"${MDC(LogKeys.PUSHED_FILTERS, taken)}.") + } + builder match { + case pruning: SupportsPushDownRequiredColumns => + pruning.pruneColumns(DataTypeUtils.fromAttributes(columns)) + case _ => + } + builder.build() + } + // The scan of all the columns with no filter, as without these pushdowns. + def scanOfAllColumns(builder: ScanBuilder): StreamingScan = { + val scan = build(builder, output, Nil) + StreamingScan(scan, createStream(scan), None, Nil) + } + + if (requiredOutput.length == output.length && filters.isEmpty) { + scanOfAllColumns(builder) + } else { + // Logs why the scan with the pushdowns could not be built, and builds the scan of all the + // columns with no filter instead. + def fallBack(reason: String, cause: Option[Throwable]): StreamingScan = { + val columns = if (requiredOutput.length < output.length) { + requiredOutput.map(_.name).mkString(", ") + } else { + "all" + } + val filterList = if (filters.nonEmpty) filters.map(_.sql).mkString(", ") else "none" + val message = log"The scan of the streaming table " + + log"${MDC(LogKeys.STREAMING_TABLE, table)} could not be built with the pushdowns " + + log"(columns: ${MDC(LogKeys.COLUMN_NAMES, columns)}, filters: " + + log"${MDC(LogKeys.FILTER, filterList)}): ${MDC(LogKeys.REASON, reason)}. It reads all " + + log"the columns with no filter instead." + cause match { + case Some(e) => logWarning(message, e) + case None => logWarning(message) + } + scanOfAllColumns(table.newScanBuilder(options)) + } + // The scan with the pushdowns and its stream, or why the read schema of the pruned scan + // does not follow `SupportsPushDownRequiredColumns`. + val pushedDown = Try { + val scan = build(builder, requiredOutput, filters) + val prunedOutput = if (requiredOutput.length == output.length) { + Right(None) + } else { + prunedScanOutput(scan.readSchema(), output, requiredOutput).map { scanOutput => + if (scanOutput.map(_.exprId) == output.map(_.exprId)) None else Some(scanOutput) + } + } + prunedOutput.map(StreamingScan(scan, createStream(scan), _, filters)) + } + pushedDown match { + case Success(Right(streamingScan)) => streamingScan + case Success(Left(reason)) => fallBack(s"its read schema $reason", None) + case Failure(e) => fallBack(s"it failed with $e", Some(e)) + } + } + } + + /** + * Returns `relation`, the relation of a streaming source in the plan of the query, with the + * filters that its scan builder got applied to its rows. The query applies them anyway, as they + * are the filters that it applies to all the rows that it reads from the source, so the plan of + * each micro-batch has the same result with them. With them, it also applies all the filters + * that the scan may apply, in any form, however the plan of the micro-batch is optimized, so the + * scan cannot drop a row that the plan keeps. + */ + def withPushedFilters(relation: StreamingDataSourceV2ScanRelation): LogicalPlan = { + relation.pushedFilters.reduceOption(And).map(Filter(_, relation)).getOrElse(relation) Review Comment: Done in 1873787: `pushableFilters` selects the filters of each read as `splitPushable` does, in the order in which the plan evaluates them: all of them up to the first one that is not pushed down, and after it only those that cannot throw (`Expression.throwable`). With several reads of a relation, the filters that may throw are left out, since each read evaluates the pushed filters, in the order of the first read, before its own. The new test "push down no filter that may throw after a filter that is not pushed down" runs your query with a row of another owner whose quantity is 0, which fails with `USER_RAISED_EXCEPTION` when the `CASE` is pushed, and checks that a filter that cannot throw is still pushed after the filter on the owner, that one that may throw is pushed before it, and that none that may throw is pushed for two reads. ########## sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingV2ScanPushDownSuite.scala: ########## @@ -0,0 +1,1146 @@ +/* + * 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 + +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, And} +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) + 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. Review Comment: Done in 1873787: with the new property `simulate-streaming-filters`, the micro-batch streams of the in-memory table drop the rows that fail the pushed filters, with a small evaluator of the filters of the tests that keeps the rows it cannot decide. "drop the rows that fail the pushed filters in the stream of a source" checks the answers and `numInputRows`, which only counts the rows that the stream returns, and the tables of "plan the micro-batches whose filters fold the pushed filters in part" use it too, so their answers depend on the pushed filters. `PruningContinuousTable` can take the filters, and its readers then drop the rows that fail them: "filter the rows of a continuous stream with the filters that its scan builder took" checks the answer, the rows that the readers returned, and the `Filter` of the plan of the continuous query. The list at L565 has the special strings of your comment on L219. -- 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]
