HyukjinKwon commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4479853546
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -270,15 +271,17 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
if r.startOffset.isDefined && r.endOffset.isDefined =>
val microBatchStream = r.stream.asInstanceOf[MicroBatchStream]
+ val output = DataSourceV2Strategy.streamingScanOutput(r, p, f)
val scanExec = MicroBatchScanExec(
- r.output, r.scan, microBatchStream, r.startOffset.get, r.endOffset.get)
+ output, r.scan, microBatchStream, r.startOffset.get, r.endOffset.get)
// Add a Project here to make sure we produce unsafe rows.
DataSourceV2Strategy.withProjectAndFilter(p, f, scanExec,
!scanExec.supportsColumnar) :: Nil
case PhysicalOperation(p, f, r: StreamingDataSourceV2ScanRelation)
if r.startOffset.isDefined && r.endOffset.isEmpty =>
+ val output = DataSourceV2Strategy.streamingScanOutput(r, p, f)
Review Comment:
Done in 89114e3: the output of the scan node is computed after the check of
the real-time mode, so a stream without `SupportsRealTimeMode` gets
`STREAMING_REAL_TIME_MODE.INPUT_STREAM_NOT_SUPPORTED`.
---
<!-- GITHUB_MCP_FOOTER: This attribution is automatically appended by GitHub
MCP. -->
_This comment was generated with [GitHub MCP](http://go/mcps)._
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java:
##########
@@ -170,5 +170,44 @@ public enum TableCapability {
*
* @since 4.4.0
*/
- EXTERNAL_LOOKUP
+ EXTERNAL_LOOKUP,
+
+ /**
+ * Signals that the scans of the streaming reads of the table may only read
the columns that the
+ * streaming query reads, which Spark asks a scan builder that implements
+ * {@link
org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns} for, as it
does
+ * for the scans of batch reads.
+ * <p>
+ * Spark builds the scan of a streaming read once, when the query starts,
from the plan of the
+ * query optimized then, and creates the micro-batch or continuous stream of
the read from that
+ * scan. By returning this capability, a table declares that the partition
readers of such a
+ * stream return the columns of {@link
org.apache.spark.sql.connector.read.Scan#readSchema()},
+ * as those of the batches of the scan do. Spark asks the scan for its read
schema before it
+ * creates the stream. Without this capability, the scans of streaming reads
read all the
+ * columns of the table.
+ *
+ * @since 4.4.0
+ */
+ STREAMING_COLUMN_PRUNING,
+
+ /**
+ * Signals that the scans of the streaming reads of the table may get the
filters that the
+ * streaming query applies first to all the rows that it reads from the
table, before any other
+ * filter, which Spark pushes to a scan builder that implements
+ * {@link org.apache.spark.sql.connector.read.SupportsPushDownFilters} or
+ * {@link org.apache.spark.sql.connector.read.SupportsPushDownV2Filters}, as
it does for the
+ * scans of batch reads.
+ * <p>
+ * Spark builds the scan of a streaming read once, when the query starts,
from the plan of the
+ * query optimized then, and creates the micro-batch or continuous stream of
the read from that
+ * scan. By returning this capability, a table declares that its scan
builders can take filters
+ * for such a scan, as for the scan of a batch read. The stream may then
drop the rows that fail
+ * the filters that the builder took: the query applies all the filters that
the builder got to
+ * the rows that it reads from the table, and Spark evaluates all the
filters of the query after
+ * the scan. The progress of the query only counts the rows that the stream
returns. Without this
+ * capability, the scans of streaming reads get no filter.
+ *
+ * @since 4.4.0
+ */
+ STREAMING_FILTER_PUSHDOWN
Review Comment:
Done: 9cbc7b8 moves the filter pushdown to SPARK-60926, in the draft #60129,
which is based on this PR. This PR only prunes the columns now, and its title
and description say so. The comments on the filters from this round are listed
as open items there.
---
<!-- GITHUB_MCP_FOOTER: This attribution is automatically appended by GitHub
MCP. -->
_This comment was generated with [GitHub MCP](http://go/mcps)._
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,475 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming.runtime
+
+import scala.collection.mutable
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation,
ResolvedInlineTable}
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute,
AttributeReference, AttributeSet, Cast, Expression, SubqueryExpression,
Unevaluable}
+import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime,
InferFiltersFromConstraints, ReplaceExpressions, SpecialDatetimeValues}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode,
LogicalPlan, Statistics}
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST, CURRENT_LIKE,
RUNTIME_REPLACEABLE}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.classic.SparkSession
+import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability}
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder,
SupportsPushDownRequiredColumns}
+import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
LogicalRelation}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata,
OffsetSeqMetadataBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, StringType, StructType,
TimestampLTZNanosType, TimestampType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Pushes the top-level columns and the filters that the optimized plan of a
streaming query reads
+ * from each Data Source V2 streaming source down to the scan of the source, as
+ * `V2ScanRelationPushDown` does for the scans of batch queries, if the table
of the source has the
+ * capability `STREAMING_COLUMN_PRUNING` or `STREAMING_FILTER_PUSHDOWN`.
+ *
+ * The stream of a source is created from its scan, and has to exist before
any micro-batch is
+ * planned, so the scan is built when the query starts. The query is optimized
once at that point,
+ * by the optimizer of its micro-batches and with their configuration, and the
scan of each source
+ * gets what the planner takes from the optimized plan for the relation of the
source (see
+ * `DataSourceV2Strategy`): the projections and the filters over each
occurrence of the relation.
+ *
+ * The plan of each micro-batch is the one without these pushdowns, but for a
filter of the rows
+ * of the relation with the filters that the scan builder got, which the query
applies anyway, so
+ * that the scan cannot drop a row that the plan keeps (see
`withPushedFilters`). The optimizer
+ * keeps deciding which columns every operator gets, and in which order,
including the stateful
+ * ones, and Spark still evaluates all the filters over the relation. Only the
scan node of the
+ * physical plan outputs the columns that the scan reads, and the planner
checks that the
+ * micro-batch reads no other column of the relation.
+ *
+ * The other operations that the scans of batch queries can apply, such as
aggregates and limits,
+ * apply to the data of all the micro-batches of a streaming query, which the
scan of a micro-batch
+ * does not read, so they are not pushed down.
+ */
+private[sql] object StreamingV2ScanPushDown extends Logging {
+
+ /**
+ * What the optimized plan of a streaming query reads from a streaming
relation.
+ *
+ * @param requiredColumns the columns of the relation that the plan reads
+ * @param filters the filters that the scan of the relation may get, in the
order in which the
+ * plan evaluates them: those that every occurrence of the
relation applies first,
+ * see `pushableFilters`
+ */
+ case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression])
+
+ /**
+ * The scan of a streaming source, and its stream.
+ *
+ * @param prunedOutput the columns of the relation that the scan reads, in
the order of its read
+ * schema, unless these are all its columns
+ * @param pushedFilters the filters that the scan builder got, which the
scan may apply in any
+ * form, so the plan of each micro-batch applies them
over the relation
+ */
+ case class StreamingScan(
+ scan: Scan,
+ stream: SparkDataStream,
+ prunedOutput: Option[Seq[AttributeReference]],
+ pushedFilters: Seq[Expression]) {
+
+ /** Returns the scan relation of the source whose relation is `relation`.
*/
+ def toRelation(
+ relation: StreamingDataSourceV2Relation):
StreamingDataSourceV2ScanRelation = {
+ StreamingDataSourceV2ScanRelation(relation, scan, relation.output,
stream,
+ prunedOutput = prunedOutput, pushedFilters = pushedFilters)
+ }
+ }
+
+ /**
+ * Returns what the optimized `plan` reads from each of its streaming
relations of Data Source
+ * V2: what the planner takes from the projections and the filters over each
occurrence of the
+ * relation, or all its columns and no filter where nothing is projected or
filtered over it. A
+ * relation that the optimized plan does not have has no entry, so that its
scan reads all its
+ * columns and gets no filter, as no relation has if the optimizer fails, or
if the optimized
+ * plan has a streaming relation that `plan` does not have.
+ *
+ * The plan is optimized as the micro-batches are: with the optimizer and
the configuration of
+ * the stream, the values of the configurations that the offset log of a
restarted query pins, and
+ * without the cost-based reordering of joins, which `StreamExecution`
disables for its queries.
+ * The plan does not depend on when it is optimized, and the scan of a
relation only gets what the
+ * query reads from that relation:
+ * - The expressions whose values depend on when they are evaluated or on
the state of the
+ * session, such as the current time, the casts of special strings such
as 'today' or 'now'
+ * to dates and timestamps, the casts of strings with only a time to
timestamps, or the
+ * current database, which each micro-batch gets the values of, become
placeholders that the
+ * optimizer does not take as deterministic, as the current time of a
micro-batch, so neither
+ * its filters nor its columns depend on them, and no filter on them is
pushed down.
+ * - The filters that the optimizer infers from the constraints of the
plan, such as those of the
+ * filters over another relation joined with the relation, are not
inferred, so that the scan
+ * only gets the filters that the query applies to the relation itself.
+ * - The relations of batch scans become placeholders, so that their scans
are neither built
+ * nor planned: they do not change what the plan reads from its streaming
relations.
+ *
+ * @param session the session of the stream
+ * @param offsetLogMetadata the metadata of the last entry of the offset log
of the query, if any
+ */
+ def reads(
+ session: SparkSession,
+ plan: LogicalPlan,
+ offsetLogMetadata: Option[OffsetSeqMetadataBase]):
Map[StreamingRelationV2, Reads] = {
+ val conf = session.sessionState.conf.clone()
+ offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf,
logChanges = false))
+ conf.setConf(SQLConf.CBO_ENABLED, false)
+ conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key,
+ (conf.optimizerExcludedRules.toSeq :+
InferFiltersFromConstraints.ruleName).mkString(","))
+ session.withActive {
+ SQLConf.withExistingConf(conf) {
+ try {
+ val optimized =
session.sessionState.optimizer.execute(forStartup(plan))
+ // The planner of a micro-batch gets the values of the expressions
that depend on when
+ // they are evaluated, which are deterministic.
+ val planned = optimized.transformUpWithSubqueries {
+ case node => node.transformExpressions {
+ case value: StartupValue => value.copy(planned = true)
+ }
+ }
+ val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2,
Reads)]
+ // Visits the nodes as the planner plans them: the projections and
the filters over a
+ // relation are planned with its scan, and the other nodes on their
own.
+ def visit(node: LogicalPlan): Unit = node match {
+ case PhysicalOperation(project, filters, relation:
StreamingRelationV2) =>
+ occurrences += relation -> Reads(
+ AttributeSet((project ++ filters).flatMap(_.references)),
+ pushableFilters(filters))
+ case _ =>
+ node.children.foreach(visit)
+ }
+ visit(planned)
+ val relations = plan.collect { case relation: StreamingRelationV2 =>
relation }.toSet
+ val unknown = occurrences.map(_._1).filterNot(relations.contains)
+ if (unknown.nonEmpty) {
+ // The optimizer does not copy streaming relations:
`DeduplicateRelations` keeps
+ // streaming plans as they are, so all the references to a common
table expression,
+ // for example, have the same relation. A copy would share the
scan of its relation in
+ // the micro-batches, so the scan would have to read what the copy
reads too.
+ logWarning(log"The optimized streaming query reads the streaming
relations " +
+ log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(",
"))} that its " +
+ log"plan does not have, so the scans of its sources read all
their columns with " +
+ log"no filter.")
+ Map.empty
+ } else {
+ combine(occurrences.toSeq)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Could not optimize the streaming query to push down
what it reads " +
+ log"to the scans of its sources, which read all their columns
with no filter.", e)
+ Map.empty
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns what the occurrences of each relation read, combined: the scan of
a relation reads the
+ * rows of all its occurrences, so it reads the columns that any of them
reads, and can only
+ * apply the filters that all of them apply. Each occurrence evaluates the
pushed filters before
+ * its own, so the pushed filters are those that all the occurrences
evaluate first, in the same
+ * order: the longest common prefix of their filters, see `pushableFilters`.
+ */
+ private def combine(
+ occurrences: Seq[(StreamingRelationV2, Reads)]):
Map[StreamingRelationV2, Reads] = {
+ occurrences.groupBy(_._1).map { case (relation, reads) =>
+ val filters = reads.map(_._2.filters).reduce { (left, right) =>
+ left.zip(right).takeWhile { case (l, r) => l.semanticEquals(r)
}.map(_._1)
+ }
+ relation ->
Reads(AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters)
+ }
+ }
+
+ /**
+ * Returns the filters that the scan of a relation may get of `filters`, the
filters over an
+ * occurrence of the relation in the order in which the plan evaluates them:
the deterministic
+ * ones without subqueries or placeholders, up to the first filter that is
not, which is not
+ * pushed down, and neither are those after it. The plan of each micro-batch
evaluates the pushed
+ * filters first, over the relation (see `withPushedFilters`), so a pushed
filter does not see a
+ * row that a filter written before it drops: it could fail on such a row,
as a filter with a
+ * cast or a division does under ANSI, or with `raise_error`.
+ */
+ private def pushableFilters(filters: Seq[Expression]): Seq[Expression] = {
+ filters.takeWhile { filter =>
+ filter.deterministic && !SubqueryExpression.hasSubquery(filter) &&
+ !filter.exists(_.isInstanceOf[StartupValue])
+ }
+ }
+
+ /**
+ * Returns `plan` for the optimization when the query starts, see `reads`,
with its subqueries.
+ * The expressions that are replaced keep their children, so that the plan
still reads the same
+ * columns. Visible for tests.
+ */
+ private[sql] def forStartup(plan: LogicalPlan): LogicalPlan =
plan.transformUpWithSubqueries {
+ case relation: DataSourceV2Relation => BatchRelation(relation.output)
+ case relation: LogicalRelation if !relation.isStreaming =>
BatchRelation(relation.output)
+ case relation: HiveTableRelation => BatchRelation(relation.output)
+ // The rows of an inline table may have values of the current time, which
the optimizer
+ // evaluates.
+ case table: ResolvedInlineTable => BatchRelation(table.output)
+ case node =>
+ // The expressions that ReplaceExpressions replaces with others when the
optimization starts
+ // are replaced first, as the optimizer does, so that the casts that
they stand for, such as
+ // the one of to_date('today'), are replaced below.
+ val replaced = if (node.containsPattern(RUNTIME_REPLACEABLE)) {
+ node.mapExpressions(ReplaceExpressions.replace)
+ } else {
+ node
+ }
+ replaced.transformExpressionsUpWithPruning(
+ bits => bits.containsPattern(CURRENT_LIKE) ||
bits.containsPattern(CAST)) {
+ // The children of an expression are replaced before it, as are the
subqueries, so an
+ // expression that still has the pattern CURRENT_LIKE has it itself:
its value is the
+ // current time, or comes from the state of the session, as the
current database does.
+ case e if e.containsPattern(CURRENT_LIKE) => StartupValue(e,
e.children)
+ // So does a cast that ComputeCurrentTime or SpecialDatetimeValues
rewrites: a cast of a
+ // time to a timestamp, which takes the current date, or a cast of a
special string, such
+ // as 'today' or 'now', to a date or a timestamp. These rules only
rewrite casts besides
+ // the expressions with the pattern CURRENT_LIKE. A cast of a string
with only a time,
+ // such as '12:00:00', to a timestamp takes the current date too,
which the optimizer
+ // folds in.
+ case cast: Cast if ComputeCurrentTime.applyForExpression(cast) != cast
||
+ SpecialDatetimeValues.applyForExpression(cast) != cast ||
castsTimeOnly(cast) =>
+ StartupValue(cast, cast.children)
+ }
+ }
+
+ /**
+ * Returns whether `cast` casts a foldable string with only a time to a
timestamp with local time
+ * zone, of any precision, which takes the date of when it is evaluated. A
cast to a timestamp
+ * without time zone or to a date does not take the current date: it rejects
such a string.
+ */
+ private def castsTimeOnly(cast: Cast): Boolean = {
+ val toTimestampLTZ = cast.dataType match {
+ case TimestampType | _: TimestampLTZNanosType => true
+ case _ => false
+ }
+ toTimestampLTZ && cast.child.dataType.isInstanceOf[StringType] &&
cast.child.foldable &&
+ Try(cast.child.eval()).toOption.exists {
+ case string: UTF8String =>
DateTimeUtils.parseTimestampString(string)._3
+ case _ => false
+ }
+ }
+
+ /**
+ * Builds the scan of a streaming source, and creates its stream with
`createStream`. If the
+ * table of the source has the capability `STREAMING_COLUMN_PRUNING` or
+ * `STREAMING_FILTER_PUSHDOWN`, its scan builder supports the pushdown, and
`conf` enables it,
+ * the scan only reads the columns of `output` that the query reads, or gets
the filters that
+ * the query applies to all the rows that it reads from the source (see
`reads`). `reads` is only
+ * evaluated for such a scan. Otherwise, the scan reads all the columns of
`output` with no
+ * filter, as the scans of streaming queries did before these pushdowns.
+ *
+ * A pruned scan must read the required columns and only columns of
`output`, each once, as
+ * `SupportsPushDownRequiredColumns` requires. Spark asks a scan for its
read schema before it
+ * creates its stream, as for a batch. If the scan builder, the scan or its
stream fails with the
+ * pushdowns, or the read schema of a pruned scan cannot be read or does not
follow
+ * `SupportsPushDownRequiredColumns`, a scan of all the columns with no
filter is built instead,
+ * from a new scan builder, as without these pushdowns.
+ *
+ * @param output the output of the relation of the source, with the metadata
columns that the
+ * query reads, which the analyzer added to it, so that
connectors implementing
+ * `SupportsMetadataColumns` include them in `readSchema()`
+ * @param conf the configuration of the query
+ * @param reads what the query reads from the relation, or None to read all
its columns
+ */
+ def buildScan(
+ table: SupportsRead,
+ options: CaseInsensitiveStringMap,
+ output: Seq[AttributeReference],
+ conf: SQLConf,
+ reads: => Option[Reads])(
+ createStream: Scan => SparkDataStream): StreamingScan = {
+ val builder = table.newScanBuilder(options)
+ val capabilities = table.capabilities()
+ val prunesColumns = conf.streamingV2ColumnPruningEnabled &&
+ capabilities.contains(TableCapability.STREAMING_COLUMN_PRUNING) &&
+ builder.isInstanceOf[SupportsPushDownRequiredColumns]
+ val pushesFilters = conf.streamingV2FilterPushdownEnabled &&
+ capabilities.contains(TableCapability.STREAMING_FILTER_PUSHDOWN) &&
+ PushDownUtils.supportsFilters(builder)
+ val known = if (prunesColumns || pushesFilters) reads else None
+ val requiredOutput = known.filter(_ => prunesColumns)
+ .map(r => output.filter(r.requiredColumns.contains))
+ .getOrElse(output)
+ // The filters that the scan builder gets in some form, which each
micro-batch applies first:
+ // the others, such as a filter with a UDF that cannot be translated to a
filter of the source,
+ // are evaluated after them, as for the scans of batch queries, rather
than on all the rows of
+ // the source.
+ val filters = known.filter(_ => pushesFilters)
+ .map { r =>
+ PushDownUtils.filtersOfScanBuilder(builder,
+ DataSourceStrategy.normalizeExprs(r.filters, output))
+ }
+ .getOrElse(Nil)
+
+ // Builds the scan of `columns` with `builder`, once it pushed `filters`
down to it.
+ def build(
+ builder: ScanBuilder,
+ columns: Seq[AttributeReference],
+ filters: Seq[Expression]): Scan = {
+ if (filters.nonEmpty) {
+ // The plan of each micro-batch applies these filters to the rows of
the source, see
+ // `withPushedFilters`, so the scan may apply any of them, in any form.
+ val (pushed, _) = PushDownUtils.pushFilters(builder, filters,
partitionFields = None)
+ val taken = pushed.fold(_.mkString(", "), _.mkString(", "))
+ logInfo(log"The scan builder of the streaming table " +
+ log"${MDC(LogKeys.STREAMING_TABLE, table)} took the filters " +
+ log"${MDC(LogKeys.PUSHED_FILTERS, taken)}.")
+ }
+ builder match {
+ case pruning: SupportsPushDownRequiredColumns =>
+ pruning.pruneColumns(DataTypeUtils.fromAttributes(columns))
+ case _ =>
+ }
+ builder.build()
+ }
+ // The scan of all the columns with no filter, as without these pushdowns.
+ def scanOfAllColumns(builder: ScanBuilder): StreamingScan = {
+ val scan = build(builder, output, Nil)
+ StreamingScan(scan, createStream(scan), None, Nil)
+ }
+
+ if (requiredOutput.length == output.length && filters.isEmpty) {
+ scanOfAllColumns(builder)
+ } else {
+ // Logs why the scan with the pushdowns could not be built, and builds
the scan of all the
+ // columns with no filter instead.
+ def fallBack(reason: String, cause: Option[Throwable]): StreamingScan = {
Review Comment:
I kept it. It was asked for in
https://github.com/apache/spark/pull/59943#discussion_r4463353287 and
https://github.com/apache/spark/pull/59943#discussion_r4471743202, so that a
source that declares the capability but whose pruned scan fails, such as a
`readSchema()` that throws for a schema that it cannot honor, keeps reading all
its columns, with a warning, rather than failing the query when it starts. It
only applies to the tables that declare `STREAMING_COLUMN_PRUNING`, and costs
another scan builder when it happens. I can drop it if you both prefer the
behavior of batch scans.
---
<!-- GITHUB_MCP_FOOTER: This attribution is automatically appended by GitHub
MCP. -->
_This comment was generated with [GitHub MCP](http://go/mcps)._
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,475 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming.runtime
+
+import scala.collection.mutable
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation,
ResolvedInlineTable}
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute,
AttributeReference, AttributeSet, Cast, Expression, SubqueryExpression,
Unevaluable}
+import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime,
InferFiltersFromConstraints, ReplaceExpressions, SpecialDatetimeValues}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode,
LogicalPlan, Statistics}
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST, CURRENT_LIKE,
RUNTIME_REPLACEABLE}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.classic.SparkSession
+import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability}
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder,
SupportsPushDownRequiredColumns}
+import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
LogicalRelation}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata,
OffsetSeqMetadataBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, StringType, StructType,
TimestampLTZNanosType, TimestampType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Pushes the top-level columns and the filters that the optimized plan of a
streaming query reads
+ * from each Data Source V2 streaming source down to the scan of the source, as
+ * `V2ScanRelationPushDown` does for the scans of batch queries, if the table
of the source has the
+ * capability `STREAMING_COLUMN_PRUNING` or `STREAMING_FILTER_PUSHDOWN`.
+ *
+ * The stream of a source is created from its scan, and has to exist before
any micro-batch is
+ * planned, so the scan is built when the query starts. The query is optimized
once at that point,
+ * by the optimizer of its micro-batches and with their configuration, and the
scan of each source
+ * gets what the planner takes from the optimized plan for the relation of the
source (see
+ * `DataSourceV2Strategy`): the projections and the filters over each
occurrence of the relation.
+ *
+ * The plan of each micro-batch is the one without these pushdowns, but for a
filter of the rows
+ * of the relation with the filters that the scan builder got, which the query
applies anyway, so
+ * that the scan cannot drop a row that the plan keeps (see
`withPushedFilters`). The optimizer
+ * keeps deciding which columns every operator gets, and in which order,
including the stateful
+ * ones, and Spark still evaluates all the filters over the relation. Only the
scan node of the
+ * physical plan outputs the columns that the scan reads, and the planner
checks that the
+ * micro-batch reads no other column of the relation.
+ *
+ * The other operations that the scans of batch queries can apply, such as
aggregates and limits,
+ * apply to the data of all the micro-batches of a streaming query, which the
scan of a micro-batch
+ * does not read, so they are not pushed down.
+ */
+private[sql] object StreamingV2ScanPushDown extends Logging {
+
+ /**
+ * What the optimized plan of a streaming query reads from a streaming
relation.
+ *
+ * @param requiredColumns the columns of the relation that the plan reads
+ * @param filters the filters that the scan of the relation may get, in the
order in which the
+ * plan evaluates them: those that every occurrence of the
relation applies first,
+ * see `pushableFilters`
+ */
+ case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression])
+
+ /**
+ * The scan of a streaming source, and its stream.
+ *
+ * @param prunedOutput the columns of the relation that the scan reads, in
the order of its read
+ * schema, unless these are all its columns
+ * @param pushedFilters the filters that the scan builder got, which the
scan may apply in any
+ * form, so the plan of each micro-batch applies them
over the relation
+ */
+ case class StreamingScan(
+ scan: Scan,
+ stream: SparkDataStream,
+ prunedOutput: Option[Seq[AttributeReference]],
+ pushedFilters: Seq[Expression]) {
+
+ /** Returns the scan relation of the source whose relation is `relation`.
*/
+ def toRelation(
+ relation: StreamingDataSourceV2Relation):
StreamingDataSourceV2ScanRelation = {
+ StreamingDataSourceV2ScanRelation(relation, scan, relation.output,
stream,
+ prunedOutput = prunedOutput, pushedFilters = pushedFilters)
+ }
+ }
+
+ /**
+ * Returns what the optimized `plan` reads from each of its streaming
relations of Data Source
+ * V2: what the planner takes from the projections and the filters over each
occurrence of the
+ * relation, or all its columns and no filter where nothing is projected or
filtered over it. A
+ * relation that the optimized plan does not have has no entry, so that its
scan reads all its
+ * columns and gets no filter, as no relation has if the optimizer fails, or
if the optimized
+ * plan has a streaming relation that `plan` does not have.
+ *
+ * The plan is optimized as the micro-batches are: with the optimizer and
the configuration of
+ * the stream, the values of the configurations that the offset log of a
restarted query pins, and
+ * without the cost-based reordering of joins, which `StreamExecution`
disables for its queries.
+ * The plan does not depend on when it is optimized, and the scan of a
relation only gets what the
+ * query reads from that relation:
+ * - The expressions whose values depend on when they are evaluated or on
the state of the
+ * session, such as the current time, the casts of special strings such
as 'today' or 'now'
+ * to dates and timestamps, the casts of strings with only a time to
timestamps, or the
+ * current database, which each micro-batch gets the values of, become
placeholders that the
+ * optimizer does not take as deterministic, as the current time of a
micro-batch, so neither
+ * its filters nor its columns depend on them, and no filter on them is
pushed down.
+ * - The filters that the optimizer infers from the constraints of the
plan, such as those of the
+ * filters over another relation joined with the relation, are not
inferred, so that the scan
+ * only gets the filters that the query applies to the relation itself.
+ * - The relations of batch scans become placeholders, so that their scans
are neither built
+ * nor planned: they do not change what the plan reads from its streaming
relations.
+ *
+ * @param session the session of the stream
+ * @param offsetLogMetadata the metadata of the last entry of the offset log
of the query, if any
+ */
+ def reads(
+ session: SparkSession,
+ plan: LogicalPlan,
+ offsetLogMetadata: Option[OffsetSeqMetadataBase]):
Map[StreamingRelationV2, Reads] = {
+ val conf = session.sessionState.conf.clone()
+ offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf,
logChanges = false))
+ conf.setConf(SQLConf.CBO_ENABLED, false)
+ conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key,
+ (conf.optimizerExcludedRules.toSeq :+
InferFiltersFromConstraints.ruleName).mkString(","))
+ session.withActive {
+ SQLConf.withExistingConf(conf) {
+ try {
+ val optimized =
session.sessionState.optimizer.execute(forStartup(plan))
+ // The planner of a micro-batch gets the values of the expressions
that depend on when
+ // they are evaluated, which are deterministic.
+ val planned = optimized.transformUpWithSubqueries {
+ case node => node.transformExpressions {
+ case value: StartupValue => value.copy(planned = true)
+ }
+ }
+ val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2,
Reads)]
+ // Visits the nodes as the planner plans them: the projections and
the filters over a
+ // relation are planned with its scan, and the other nodes on their
own.
+ def visit(node: LogicalPlan): Unit = node match {
+ case PhysicalOperation(project, filters, relation:
StreamingRelationV2) =>
+ occurrences += relation -> Reads(
+ AttributeSet((project ++ filters).flatMap(_.references)),
+ pushableFilters(filters))
+ case _ =>
+ node.children.foreach(visit)
+ }
+ visit(planned)
+ val relations = plan.collect { case relation: StreamingRelationV2 =>
relation }.toSet
+ val unknown = occurrences.map(_._1).filterNot(relations.contains)
+ if (unknown.nonEmpty) {
+ // The optimizer does not copy streaming relations:
`DeduplicateRelations` keeps
+ // streaming plans as they are, so all the references to a common
table expression,
+ // for example, have the same relation. A copy would share the
scan of its relation in
+ // the micro-batches, so the scan would have to read what the copy
reads too.
+ logWarning(log"The optimized streaming query reads the streaming
relations " +
+ log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(",
"))} that its " +
+ log"plan does not have, so the scans of its sources read all
their columns with " +
+ log"no filter.")
+ Map.empty
+ } else {
+ combine(occurrences.toSeq)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Could not optimize the streaming query to push down
what it reads " +
+ log"to the scans of its sources, which read all their columns
with no filter.", e)
+ Map.empty
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns what the occurrences of each relation read, combined: the scan of
a relation reads the
+ * rows of all its occurrences, so it reads the columns that any of them
reads, and can only
+ * apply the filters that all of them apply. Each occurrence evaluates the
pushed filters before
+ * its own, so the pushed filters are those that all the occurrences
evaluate first, in the same
+ * order: the longest common prefix of their filters, see `pushableFilters`.
+ */
+ private def combine(
+ occurrences: Seq[(StreamingRelationV2, Reads)]):
Map[StreamingRelationV2, Reads] = {
+ occurrences.groupBy(_._1).map { case (relation, reads) =>
+ val filters = reads.map(_._2.filters).reduce { (left, right) =>
+ left.zip(right).takeWhile { case (l, r) => l.semanticEquals(r)
}.map(_._1)
+ }
+ relation ->
Reads(AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)), filters)
+ }
+ }
+
+ /**
+ * Returns the filters that the scan of a relation may get of `filters`, the
filters over an
+ * occurrence of the relation in the order in which the plan evaluates them:
the deterministic
+ * ones without subqueries or placeholders, up to the first filter that is
not, which is not
+ * pushed down, and neither are those after it. The plan of each micro-batch
evaluates the pushed
+ * filters first, over the relation (see `withPushedFilters`), so a pushed
filter does not see a
+ * row that a filter written before it drops: it could fail on such a row,
as a filter with a
+ * cast or a division does under ANSI, or with `raise_error`.
+ */
+ private def pushableFilters(filters: Seq[Expression]): Seq[Expression] = {
+ filters.takeWhile { filter =>
+ filter.deterministic && !SubqueryExpression.hasSubquery(filter) &&
+ !filter.exists(_.isInstanceOf[StartupValue])
+ }
+ }
+
+ /**
+ * Returns `plan` for the optimization when the query starts, see `reads`,
with its subqueries.
+ * The expressions that are replaced keep their children, so that the plan
still reads the same
+ * columns. Visible for tests.
+ */
+ private[sql] def forStartup(plan: LogicalPlan): LogicalPlan =
plan.transformUpWithSubqueries {
Review Comment:
With the filter pushdown moved to SPARK-60926 (#60129), this PR adds no
filter to the plan of a micro-batch: a value that `forStartup` misses can only
change the columns of a scan, and the planner then fails the micro-batch with
the internal error that names `spark.sql.streaming.v2ColumnPruning.enabled`,
rather than dropping rows. 89114e3 also replaces every cast of a string to a
timestamp, the shape that the last rounds found. Comparing the pushed filters
with the plan of the first micro-batch is listed in #60129.
---
<!-- GITHUB_MCP_FOOTER: This attribution is automatically appended by GitHub
MCP. -->
_This comment was generated with [GitHub MCP](http://go/mcps)._
--
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]
