dongjoon-hyun commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4471743193
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -1142,6 +1145,78 @@ private[sql] object DataSourceV2Strategy extends Logging
{
withFilter
}
}
+
+ /**
+ * Returns the output of the scan node of a streaming relation: the columns
that its scan reads.
+ * The projections and the filters over the relation, which the scan node is
planned with, must
+ * still imply the filters that the scan got, and read no column of the
relation that the scan
+ * does not read, if it was built to read fewer columns than the relation
outputs (see
+ * `StreamingV2ScanPushDown`).
+ */
+ private def streamingScanOutput(
+ relation: StreamingDataSourceV2ScanRelation,
+ project: Seq[NamedExpression],
+ filters: Seq[Expression]): Seq[AttributeReference] = {
+ val missingFilters = relation.pushedFilters.filterNot(implies(filters, _))
Review Comment:
With `implies`, this check needs to mirror the rewrites of the micro-batch
optimizer: `ConstantPropagation` with the literal equalities, and
`ConstantFolding` and `BooleanSimplification` in `isTrue`. Any rewrite which it
does not mirror, such as the partial folding in the comment on L1205, or one of
an optimizer rule added later, fails a running query with `INTERNAL_ERROR`,
possibly long after it started, and the only way out is a restart with
`spark.sql.streaming.v2FilterPushdown.enabled=false`. Since the pushed filters
come from an optimization of the same query, most of these failures are false
positives.
Could the check compare both sides after the same normalization, e.g., by
running `withValues(pushed)` and the filters of the micro-batch through the
same expression rules as the operator optimization batch, rather than a
hand-picked subset? Alternatively, each micro-batch could apply the pushed
filters explicitly, e.g., with a `Filter` of `pushedFilters` over the relation
before its optimization, so that the scan can never drop a row which the plan
keeps, and no check is needed. That would rely on the startup derivation being
right, which the exclusions at startup (`InferFiltersFromConstraints`,
`StartupValue`) are there for, instead of checking it at each micro-batch.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -1142,6 +1145,78 @@ private[sql] object DataSourceV2Strategy extends Logging
{
withFilter
}
}
+
+ /**
+ * Returns the output of the scan node of a streaming relation: the columns
that its scan reads.
+ * The projections and the filters over the relation, which the scan node is
planned with, must
+ * still imply the filters that the scan got, and read no column of the
relation that the scan
+ * does not read, if it was built to read fewer columns than the relation
outputs (see
+ * `StreamingV2ScanPushDown`).
+ */
+ private def streamingScanOutput(
+ relation: StreamingDataSourceV2ScanRelation,
+ project: Seq[NamedExpression],
+ filters: Seq[Expression]): Seq[AttributeReference] = {
+ val missingFilters = relation.pushedFilters.filterNot(implies(filters, _))
+ if (missingFilters.nonEmpty) {
+ throw SparkException.internalError(
+ s"The plan does not apply the filters
${missingFilters.map(_.sql).mkString(", ")} " +
+ s"that the scan of the streaming relation ${relation.name} got. Set
" +
+ s"${SQLConf.STREAMING_V2_FILTER_PUSHDOWN_ENABLED.key} to false to
push no filter " +
+ "down to the scans of streaming queries.")
+ }
+ relation.prunedOutput match {
+ case Some(prunedOutput) =>
+ val missing = AttributeSet((project ++ filters).flatMap(_.references))
-- prunedOutput
+ if (missing.nonEmpty) {
+ throw SparkException.internalError(
+ s"The scan of the streaming relation ${relation.name} does not
read the columns " +
+ s"${missing.toSeq.map(_.name).mkString(", ")} that the plan
reads from it. Set " +
+ s"${SQLConf.STREAMING_V2_COLUMN_PRUNING_ENABLED.key} to false to
make the scans " +
+ "of streaming queries read all the columns of their sources.")
+ }
+ prunedOutput
+ case None =>
+ relation.output
+ }
+ }
+
+ /**
+ * Returns whether `filters`, the filters over a streaming relation in a
micro-batch, imply
+ * `pushed`, a filter that its scan got when the query started: they have
it, or it holds for the
+ * values of the columns that they make equal to literals. The optimizer of
a micro-batch puts
+ * these values in place of the columns in the other filters, with
`ConstantPropagation`, for
+ * example with an equality that it infers through a join, and then drops a
filter that becomes
+ * true, as `id > 1` does with `id = 5`.
+ */
+ private def implies(filters: Seq[Expression], pushed: Expression): Boolean =
{
+ ExpressionSet(filters).contains(pushed) || {
+ val values = AttributeMap(filters.collect {
+ case EqualTo(column: AttributeReference, value: Literal) => column ->
value
+ case EqualTo(value: Literal, column: AttributeReference) => column ->
value
+ case EqualNullSafe(column: AttributeReference, value: Literal) =>
column -> value
+ case EqualNullSafe(value: Literal, column: AttributeReference) =>
column -> value
+ })
+ def withValues(e: Expression): Expression = e.transform {
+ case column: AttributeReference if values.contains(column) =>
values(column)
+ }
+ values.nonEmpty && {
+ val condition = withValues(pushed)
+ ExpressionSet(filters.map(withValues)).contains(condition) ||
isTrue(condition)
Review Comment:
`implies` puts the values of the `column = literal` equalities in place of
the columns, and then requires the result to be one of the filters as they are,
or to fold to `true`. But a micro-batch also folds what `ConstantPropagation`
leaves, so a pushed filter which only partly folds is still reported as
missing, and a valid query fails with `INTERNAL_ERROR`:
```scala
spark.readStream.table(t).where("owner = current_user() AND (owner = '' OR
id > 5)")
```
- At startup, `owner = StartupValue` is not pushed down, but `owner = '' OR
id > 5` is.
- In each micro-batch, `ReplaceCurrentLike` folds `current_user()` to, e.g.,
`'bob'`, `ConstantPropagation` rewrites the other filter to `'bob' = '' OR id >
5`, and `ConstantFolding` and `BooleanSimplification` turn it into `id > 5`.
- Here, `withValues(pushed)` is `'bob' = '' OR id > 5`, which is neither in
`filters.map(withValues)` (`'bob' = 'bob'` and `id > 5`) nor `true`. So the
first micro-batch fails.
The same happens with an equality which a micro-batch infers through a
stream-static join, e.g., `b = 0` with a pushed `a > 1 OR b > 2`, and with the
values of the time: `d = to_date(now()) AND (d < DATE'2026-11-05' OR id > 5)`
runs today, because `isTrue` holds, and fails from tomorrow, when the pushed
filter folds to `id > 5`.
Could `implies` simplify `withValues(pushed)` with the rules of `isTrue`,
and then accept it when each of its conjuncts is in `filters` or is `true`? A
test for each of these shapes would help too.
##########
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]
Review Comment:
This is on by default for every scan builder which implements
`SupportsPushDownRequiredColumns`. Before this PR, the scans of streaming
queries always got `pruneColumns` with all the columns of the relation, so a
connector whose batch scans honor the pruning, but whose micro-batch or
continuous readers were written for the full schema, was correct. Now such a
connector reports a pruned `readSchema()`, which passes `prunedScanOutput`, so
the scan node outputs the pruned columns while its readers still return all of
them. Nothing checks the rows, so the projection over the scan binds to the
wrong ordinals:
- For a table `(a INT, b INT, c INT)` and `select("c")`, the read schema and
the output of the scan node are `[c]`, and the reader returns `[a, b, c]`, so
the query returns the values of `a` without any error.
- With different types, it reads garbage from an `UnsafeRow` or fails with a
`ClassCastException`.
This PR had to add `ColumnarProjection` to `ColumnarScan` for the same
reason. The migration guide documents the new requirement, but the failure mode
is silent wrong results in an existing query after an upgrade. Could the
pruning of streaming scans be opt-in, e.g., with a mix-in or a method which
tells that the stream readers return the read schema, or off by default for a
release?
##########
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)
Review Comment:
Following up on the fallback
[thread](https://github.com/apache/spark/pull/59943#discussion_r4463353287):
the fallback now covers a pruned `readSchema()` which throws or does not follow
`SupportsPushDownRequiredColumns`, but not the other calls which a streaming
scan did not get before this PR. `pushFilters` / `pushPredicates`,
`pruneColumns` with a partial schema and `build()` of the pruned or filtered
builder (L283 and L285), and `createStream` of the pruned scan (L294) run
outside the `try`. So a connector which throws in any of them, e.g., one whose
`pruneColumns` requires a key column, or whose `pushFilters` needs the state of
a batch read as the Python sources do, now fails to start a query which started
before the upgrade, instead of reading all the columns with no filter.
Could the fallback also cover these calls, with a new scan builder, as L305
does?
##########
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:
Following up on the
[thread](https://github.com/apache/spark/pull/59943#discussion_r4470308884):
`takenFilters` keeps a filter only when its own translation is exactly one of
`pushedFilters()` / `pushedPredicates()`. A builder may report what it applies
in another form, e.g., only the part of `a > 1 OR (b > 2 AND c > 3)` which it
can evaluate, such as `a > 1 OR b > 2`. The scan applies such a filter, but
`StreamingDataSourceV2ScanRelation.pushedFilters` does not have it, so
`streamingScanOutput` does not check it, and a divergence of a micro-batch
would drop rows silently instead of failing.
`PushDownUtils.pushFilters` already keeps `translatedFilterToExpr`, and
rebuilds the post-scan filters from it with `rebuildExpressionFromFilter` /
`rebuildExpressions`. Could it rebuild the reported pushed filters the same
way, so that the record is what the scan may apply, and the filters are not
translated twice?
##########
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:
Nit: `StartupValue` keeps only the children, the data type and the
nullability of the expression, so different expressions become equal:
`current_user()`, `current_database()` and `current_catalog()` all become
`StartupValue(Nil, StringType, false)`, with the same `canonicalized`. I did
not find a rule which merges them today, since they are not deterministic at
startup and `PullOutGroupingExpressions` skips leaves, but a rule which keys
expressions by `canonicalized` could. Could it keep the original expression,
e.g., as a field which is not a child, so that each stays distinct?
##########
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:
Following up on the
[thread](https://github.com/apache/spark/pull/59943#discussion_r4470308931):
thanks for the mix-in. Since it is in
`org.apache.spark.sql.internal.connector`, only the builders of this repo can
use it. A third-party `SupportsPushDownFilters` builder whose `pushFilters`
needs the state of a batch read, as the Python one does, still gets it called
at every stream start by default, and can only avoid it with
`spark.sql.streaming.v2FilterPushdown.enabled=false`, which disables the filter
pushdown for all the sources of the query.
Could the opt-out be a public API of `org.apache.spark.sql.connector.read`,
e.g., a method of `SupportsPushDownFilters` / `SupportsPushDownV2Filters` which
tells whether streaming scans may get the filters?
##########
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:
Minor: `pruneColumns` moved from the scan builder into this lazy
initializer, outside the memoized `Try(planner.plan())` of `ColumnarBatchScan`.
If it throws, `batch` stays uninitialized, so the next `toBatch` calls
`pruneColumns` again on the same `pushedDownPlanner`. The contract of
`ColumnarBatchPlanner` allows `pushLimit` and `pruneColumns` each at most once,
and `ColumnarBatchScan` plans "a failure once too" for the same reason. Could
the pruning be inside the memoized `Try` as well?
##########
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:
Nit: this wraps the batch in a new `ProjectedColumnarBatch` at every call,
while `PartitionReader.get()` "should return same value until `next` is
called". A consumer which calls `get()` twice gets two batches, and closing one
of them releases the columns of the other. Could the reader create the
projected batch once per `next()`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,396 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming.runtime
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation,
ResolvedInlineTable}
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, AttributeSet, Expression, ExpressionSet,
SubqueryExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime,
InferFiltersFromConstraints}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan,
Statistics}
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST_TO_TIMESTAMP,
CURRENT_LIKE}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.classic.SparkSession
+import org.apache.spark.sql.connector.catalog.SupportsRead
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder,
SupportsPushDownFilters, SupportsPushDownRequiredColumns,
SupportsPushDownV2Filters}
+import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
LogicalRelation}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
DataSourceV2Strategy, PushDownUtils}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata,
OffsetSeqMetadataBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.{BatchOnlyFilterPushDown,
SupportsPushDownCatalystFilters}
+import org.apache.spark.sql.sources
+import org.apache.spark.sql.types.{DataType, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Pushes the top-level columns and the filters that the optimized plan of a
streaming query reads
+ * from each Data Source V2 streaming source down to the scan of the source, as
+ * `V2ScanRelationPushDown` does for the scans of batch queries.
+ *
+ * The stream of a source is created from its scan, and has to exist before
any micro-batch is
+ * planned, so the scan is built when the query starts. The query is optimized
once at that point,
+ * by the optimizer of its micro-batches and with their configuration, and the
scan of each source
+ * gets what the planner takes from the optimized plan for the relation of the
source (see
+ * `DataSourceV2Strategy`): the projections and the filters over each
occurrence of the relation.
+ *
+ * The plan of the query does not change, so each micro-batch is optimized as
without these
+ * pushdowns: the optimizer keeps deciding which columns every operator gets,
and in which order,
+ * including the stateful ones, and Spark still evaluates all the filters over
the relation. Only
+ * the scan node of the physical plan outputs the columns that the scan reads,
and the planner
+ * checks that the micro-batch reads no other column of the relation, and that
its filters over the
+ * relation still imply the filters that the scan builder took.
+ *
+ * The other operations that the scans of batch queries can apply, such as
aggregates and limits,
+ * apply to the data of all the micro-batches of a streaming query, which the
scan of a micro-batch
+ * does not read, so they are not pushed down.
+ */
+private[sql] object StreamingV2ScanPushDown extends Logging {
+
+ /**
+ * What the optimized plan of a streaming query reads from a streaming
relation.
+ *
+ * @param requiredColumns the columns of the relation that the plan reads
+ * @param filters the deterministic filters without subqueries that the plan
applies over every
+ * occurrence of the relation, which its scan may apply too
+ */
+ case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression])
+
+ /**
+ * The scan of a streaming source, and its stream.
+ *
+ * @param prunedOutput the columns of the relation that the scan reads, in
the order of its read
+ * schema, unless these are all its columns
+ * @param pushedFilters the filters that the scan builder took, which the
scan may apply, so the
+ * plan of each micro-batch must still apply them over
the relation
+ */
+ case class StreamingScan(
+ scan: Scan,
+ stream: SparkDataStream,
+ prunedOutput: Option[Seq[AttributeReference]],
+ pushedFilters: Seq[Expression])
+
+ /**
+ * Returns what the optimized `plan` reads from each of its streaming
relations of Data Source
+ * V2: what the planner takes from the projections and the filters over each
occurrence of the
+ * relation, or all its columns and no filter where nothing is projected or
filtered over it. A
+ * relation that the optimized plan does not have has no entry, so that its
scan reads all its
+ * columns and gets no filter, as no relation has if the optimizer fails, or
if the optimized
+ * plan has a streaming relation that `plan` does not have.
+ *
+ * The plan is optimized as the micro-batches are: with the optimizer and
the configuration of
+ * the stream, the values of the configurations that the offset log of a
restarted query pins, and
+ * without the cost-based reordering of joins, which `StreamExecution`
disables for its queries.
+ * The plan does not depend on when it is optimized, nor on the data of the
other relations:
+ * - The expressions whose values depend on when they are evaluated or on
the state of the
+ * session, such as the current time or the current database, which each
micro-batch gets the
+ * values of, become placeholders that the optimizer does not take as
deterministic, as the
+ * current time of a micro-batch, and no filter on them is pushed down.
+ * - The filters that the optimizer infers from the constraints of the
plan, such as those of the
+ * filters over another relation joined with the relation, are not
inferred. The optimizer of a
+ * micro-batch may not infer them, for example when the other relation is
a source without new
+ * data in the micro-batch, which the optimizer removes with its filters.
+ * - The relations of batch scans become placeholders, so that their scans
are neither built
+ * nor planned: they do not change what the plan reads from its streaming
relations.
+ *
+ * @param session the session of the stream
+ * @param offsetLogMetadata the metadata of the last entry of the offset log
of the query, if any
+ */
+ def reads(
+ session: SparkSession,
+ plan: LogicalPlan,
+ offsetLogMetadata: Option[OffsetSeqMetadataBase]):
Map[StreamingRelationV2, Reads] = {
+ val conf = session.sessionState.conf.clone()
+ offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf,
logChanges = false))
+ conf.setConf(SQLConf.CBO_ENABLED, false)
+ conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key,
+ (conf.optimizerExcludedRules.toSeq :+
InferFiltersFromConstraints.ruleName).mkString(","))
+ session.withActive {
+ SQLConf.withExistingConf(conf) {
+ try {
+ val optimized =
session.sessionState.optimizer.execute(forStartup(plan))
+ // The planner of a micro-batch gets the values of the expressions
that depend on when
+ // they are evaluated, which are deterministic.
+ val planned = optimized.transformUpWithSubqueries {
+ case node => node.transformExpressions {
+ case value: StartupValue => value.copy(planned = true)
+ }
+ }
+ val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2,
Reads)]
+ // Visits the nodes as the planner plans them: the projections and
the filters over a
+ // relation are planned with its scan, and the other nodes on their
own.
+ def visit(node: LogicalPlan): Unit = node match {
+ case PhysicalOperation(project, filters, relation:
StreamingRelationV2) =>
+ occurrences += relation -> Reads(
+ AttributeSet((project ++ filters).flatMap(_.references)),
+ filters.filter { filter =>
+ filter.deterministic &&
!SubqueryExpression.hasSubquery(filter) &&
+ !filter.exists(_.isInstanceOf[StartupValue])
+ })
+ case _ =>
+ node.children.foreach(visit)
+ }
+ visit(planned)
+ val relations = plan.collect { case relation: StreamingRelationV2 =>
relation }.toSet
+ val unknown = occurrences.map(_._1).filterNot(relations.contains)
+ if (unknown.nonEmpty) {
+ // The optimizer does not copy streaming relations:
`DeduplicateRelations` keeps
+ // streaming plans as they are, so all the references to a common
table expression,
+ // for example, have the same relation. A copy would share the
scan of its relation in
+ // the micro-batches, so the scan would have to read what the copy
reads too.
+ logWarning(log"The optimized streaming query reads the streaming
relations " +
+ log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(",
"))} that its " +
+ log"plan does not have, so the scans of its sources read all
their columns with " +
+ log"no filter.")
+ Map.empty
+ } else {
+ combine(occurrences.toSeq)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Could not optimize the streaming query to push down
what it reads " +
+ log"to the scans of its sources, which read all their columns
with no filter.", e)
+ Map.empty
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns what the occurrences of each relation read, combined: the scan of
a relation reads the
+ * rows of all its occurrences, so it reads the columns that any of them
reads, and can only
+ * apply the filters that all of them apply.
+ */
+ private def combine(
+ occurrences: Seq[(StreamingRelationV2, Reads)]):
Map[StreamingRelationV2, Reads] = {
+ occurrences.groupBy(_._1).map { case (relation, reads) =>
+ val filters = reads.map(r => ExpressionSet(r._2.filters)).reduce {
(left, right) =>
+ ExpressionSet(left.filter(right.contains))
+ }
+ relation -> Reads(
+ AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)),
filters.toSeq)
+ }
+ }
+
+ /**
+ * Returns `plan` for the optimization when the query starts, see `reads`,
with its subqueries.
+ * The expressions that are replaced keep their children, so that the plan
still reads the same
+ * columns.
+ */
+ private def forStartup(plan: LogicalPlan): LogicalPlan =
plan.transformUpWithSubqueries {
+ case relation: DataSourceV2Relation => BatchRelation(relation.output)
+ case relation: LogicalRelation if !relation.isStreaming =>
BatchRelation(relation.output)
+ case relation: HiveTableRelation => BatchRelation(relation.output)
+ // The rows of an inline table may have values of the current time, which
the optimizer
+ // evaluates.
+ case table: ResolvedInlineTable => BatchRelation(table.output)
+ case node =>
+ node.transformExpressionsUpWithPruning(
+ bits => bits.containsPattern(CURRENT_LIKE) ||
bits.containsPattern(CAST_TO_TIMESTAMP)) {
+ // The children of an expression are replaced before it, as are the
subqueries, so an
+ // expression that still has the pattern CURRENT_LIKE has it itself:
its value is the
+ // current time, or comes from the state of the session, as the
current database does.
+ // So is one that ComputeCurrentTime rewrites, such as a cast of a
time to a timestamp,
+ // which takes the current date.
+ case e
+ if e.containsPattern(CURRENT_LIKE) ||
ComputeCurrentTime.applyForExpression(e) != e =>
+ StartupValue(e.children, e.dataType, e.nullable)
+ }
+ }
+
+ /**
+ * Builds the scan of a streaming source, and creates its stream with
`createStream`. If its
+ * scan builder supports it, and `conf` enables it, the scan only reads the
columns of `output`
+ * that the query reads, and gets the filters that the query applies to all
the rows that it reads
+ * from the source (see `reads`). `reads` is only evaluated for such a scan
builder.
+ *
+ * A pruned scan must read the required columns and only columns of
`output`, each once, as
+ * `SupportsPushDownRequiredColumns` requires. A scan that does not, or
whose read schema cannot
+ * be read, is replaced by a scan of all the columns, as without pruning,
before its stream is
+ * created. Spark asks a scan for its read schema before it creates its
stream, as for a batch.
+ *
+ * @param output the output of the relation of the source, with the metadata
columns that the
+ * query reads, which the analyzer added to it, so that
connectors implementing
+ * `SupportsMetadataColumns` include them in `readSchema()`
+ * @param conf the configuration of the query
+ * @param reads what the query reads from the relation, or None to read all
its columns
+ */
+ def buildScan(
+ table: SupportsRead,
+ options: CaseInsensitiveStringMap,
+ output: Seq[AttributeReference],
+ conf: SQLConf,
+ reads: => Option[Reads])(
+ createStream: Scan => SparkDataStream): StreamingScan = {
+ val builder = table.newScanBuilder(options)
+ val prunesColumns = conf.streamingV2ColumnPruningEnabled &&
+ builder.isInstanceOf[SupportsPushDownRequiredColumns]
+ val pushesFilters = conf.streamingV2FilterPushdownEnabled &&
supportsFilters(builder)
+ val known = if (prunesColumns || pushesFilters) reads else None
+ val requiredOutput = known.filter(_ => prunesColumns)
+ .map(r => output.filter(r.requiredColumns.contains))
+ .getOrElse(output)
+ val filters = known.filter(_ => pushesFilters)
+ .map(r => DataSourceStrategy.normalizeExprs(r.filters, output))
+ .getOrElse(Nil)
+
+ // Builds the scan, and returns it with the filters that the builder took.
+ def build(builder: ScanBuilder, columns: Seq[AttributeReference]): (Scan,
Seq[Expression]) = {
+ val pushedFilters = if (filters.nonEmpty) {
+ // The plan of the query keeps the filters, which Spark evaluates
after the scan, so the
+ // scan may apply any of them.
+ val (pushed, _) = PushDownUtils.pushFilters(builder, filters,
partitionFields = None)
+ takenFilters(filters, pushed)
+ } else {
+ Nil
+ }
+ builder match {
+ case pruning: SupportsPushDownRequiredColumns =>
+ pruning.pruneColumns(DataTypeUtils.fromAttributes(columns))
+ case _ =>
+ }
+ (builder.build(), pushedFilters)
+ }
+ def withStream(
+ built: (Scan, Seq[Expression]),
+ prunedOutput: Option[Seq[AttributeReference]]): StreamingScan = {
+ val (scan, pushedFilters) = built
+ StreamingScan(scan, createStream(scan), prunedOutput, pushedFilters)
+ }
+
+ if (requiredOutput.length == output.length) {
+ withStream(build(builder, output), None)
+ } else {
+ val built = build(builder, requiredOutput)
+ val scanOutput = try {
+ prunedScanOutput(built._1.readSchema(), output,
requiredOutput).left.map(_ -> None)
Review Comment:
Nit: the reason and the cause go through `Left((String,
Option[Throwable]))`, built by `.left.map(_ -> None)` here and by `Left(... ->
Some(e))` at L289, and are matched again at L301-304. Could a local function
which logs the warning and builds the scan of all the columns be called from
the `catch`, with the throwable, and from the `Left` case of
`prunedScanOutput`, without it?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala:
##########
@@ -123,39 +125,55 @@ class ColumnarScanBuilder(
override def pruneColumns(requiredSchema: StructType): Unit = {
// Only top-level columns are pruned by the data source. Spark prunes the
nested fields of the
- // columns it reads.
+ // columns it reads. The required columns are only recorded here, see
ColumnarScan.
val required =
StructType(requiredSchema.fieldNames.flatMap(name => schema.find(_.name
== name)))
- if (required.length < schema.length &&
batchPlanner.pruneColumns(required)) {
- readSchema = required
+ if (required.length < schema.length) {
+ this.requiredSchema = Some(required)
}
}
// The scan takes over the planner, so it is built once.
private lazy val scan = new ColumnarScan(
- shortName, schema, readSchema, dataSource, Option(planner), pushedLimit)
+ shortName, schema, requiredSchema, dataSource, Option(planner),
pushedLimit)
override def build(): Scan = scan
}
/**
+ * A scan of a [[ColumnarDataSource]]. It reads the top-level columns that the
query needs, the
+ * columns of its read schema: a batch scan asks its planner to prune the
other columns when it
+ * plans its partitions, and drops them from the batches of the reader if the
planner does not,
+ * while a streaming scan drops them from the batches of its stream reader,
which reads all the
+ * columns of the full schema.
+ *
+ * @param requiredSchema the top-level columns that the query needs, if fewer
than those of the
+ * full schema
* @param pushedDownPlanner the planner that the operations were pushed down
to, if any. Otherwise,
* the scan creates one when its batch is planned.
*/
class ColumnarScan(
shortName: String,
fullSchema: StructType,
- prunedSchema: StructType,
+ requiredSchema: Option[StructType],
dataSource: ColumnarDataSource,
pushedDownPlanner: Option[ColumnarBatchPlanner],
pushedLimit: Option[Int])
extends Scan with SupportsMetadata {
- // Spark may call `toBatch` more than once for the same scan; the partitions
are planned once.
- private lazy val batch = new ColumnarBatchScan(
- pushedDownPlanner.getOrElse(dataSource.batchPlanner(fullSchema)))
+ // The ordinals in the full schema of the columns that the scan reads, if it
reads fewer.
+ private def requiredOrdinals: Option[Array[Int]] =
+ requiredSchema.map(_.fieldNames.map(fullSchema.fieldIndex))
+
+ // Spark may call `toBatch` more than once for the same scan; the partitions
are planned once,
+ // after the planner prunes the columns, if it does.
+ private lazy val batch = {
+ val planner =
pushedDownPlanner.getOrElse(dataSource.batchPlanner(fullSchema))
+ val pruned = requiredSchema.forall(planner.pruneColumns)
+ new ColumnarBatchScan(planner, if (pruned) None else requiredOrdinals)
+ }
- override def readSchema(): StructType = prunedSchema
+ override def readSchema(): StructType = requiredSchema.getOrElse(fullSchema)
Review Comment:
Nit: `readSchema()`, and `ReadSchema` of `getMetaData()` at L190, are now
the required columns also when the planner does not prune them, e.g., for a
native source whose `pruneColumns` returns `false`, or for a stream, whose
reader reads the full schema. The scan returns the right columns thanks to
`ColumnarProjection`, but `explain` shows, e.g., `ReadSchema:
struct<name:string>` while the source reads all its columns, whereas before
this PR `ReadSchema` showed what the planner accepted. Could the metadata tell
when the columns are dropped after the read?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -1142,6 +1145,78 @@ private[sql] object DataSourceV2Strategy extends Logging
{
withFilter
}
}
+
+ /**
+ * Returns the output of the scan node of a streaming relation: the columns
that its scan reads.
+ * The projections and the filters over the relation, which the scan node is
planned with, must
+ * still imply the filters that the scan got, and read no column of the
relation that the scan
+ * does not read, if it was built to read fewer columns than the relation
outputs (see
+ * `StreamingV2ScanPushDown`).
+ */
+ private def streamingScanOutput(
+ relation: StreamingDataSourceV2ScanRelation,
+ project: Seq[NamedExpression],
+ filters: Seq[Expression]): Seq[AttributeReference] = {
+ val missingFilters = relation.pushedFilters.filterNot(implies(filters, _))
+ if (missingFilters.nonEmpty) {
+ throw SparkException.internalError(
+ s"The plan does not apply the filters
${missingFilters.map(_.sql).mkString(", ")} " +
+ s"that the scan of the streaming relation ${relation.name} got. Set
" +
+ s"${SQLConf.STREAMING_V2_FILTER_PUSHDOWN_ENABLED.key} to false to
push no filter " +
+ "down to the scans of streaming queries.")
+ }
+ relation.prunedOutput match {
+ case Some(prunedOutput) =>
+ val missing = AttributeSet((project ++ filters).flatMap(_.references))
-- prunedOutput
+ if (missing.nonEmpty) {
+ throw SparkException.internalError(
+ s"The scan of the streaming relation ${relation.name} does not
read the columns " +
+ s"${missing.toSeq.map(_.name).mkString(", ")} that the plan
reads from it. Set " +
+ s"${SQLConf.STREAMING_V2_COLUMN_PRUNING_ENABLED.key} to false to
make the scans " +
+ "of streaming queries read all the columns of their sources.")
+ }
+ prunedOutput
+ case None =>
+ relation.output
+ }
+ }
+
+ /**
+ * Returns whether `filters`, the filters over a streaming relation in a
micro-batch, imply
+ * `pushed`, a filter that its scan got when the query started: they have
it, or it holds for the
+ * values of the columns that they make equal to literals. The optimizer of
a micro-batch puts
+ * these values in place of the columns in the other filters, with
`ConstantPropagation`, for
+ * example with an equality that it infers through a join, and then drops a
filter that becomes
+ * true, as `id > 1` does with `id = 5`.
+ */
+ private def implies(filters: Seq[Expression], pushed: Expression): Boolean =
{
Review Comment:
Nit: `ExpressionSet(filters)`, the `AttributeMap` of the values and
`filters.map(withValues)` only depend on `filters`, but they are built again
for every pushed filter, at every micro-batch. Could `streamingScanOutput`
build them once for all the pushed filters?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,396 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming.runtime
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation,
ResolvedInlineTable}
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, AttributeSet, Expression, ExpressionSet,
SubqueryExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime,
InferFiltersFromConstraints}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
+import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan,
Statistics}
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.catalyst.trees.TreePattern.{CAST_TO_TIMESTAMP,
CURRENT_LIKE}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.classic.SparkSession
+import org.apache.spark.sql.connector.catalog.SupportsRead
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder,
SupportsPushDownFilters, SupportsPushDownRequiredColumns,
SupportsPushDownV2Filters}
+import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
LogicalRelation}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
DataSourceV2Strategy, PushDownUtils}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata,
OffsetSeqMetadataBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.{BatchOnlyFilterPushDown,
SupportsPushDownCatalystFilters}
+import org.apache.spark.sql.sources
+import org.apache.spark.sql.types.{DataType, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Pushes the top-level columns and the filters that the optimized plan of a
streaming query reads
+ * from each Data Source V2 streaming source down to the scan of the source, as
+ * `V2ScanRelationPushDown` does for the scans of batch queries.
+ *
+ * The stream of a source is created from its scan, and has to exist before
any micro-batch is
+ * planned, so the scan is built when the query starts. The query is optimized
once at that point,
+ * by the optimizer of its micro-batches and with their configuration, and the
scan of each source
+ * gets what the planner takes from the optimized plan for the relation of the
source (see
+ * `DataSourceV2Strategy`): the projections and the filters over each
occurrence of the relation.
+ *
+ * The plan of the query does not change, so each micro-batch is optimized as
without these
+ * pushdowns: the optimizer keeps deciding which columns every operator gets,
and in which order,
+ * including the stateful ones, and Spark still evaluates all the filters over
the relation. Only
+ * the scan node of the physical plan outputs the columns that the scan reads,
and the planner
+ * checks that the micro-batch reads no other column of the relation, and that
its filters over the
+ * relation still imply the filters that the scan builder took.
+ *
+ * The other operations that the scans of batch queries can apply, such as
aggregates and limits,
+ * apply to the data of all the micro-batches of a streaming query, which the
scan of a micro-batch
+ * does not read, so they are not pushed down.
+ */
+private[sql] object StreamingV2ScanPushDown extends Logging {
+
+ /**
+ * What the optimized plan of a streaming query reads from a streaming
relation.
+ *
+ * @param requiredColumns the columns of the relation that the plan reads
+ * @param filters the deterministic filters without subqueries that the plan
applies over every
+ * occurrence of the relation, which its scan may apply too
+ */
+ case class Reads(requiredColumns: AttributeSet, filters: Seq[Expression])
+
+ /**
+ * The scan of a streaming source, and its stream.
+ *
+ * @param prunedOutput the columns of the relation that the scan reads, in
the order of its read
+ * schema, unless these are all its columns
+ * @param pushedFilters the filters that the scan builder took, which the
scan may apply, so the
+ * plan of each micro-batch must still apply them over
the relation
+ */
+ case class StreamingScan(
+ scan: Scan,
+ stream: SparkDataStream,
+ prunedOutput: Option[Seq[AttributeReference]],
+ pushedFilters: Seq[Expression])
+
+ /**
+ * Returns what the optimized `plan` reads from each of its streaming
relations of Data Source
+ * V2: what the planner takes from the projections and the filters over each
occurrence of the
+ * relation, or all its columns and no filter where nothing is projected or
filtered over it. A
+ * relation that the optimized plan does not have has no entry, so that its
scan reads all its
+ * columns and gets no filter, as no relation has if the optimizer fails, or
if the optimized
+ * plan has a streaming relation that `plan` does not have.
+ *
+ * The plan is optimized as the micro-batches are: with the optimizer and
the configuration of
+ * the stream, the values of the configurations that the offset log of a
restarted query pins, and
+ * without the cost-based reordering of joins, which `StreamExecution`
disables for its queries.
+ * The plan does not depend on when it is optimized, nor on the data of the
other relations:
+ * - The expressions whose values depend on when they are evaluated or on
the state of the
+ * session, such as the current time or the current database, which each
micro-batch gets the
+ * values of, become placeholders that the optimizer does not take as
deterministic, as the
+ * current time of a micro-batch, and no filter on them is pushed down.
+ * - The filters that the optimizer infers from the constraints of the
plan, such as those of the
+ * filters over another relation joined with the relation, are not
inferred. The optimizer of a
+ * micro-batch may not infer them, for example when the other relation is
a source without new
+ * data in the micro-batch, which the optimizer removes with its filters.
+ * - The relations of batch scans become placeholders, so that their scans
are neither built
+ * nor planned: they do not change what the plan reads from its streaming
relations.
+ *
+ * @param session the session of the stream
+ * @param offsetLogMetadata the metadata of the last entry of the offset log
of the query, if any
+ */
+ def reads(
+ session: SparkSession,
+ plan: LogicalPlan,
+ offsetLogMetadata: Option[OffsetSeqMetadataBase]):
Map[StreamingRelationV2, Reads] = {
+ val conf = session.sessionState.conf.clone()
+ offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf,
logChanges = false))
+ conf.setConf(SQLConf.CBO_ENABLED, false)
+ conf.setConfString(SQLConf.OPTIMIZER_EXCLUDED_RULES.key,
+ (conf.optimizerExcludedRules.toSeq :+
InferFiltersFromConstraints.ruleName).mkString(","))
+ session.withActive {
+ SQLConf.withExistingConf(conf) {
+ try {
+ val optimized =
session.sessionState.optimizer.execute(forStartup(plan))
+ // The planner of a micro-batch gets the values of the expressions
that depend on when
+ // they are evaluated, which are deterministic.
+ val planned = optimized.transformUpWithSubqueries {
+ case node => node.transformExpressions {
+ case value: StartupValue => value.copy(planned = true)
+ }
+ }
+ val occurrences = mutable.ArrayBuffer.empty[(StreamingRelationV2,
Reads)]
+ // Visits the nodes as the planner plans them: the projections and
the filters over a
+ // relation are planned with its scan, and the other nodes on their
own.
+ def visit(node: LogicalPlan): Unit = node match {
+ case PhysicalOperation(project, filters, relation:
StreamingRelationV2) =>
+ occurrences += relation -> Reads(
+ AttributeSet((project ++ filters).flatMap(_.references)),
+ filters.filter { filter =>
+ filter.deterministic &&
!SubqueryExpression.hasSubquery(filter) &&
+ !filter.exists(_.isInstanceOf[StartupValue])
+ })
+ case _ =>
+ node.children.foreach(visit)
+ }
+ visit(planned)
+ val relations = plan.collect { case relation: StreamingRelationV2 =>
relation }.toSet
+ val unknown = occurrences.map(_._1).filterNot(relations.contains)
+ if (unknown.nonEmpty) {
+ // The optimizer does not copy streaming relations:
`DeduplicateRelations` keeps
+ // streaming plans as they are, so all the references to a common
table expression,
+ // for example, have the same relation. A copy would share the
scan of its relation in
+ // the micro-batches, so the scan would have to read what the copy
reads too.
+ logWarning(log"The optimized streaming query reads the streaming
relations " +
+ log"${MDC(LogKeys.STREAMING_TABLE, unknown.distinct.mkString(",
"))} that its " +
+ log"plan does not have, so the scans of its sources read all
their columns with " +
+ log"no filter.")
+ Map.empty
+ } else {
+ combine(occurrences.toSeq)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Could not optimize the streaming query to push down
what it reads " +
+ log"to the scans of its sources, which read all their columns
with no filter.", e)
+ Map.empty
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns what the occurrences of each relation read, combined: the scan of
a relation reads the
+ * rows of all its occurrences, so it reads the columns that any of them
reads, and can only
+ * apply the filters that all of them apply.
+ */
+ private def combine(
+ occurrences: Seq[(StreamingRelationV2, Reads)]):
Map[StreamingRelationV2, Reads] = {
+ occurrences.groupBy(_._1).map { case (relation, reads) =>
+ val filters = reads.map(r => ExpressionSet(r._2.filters)).reduce {
(left, right) =>
+ ExpressionSet(left.filter(right.contains))
+ }
+ relation -> Reads(
+ AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)),
filters.toSeq)
+ }
+ }
+
+ /**
+ * Returns `plan` for the optimization when the query starts, see `reads`,
with its subqueries.
+ * The expressions that are replaced keep their children, so that the plan
still reads the same
+ * columns.
+ */
+ private def forStartup(plan: LogicalPlan): LogicalPlan =
plan.transformUpWithSubqueries {
+ case relation: DataSourceV2Relation => BatchRelation(relation.output)
+ case relation: LogicalRelation if !relation.isStreaming =>
BatchRelation(relation.output)
+ case relation: HiveTableRelation => BatchRelation(relation.output)
+ // The rows of an inline table may have values of the current time, which
the optimizer
+ // evaluates.
+ case table: ResolvedInlineTable => BatchRelation(table.output)
+ case node =>
+ node.transformExpressionsUpWithPruning(
+ bits => bits.containsPattern(CURRENT_LIKE) ||
bits.containsPattern(CAST_TO_TIMESTAMP)) {
+ // The children of an expression are replaced before it, as are the
subqueries, so an
+ // expression that still has the pattern CURRENT_LIKE has it itself:
its value is the
+ // current time, or comes from the state of the session, as the
current database does.
+ // So is one that ComputeCurrentTime rewrites, such as a cast of a
time to a timestamp,
+ // which takes the current date.
+ case e
+ if e.containsPattern(CURRENT_LIKE) ||
ComputeCurrentTime.applyForExpression(e) != e =>
Review Comment:
Nit: an expression which is not rewritten can still have the pattern
`CAST_TO_TIMESTAMP`, e.g., every ancestor of `cast(s AS TIMESTAMP)` of a string
column. For each of them, this guard runs
`ComputeCurrentTime.applyForExpression(e)` again, with a new `Instant.now()`
and `TimeSnapshot`, and walks down to the cast each time. Following up on the
[thread](https://github.com/apache/spark/pull/59943#discussion_r4470308953):
since `ComputeCurrentTime` only rewrites a `Cast` without `CURRENT_LIKE`, could
this call it only for a `Cast`, e.g., `case c: Cast if
ComputeCurrentTime.applyForExpression(c) != c`, which still reuses its guards?
--
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]
