HyukjinKwon commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4465615373
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -261,6 +260,19 @@ class MicroBatchExecution(
}
import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+ // The columns and the filters that the query reads from each Data Source
V2 source, which the
+ // scan of the source gets if it supports them. They come from the plan of
the query optimized
+ // once as its micro-batches are, with their current time and their
configuration, which a
+ // restarted query takes from the offset log. The plan is only optimized
for a scan builder
+ // that supports a pushdown.
+ lazy val sourceReads = {
+ val batchConf = streamConf.clone()
+ initialLatestOffsetSeq.flatMap(_._2.metadataOpt).foreach { metadata =>
+ OffsetSeqMetadata.setSessionConf(metadata, batchConf, logChanges =
false)
+ }
+
StreamingV2ScanPushDown.reads(sparkSessionForStream.sessionState.optimizer,
batchConf,
+ withBatchTimestamp(planWithDedupKeys, batchTimestampMs = 0L))
Review Comment:
Fixed in 9ef62c1, for both modes. The plan optimized at the start no longer
has any value of the time: every expression with the pattern `CURRENT_LIKE`
(including `current_time()`, `now()` and the nanosecond variants) and the casts
of times to timestamps, which take the current date, become placeholders that
keep their children, which the optimizer does not take as deterministic, and no
filter that has one is pushed down. `ContinuousExecution` calls the same
`reads` now, so a reconfigured continuous query plans with its new time while
its scan has no filter on the time. The new tests "do not push down a filter on
a value that depends on when it is evaluated: ..." include `tod <
current_time()`, `CAST(tod AS TIMESTAMP_NTZ) < ...` and `CAST(tod AS TIMESTAMP)
< ...`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,241 @@
+/*
+ * 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.expressions.{AttributeReference,
AttributeSet, Expression, ExpressionSet, SubqueryExpression}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.connector.catalog.SupportsRead
+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
+import org.apache.spark.sql.execution.datasources.v2.PushDownUtils
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters
+import org.apache.spark.sql.types.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.
+ *
+ * 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])
+
+ /**
+ * 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, or any relation if the
optimizer fails, has no
+ * entry, so that its scan reads all its columns and gets no filter.
+ *
+ * @param optimizer the optimizer of the micro-batches of the query
+ * @param conf the configuration that the micro-batches of the query run with
+ * @param plan the plan of the query, with the current time of a micro-batch
as the micro-batches
+ * have it, which the optimizer does not fold, so that no filter
on it is pushed down
+ */
+ def reads(
+ optimizer: RuleExecutor[LogicalPlan],
+ conf: SQLConf,
+ plan: LogicalPlan): Map[StreamingRelationV2, Reads] = {
+ SQLConf.withExistingConf(conf) {
+ try {
+ val occurrences = mutable.HashMap.empty[StreamingRelationV2,
mutable.Buffer[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.getOrElseUpdate(relation, mutable.Buffer.empty) +=
Reads(
+ AttributeSet((project ++ filters).flatMap(_.references)),
+ filters.filter(f => f.deterministic &&
!SubqueryExpression.hasSubquery(f)))
+ case _ =>
+ node.children.foreach(visit)
+ }
+ visit(optimizer.execute(plan))
+ occurrences.map { case (relation, reads) =>
+ // The scan of a relation reads the rows of all its occurrences, so
it can only apply the
+ // filters that all of them apply.
+ val filters = reads.map(r => ExpressionSet(r.filters)).reduce {
(left, right) =>
+ ExpressionSet(left.filter(right.contains))
+ }
+ relation -> Reads(
+ AttributeSet.fromAttributeSets(reads.map(_.requiredColumns)),
filters.toSeq)
+ }.toMap
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Could not optimize the streaming query to push down
what it reads to " +
+ log"the scans of its sources, which read all their columns with no
filter.", e)
+ Map.empty
+ }
+ }
+ }
+
+ /**
+ * 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. Returns the
+ * scan, its stream, and the columns of `output` that the scan reads, in the
order of its read
+ * schema, unless these are `output` itself.
+ *
+ * A pruned scan must read the required columns and only columns of
`output`, each once, as
+ * `SupportsPushDownRequiredColumns` requires. A scan that does not is
replaced by a scan of all
+ * the columns, as without pruning, and its stream is stopped. Spark asks a
scan for its read
+ * schema once the stream is created, since a scan may only know then what
its stream reads.
+ *
+ * @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[S <: SparkDataStream](
+ table: SupportsRead,
+ options: CaseInsensitiveStringMap,
+ output: Seq[AttributeReference],
+ conf: SQLConf,
+ reads: => Option[Reads])(
+ createStream: Scan => S): (Scan, S, Option[Seq[AttributeReference]]) = {
+ 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)
+
+ def build(builder: ScanBuilder, columns: Seq[AttributeReference]): Scan = {
+ 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.
+ PushDownUtils.pushFilters(builder, filters, partitionFields = None)
+ }
+ builder match {
+ case pruning: SupportsPushDownRequiredColumns =>
+ pruning.pruneColumns(DataTypeUtils.fromAttributes(columns))
+ case _ =>
+ }
+ builder.build()
+ }
+ def buildFullScan(builder: ScanBuilder): (Scan, S,
Option[Seq[AttributeReference]]) = {
+ val scan = build(builder, output)
+ (scan, createStream(scan), None)
+ }
+
+ if (requiredOutput.length == output.length) {
+ buildFullScan(builder)
+ } else {
+ val scan = build(builder, requiredOutput)
+ val stream = createStream(scan)
+ val (readSchema, scanOutput) = try {
+ val readSchema = scan.readSchema()
+ (readSchema, prunedScanOutput(readSchema, output, requiredOutput))
+ } catch {
+ case e: Throwable =>
+ try stream.stop() catch {
+ case NonFatal(suppressed) => e.addSuppressed(suppressed)
+ }
+ throw e
+ }
+ scanOutput match {
+ case Right(scanOutput) =>
+ val readsOutput = scanOutput.map(_.exprId) == output.map(_.exprId)
+ (scan, stream, if (readsOutput) None else Some(scanOutput))
+ case Left(reason) =>
+ val columns = requiredOutput.map(_.name).mkString(", ")
+ logWarning(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.SCHEMA, readSchema.simpleString)} " +
+ log"${MDC(LogKeys.REASON, reason)}. It reads all the columns
instead.")
+ stream.stop()
+ buildFullScan(table.newScanBuilder(options))
+ }
+ }
+ }
+
+ private def supportsFilters(builder: ScanBuilder): Boolean = builder match {
Review Comment:
Fixed in 9ef62c1: `supportsFilters` excludes `PythonScanBuilder`, so a
streaming scan of a Python data source gets no filter, and since the builder
does not support column pruning either, the startup optimization does not run
for it: no `pushdownFiltersInPython` worker, `reader(schema)` call or
`partitions()` at stream start. The new test in
`PythonStreamingDataSourceSuite` filters the stream of a source that only has
`streamReader`, with `spark.sql.python.filterPushdown.enabled`; it fails with
`PySparkNotImplementedError: [NOT_IMPLEMENTED] reader is not implemented`
without the fix. The PR description and the migration guide are updated.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -1101,6 +1103,30 @@ private[sql] object DataSourceV2Strategy extends Logging
{
withFilter
}
}
+
+ /**
+ * Returns the output of the scan node of a streaming relation: the columns
that its scan reads.
+ * A scan built to read fewer columns than the relation outputs only gets
them when the
+ * projections and the filters over the relation, which the scan node is
planned with, read no
+ * other column of the relation (see `StreamingV2ScanPushDown`).
+ */
+ private def streamingScanOutput(
Review Comment:
Done in 9ef62c1. `StreamingDataSourceV2ScanRelation.pushedFilters` records
the filters that the scan got (remapped by `newInstance` as `prunedOutput` is),
and `DataSourceV2Strategy.streamingScanOutput` fails with an internal error
that names `spark.sql.streaming.v2FilterPushdown.enabled` when the filters over
the relation do not contain them (`ExpressionSet`). New test: "fail to plan a
micro-batch that does not apply a filter that its scan got". With the startup
optimization of the previous commit, it turns the `now()` case into such an
error at the first micro-batch.
To make that check hold in every micro-batch of a valid query, the startup
optimization no longer infers filters from constraints
(`InferFiltersFromConstraints` is excluded there). A filter inferred from
another relation is not always over the relation in a micro-batch: in a
stream-stream join `s.join(o.where("oid < 3"), id = oid)`, `id < 3` is inferred
for `s`, but not in a micro-batch where `o` has no new data, since
`PropagateEmptyRelation` removes the filter with its empty relation (a cached
static side loses its constraints the same way). The pushed filters are now the
filters of the query over the relation, which the micro-batches apply too, and
they still infer their own. New test: "push down no filter that the optimizer
infers from the filters of another source", which runs such micro-batches.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,241 @@
+/*
+ * 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.expressions.{AttributeReference,
AttributeSet, Expression, ExpressionSet, SubqueryExpression}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.connector.catalog.SupportsRead
+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
+import org.apache.spark.sql.execution.datasources.v2.PushDownUtils
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters
+import org.apache.spark.sql.types.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.
+ *
+ * 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])
+
+ /**
+ * 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, or any relation if the
optimizer fails, has no
+ * entry, so that its scan reads all its columns and gets no filter.
+ *
+ * @param optimizer the optimizer of the micro-batches of the query
+ * @param conf the configuration that the micro-batches of the query run with
+ * @param plan the plan of the query, with the current time of a micro-batch
as the micro-batches
+ * have it, which the optimizer does not fold, so that no filter
on it is pushed down
+ */
+ def reads(
+ optimizer: RuleExecutor[LogicalPlan],
+ conf: SQLConf,
+ plan: LogicalPlan): Map[StreamingRelationV2, Reads] = {
+ SQLConf.withExistingConf(conf) {
+ try {
+ val occurrences = mutable.HashMap.empty[StreamingRelationV2,
mutable.Buffer[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.getOrElseUpdate(relation, mutable.Buffer.empty) +=
Reads(
+ AttributeSet((project ++ filters).flatMap(_.references)),
+ filters.filter(f => f.deterministic &&
!SubqueryExpression.hasSubquery(f)))
+ case _ =>
+ node.children.foreach(visit)
+ }
+ visit(optimizer.execute(plan))
Review Comment:
In 9ef62c1, the relations of the batch scans are placeholders in the plan
optimized at the start (`DataSourceV2Relation`, non-streaming `LogicalRelation`
and `HiveTableRelation`, also in subqueries), so `V2ScanRelationPushDown`,
`PruneFileSourcePartitions`, `PruneHiveTablePartitions`, `PartitionPruning` and
`InjectRuntimeFilter` have nothing to build, list or plan for them: no batch
scan, Python worker or partition listing at the start. The new test "do not
build the scans of the batch relations of a streaming query when it starts"
checks it with a batch table that counts its scan builders.
I kept the other rules: the planner now checks that each micro-batch reads
no other column and applies the pushed filters, so the startup optimization has
to end up with the same projections and filters over the relation as the
optimizer of the micro-batches. A subset of the rules, e.g. without the
subquery rewrites, `InferFiltersFromGenerate` or the extraction of Python UDFs
in the later batches, would leave other projections and filters over the
relation. The pass runs once per start, only for a source whose scan builder
supports one of the pushdowns, which excludes the built-in sources, and every
micro-batch runs the same optimizer anyway.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala:
##########
@@ -123,39 +124,59 @@ class ColumnarScanBuilder(
override def pruneColumns(requiredSchema: StructType): Unit = {
Review Comment:
Fixed in 9ef62c1: the read schema of `ColumnarScan` is now the required
columns for streams too, and its stream drops the other columns from the
batches of the stream reader, which reads the full schema until #59931 passes
the columns to read to it. So the scan node of a micro-batch only outputs the
required columns, and the startup optimization is not wasted.
`ColumnarDataSourceSuite` checks that the scan of `select("name")` reads and
outputs only `name`.
--
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]
