HyukjinKwon commented on code in PR #59943: URL: https://github.com/apache/spark/pull/59943#discussion_r4472657886
########## 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: Done in becbceb: the guard is now `case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast`, next to `case e if e.containsPattern(CURRENT_LIKE)`, so only a cast is rewritten to see whether it takes the current date. The new test "push down a filter on a cast that does not depend on when it is evaluated" checks that `CAST(data AS TIMESTAMP) < TIMESTAMP'...'` is still pushed. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala: ########## @@ -1142,6 +1145,78 @@ private[sql] object DataSourceV2Strategy extends Logging { withFilter } } + + /** + * Returns the output of the scan node of a streaming relation: the columns that its scan reads. + * The projections and the filters over the relation, which the scan node is planned with, must + * still imply the filters that the scan got, and read no column of the relation that the scan + * does not read, if it was built to read fewer columns than the relation outputs (see + * `StreamingV2ScanPushDown`). + */ + private def streamingScanOutput( + relation: StreamingDataSourceV2ScanRelation, + project: Seq[NamedExpression], + filters: Seq[Expression]): Seq[AttributeReference] = { + val missingFilters = relation.pushedFilters.filterNot(implies(filters, _)) + if (missingFilters.nonEmpty) { + throw SparkException.internalError( + s"The plan does not apply the filters ${missingFilters.map(_.sql).mkString(", ")} " + + s"that the scan of the streaming relation ${relation.name} got. Set " + + s"${SQLConf.STREAMING_V2_FILTER_PUSHDOWN_ENABLED.key} to false to push no filter " + + "down to the scans of streaming queries.") + } + relation.prunedOutput match { + case Some(prunedOutput) => + val missing = AttributeSet((project ++ filters).flatMap(_.references)) -- prunedOutput + if (missing.nonEmpty) { + throw SparkException.internalError( + s"The scan of the streaming relation ${relation.name} does not read the columns " + + s"${missing.toSeq.map(_.name).mkString(", ")} that the plan reads from it. Set " + + s"${SQLConf.STREAMING_V2_COLUMN_PRUNING_ENABLED.key} to false to make the scans " + + "of streaming queries read all the columns of their sources.") + } + prunedOutput + case None => + relation.output + } + } + + /** + * Returns whether `filters`, the filters over a streaming relation in a micro-batch, imply + * `pushed`, a filter that its scan got when the query started: they have it, or it holds for the + * values of the columns that they make equal to literals. The optimizer of a micro-batch puts + * these values in place of the columns in the other filters, with `ConstantPropagation`, for + * example with an equality that it infers through a join, and then drops a filter that becomes + * true, as `id > 1` does with `id = 5`. + */ + private def implies(filters: Seq[Expression], pushed: Expression): Boolean = { Review Comment: Gone with the check in becbceb (see L1160): `streamingScanOutput` only checks the columns 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) Review Comment: Done in becbceb: the pushdown runs in a `Try` that returns the scan or why its read schema does not fit, and a local `fallBack(reason, cause)` logs the warning and builds the scan of all the columns. It is called with the throwable for a `Failure`, and without it for the `Left` of `prunedScanOutput`, outside the `Try`, so a failure of the fallback itself is not caught again. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala: ########## @@ -123,39 +125,55 @@ class ColumnarScanBuilder( override def pruneColumns(requiredSchema: StructType): Unit = { // Only top-level columns are pruned by the data source. Spark prunes the nested fields of the - // columns it reads. + // columns it reads. The required columns are only recorded here, see ColumnarScan. val required = StructType(requiredSchema.fieldNames.flatMap(name => schema.find(_.name == name))) - if (required.length < schema.length && batchPlanner.pruneColumns(required)) { - readSchema = required + if (required.length < schema.length) { + this.requiredSchema = Some(required) } } // The scan takes over the planner, so it is built once. private lazy val scan = new ColumnarScan( - shortName, schema, readSchema, dataSource, Option(planner), pushedLimit) + shortName, schema, requiredSchema, dataSource, Option(planner), pushedLimit) override def build(): Scan = scan } /** + * A scan of a [[ColumnarDataSource]]. It reads the top-level columns that the query needs, the + * columns of its read schema: a batch scan asks its planner to prune the other columns when it + * plans its partitions, and drops them from the batches of the reader if the planner does not, + * while a streaming scan drops them from the batches of its stream reader, which reads all the + * columns of the full schema. + * + * @param requiredSchema the top-level columns that the query needs, if fewer than those of the + * full schema * @param pushedDownPlanner the planner that the operations were pushed down to, if any. Otherwise, * the scan creates one when its batch is planned. */ class ColumnarScan( shortName: String, fullSchema: StructType, - prunedSchema: StructType, + requiredSchema: Option[StructType], dataSource: ColumnarDataSource, pushedDownPlanner: Option[ColumnarBatchPlanner], pushedLimit: Option[Int]) extends Scan with SupportsMetadata { - // Spark may call `toBatch` more than once for the same scan; the partitions are planned once. - private lazy val batch = new ColumnarBatchScan( - pushedDownPlanner.getOrElse(dataSource.batchPlanner(fullSchema))) + // The ordinals in the full schema of the columns that the scan reads, if it reads fewer. + private def requiredOrdinals: Option[Array[Int]] = + requiredSchema.map(_.fieldNames.map(fullSchema.fieldIndex)) + + // Spark may call `toBatch` more than once for the same scan; the partitions are planned once, + // after the planner prunes the columns, if it does. + private lazy val batch = { + val planner = pushedDownPlanner.getOrElse(dataSource.batchPlanner(fullSchema)) + val pruned = requiredSchema.forall(planner.pruneColumns) + new ColumnarBatchScan(planner, if (pruned) None else requiredOrdinals) + } - override def readSchema(): StructType = prunedSchema + override def readSchema(): StructType = requiredSchema.getOrElse(fullSchema) Review Comment: Done in becbceb: when the scan drops columns after the read, its metadata also has `SourceReadSchema`, the columns that the data source reads: always for a stream, whose reader reads all the columns, and for a batch whose planner does not prune them. `ReadSchema` stays the read schema of the scan, as `Scan.readSchema()` returns it. The columnar and native suites check both cases. -- 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]
