HyukjinKwon commented on code in PR #59943: URL: https://github.com/apache/spark/pull/59943#discussion_r4472655259
########## 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( Review Comment: Fixed in becbceb differently: since each micro-batch now applies all the filters that the builder got to the rows of the relation (see L1160), the record no longer needs to be what the builder took. `pushedFilters` holds all the filters that the builder got, so a builder that takes `a > 1 OR b > 2` of `a > 1 OR (b > 2 AND c > 3)`, or any other form, still applies only what the plan applies, and nothing is translated twice: `takenFilters` is gone. The filters that the builder took are logged at info, as batch scans log them, and a scan can still show them in its description, as for batch scans. The test "apply all the filters that the scan builder got in the plan of each micro-batch" checks it with `id + extra > 10`, which the builder cannot take. ########## 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) Review Comment: Fixed in becbceb: `batch` is a `Try[ColumnarBatchScan]`, so a failure of `pruneColumns` is kept, and the planner is asked once, as for `plan()`. "build and plan a batch scan once" checks a planner that fails to prune, asked twice for the batch. The native stack does the same in #59932, which documents the rule. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/BatchOnlyFilterPushDown.scala: ########## @@ -0,0 +1,27 @@ +/* + * 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.internal.connector + +import org.apache.spark.sql.connector.read.ScanBuilder + +/** + * An internal mix-in for a [[ScanBuilder]] whose filter pushdown only works for batch scans, such + * as one that pushes the filters down with the reader of a batch scan, which a streaming scan does + * not use and may not have. Spark does not push the filters of a streaming query down to it. + */ +trait BatchOnlyFilterPushDown extends ScanBuilder Review Comment: Done in becbceb: the opt-out is the absence of a new public capability, `TableCapability.STREAMING_FILTER_PUSHDOWN`, by which a table declares that its scan builders can take filters for a scan whose stream Spark creates. A third-party builder whose `pushFilters` needs the state of a batch read leaves it out, without the configuration. Python tables do not declare it, so `BatchOnlyFilterPushDown` is removed. It is separate from `STREAMING_COLUMN_PRUNING`, so that a table can declare one without the other. ########## 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( Review Comment: Done in becbceb: `StartupValue` keeps the expression that it stands for as a field which is not a child, so `current_user()`, `current_database()` and `current_catalog()` get different placeholders, with different `canonicalized`. Its data type and nullability come from that expression. The new test "keep the placeholders of the values that depend on when they are evaluated apart" checks them. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala: ########## @@ -228,3 +257,53 @@ class ColumnarStreamPartitionReaderFactory(reader: ColumnarStreamReader) reader.read(partition) } } + +object ColumnarProjection { + /** + * Returns `factory`, or a factory whose readers only return the columns at `projection` of the + * batches of the readers of `factory`, if any. + */ + def apply( + factory: PartitionReaderFactory, + projection: Option[Array[Int]]): PartitionReaderFactory = projection match { + case Some(ordinals) => new ColumnarProjectionReaderFactory(factory, ordinals) + case None => factory + } +} + +/** + * Reads only the columns at `ordinals` of the batches of the readers of `factory`, for a scan + * whose partitions return more columns than the scan reads. + */ +class ColumnarProjectionReaderFactory(factory: PartitionReaderFactory, ordinals: Array[Int]) + extends ColumnarOnlyPartitionReaderFactory { + override def createColumnarReader(partition: InputPartition): PartitionReader[ColumnarBatch] = { + new ColumnarProjectionReader(factory.createColumnarReader(partition), ordinals) + } +} + +private class ColumnarProjectionReader( + reader: PartitionReader[ColumnarBatch], + ordinals: Array[Int]) + extends PartitionReader[ColumnarBatch] { + + override def next(): Boolean = reader.next() + + override def get(): ColumnarBatch = new ProjectedColumnarBatch(reader.get(), ordinals) Review Comment: Fixed in becbceb, as in #59931 (6ab9036): `next()` builds the projected batch once, and `get()` returns it until the next call of `next()`. The test of a planner that does not prune checks that `reader.get() eq reader.get()`. -- 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]
