zikangh commented on code in PR #59943: URL: https://github.com/apache/spark/pull/59943#discussion_r4472166224
########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,396 @@ +/* + * 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.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.{Attribute, AttributeReference, AttributeSet, 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.{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 +import org.apache.spark.sql.connector.expressions.filter.Predicate +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, DataSourceV2Strategy, PushDownUtils} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.internal.connector.{BatchOnlyFilterPushDown, SupportsPushDownCatalystFilters} +import org.apache.spark.sql.sources +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. + * + * 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 the query does not change, so each micro-batch is optimized as without these + * pushdowns: 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, and that its filters over the + * relation still imply the filters that the scan builder took. + * + * 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 took, which the scan may apply, so the + * plan of each micro-batch must still apply 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, nor on the data of the other relations: + * - 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. The optimizer of a + * micro-batch may not infer them, for example when the other relation is a source without new + * data in the micro-batch, which the optimizer removes with its filters. + * - 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. + */ + private 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. + // So is one that ComputeCurrentTime rewrites, such as a cast of a time to a timestamp, + // which takes the current date. + case e + if e.containsPattern(CURRENT_LIKE) || ComputeCurrentTime.applyForExpression(e) != e => Review Comment: this doesn't seem to cover SpecialDatetimeValues? e.g. `WHERE ts > 'now'` ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,396 @@ +/* + * 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.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.{Attribute, AttributeReference, AttributeSet, 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.{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 +import org.apache.spark.sql.connector.expressions.filter.Predicate +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, DataSourceV2Strategy, PushDownUtils} +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.internal.connector.{BatchOnlyFilterPushDown, SupportsPushDownCatalystFilters} +import org.apache.spark.sql.sources +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. + * + * 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 the query does not change, so each micro-batch is optimized as without these + * pushdowns: 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, and that its filters over the + * relation still imply the filters that the scan builder took. + * + * 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 took, which the scan may apply, so the + * plan of each micro-batch must still apply 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, nor on the data of the other relations: + * - 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. The optimizer of a + * micro-batch may not infer them, for example when the other relation is a source without new + * data in the micro-batch, which the optimizer removes with its filters. + * - 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. + */ + private 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. + // So is one that ComputeCurrentTime rewrites, such as a cast of a time to a timestamp, + // which takes the current date. + case e + if e.containsPattern(CURRENT_LIKE) || ComputeCurrentTime.applyForExpression(e) != e => + StartupValue(e.children, e.dataType, e.nullable) + } + } + + /** + * Builds the scan of a streaming source, and creates its stream with `createStream`. If its + * scan builder supports it, and `conf` enables it, the scan only reads the columns of `output` + * that the query reads, and 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 builder. + * + * A pruned scan must read the required columns and only columns of `output`, each once, as + * `SupportsPushDownRequiredColumns` requires. A scan that does not, or whose read schema cannot + * be read, is replaced by a scan of all the columns, as without pruning, before its stream is + * created. Spark asks a scan for its read schema before it creates its stream, as for a batch. + * + * @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 prunesColumns = conf.streamingV2ColumnPruningEnabled && + builder.isInstanceOf[SupportsPushDownRequiredColumns] + val pushesFilters = conf.streamingV2FilterPushdownEnabled && 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, and returns it with the filters that the builder took. + def build(builder: ScanBuilder, columns: Seq[AttributeReference]): (Scan, Seq[Expression]) = { + val pushedFilters = if (filters.nonEmpty) { + // The plan of the query keeps the filters, which Spark evaluates after the scan, so the + // scan may apply any of them. + val (pushed, _) = PushDownUtils.pushFilters(builder, filters, partitionFields = None) + takenFilters(filters, pushed) + } else { + Nil + } + builder match { + case pruning: SupportsPushDownRequiredColumns => + pruning.pruneColumns(DataTypeUtils.fromAttributes(columns)) + case _ => + } + (builder.build(), pushedFilters) + } + def withStream( + built: (Scan, Seq[Expression]), + prunedOutput: Option[Seq[AttributeReference]]): StreamingScan = { + val (scan, pushedFilters) = built + StreamingScan(scan, createStream(scan), prunedOutput, pushedFilters) + } + + if (requiredOutput.length == output.length) { + withStream(build(builder, output), None) + } else { + val built = build(builder, requiredOutput) + val scanOutput = try { + prunedScanOutput(built._1.readSchema(), output, requiredOutput).left.map(_ -> None) + } catch { + case NonFatal(e) => Left(s"cannot be read: $e" -> Some(e)) + } + scanOutput match { + case Right(scanOutput) => + val readsOutput = scanOutput.map(_.exprId) == output.map(_.exprId) + withStream(built, if (readsOutput) None else Some(scanOutput)) + case Left((reason, cause)) => + val columns = requiredOutput.map(_.name).mkString(", ") + val message = log"The scan of the streaming table " + + log"${MDC(LogKeys.STREAMING_TABLE, table)} was asked to read the columns " + + log"${MDC(LogKeys.COLUMN_NAMES, columns)}, but its read schema " + + log"${MDC(LogKeys.REASON, reason)}. It reads all the columns instead." + cause match { + case Some(e) => logWarning(message, e) + case None => logWarning(message) + } + withStream(build(table.newScanBuilder(options), output), None) + } + } + } + + /** + * Returns the filters of `filters` that a scan builder took, which the scan may apply: those + * whose translation is in what `PushDownUtils.pushFilters` returned as pushed, as the batch scans + * show them. The others, which cannot be translated or which the builder rejected, are only + * evaluated by Spark. + */ + private def takenFilters( + filters: Seq[Expression], + pushed: Either[Seq[sources.Filter], Seq[Predicate]]): Seq[Expression] = pushed match { + case Left(sourceFilters) => + val taken = sourceFilters.toSet + filters.filter { filter => + DataSourceStrategy.translateFilter(filter, supportNestedPredicatePushdown = true) + .exists(taken.contains) + } + case Right(predicates) => + val taken = predicates.toSet + filters.filter(DataSourceV2Strategy.translateFilterV2(_).exists(taken.contains)) + } + + /** Returns whether the scan builder of a streaming source takes filters. */ + private def supportsFilters(builder: ScanBuilder): Boolean = builder match { + case _: BatchOnlyFilterPushDown => false + case _: SupportsPushDownFilters | _: SupportsPushDownV2Filters | + _: SupportsPushDownCatalystFilters => true + case _ => false + } + + /** + * Returns the columns of `output` that a scan pruned to `requiredOutput` reads, as batch scans + * map them with `PushDownUtils.toOutputAttrs`, or why the read schema does not follow + * `SupportsPushDownRequiredColumns`. + */ + private def prunedScanOutput( + readSchema: StructType, + output: Seq[AttributeReference], + requiredOutput: Seq[AttributeReference]): Either[String, Seq[AttributeReference]] = { + val names = readSchema.fieldNames.toSeq + val outputNames = output.map(_.name).toSet + val unknown = names.filterNot(outputNames.contains) + val repeated = names.diff(names.distinct).distinct + if (unknown.nonEmpty) { + Left(s"${readSchema.simpleString} has the columns ${unknown.mkString(", ")}, which the " + + "relation does not output") + } else if (repeated.nonEmpty) { + Left(s"${readSchema.simpleString} has the columns ${repeated.mkString(", ")} more than once") + } else { + val scanOutput = PushDownUtils.toOutputAttrs(readSchema, output) + val missing = requiredOutput.filterNot(AttributeSet(scanOutput).contains) + if (missing.nonEmpty) { + Left(s"${readSchema.simpleString} does not have the columns " + + s"${missing.map(_.name).mkString(", ")}") + } else { + Right(scanOutput) + } + } + } + + /** + * Stands for the relation of a batch scan in the plan that is optimized when the query starts, + * which is never executed. + */ + private case class BatchRelation(output: Seq[Attribute]) + extends LeafNode with MultiInstanceRelation { + override def newInstance(): LogicalPlan = copy(output = output.map(_.newInstance())) + override def computeStats(): Statistics = Statistics(sizeInBytes = conf.defaultSizeInBytes) + } + + /** + * Stands for an expression whose value depends on when it is evaluated, in the plan that is + * optimized when the query starts, which is never executed. It keeps the children of the + * expression. It is not deterministic for the optimizer, as the current time of a micro-batch + * is not, so that the optimizer neither folds it nor moves operators around it as it could + * with a value. It is deterministic once `planned`, as the values that the planner of a + * micro-batch gets are. + */ + private case class StartupValue( + children: Seq[Expression], + dataType: DataType, + nullable: Boolean, + planned: Boolean = false) + extends Expression with Unevaluable { + override lazy val deterministic: Boolean = planned && children.forall(_.deterministic) Review Comment: since StartupValue is nondeterministic at startup but current_user() is a literal in each micro-batch, InlineCTE makes different decisions in the two runs. ########## sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingV2ScanPushDownSuite.scala: ########## @@ -0,0 +1,936 @@ +/* + * 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.atomic.AtomicInteger + +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.plans.logical.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, 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.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("was asked to read the columns data") && + message.contains("does not have the columns data") && + message.contains("It reads all the columns 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: since the in-memory stream ignores the pushed filters, nothing tests a source that actually drops rows. could we add one under a stateful operator, plus a SupportsPushDownV2Filters builder, a real-time mode case, and a streaming CDC read through ChangelogTable? -- 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]
