HyukjinKwon commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4472653031
##########
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:
Fixed in becbceb, following your alternative on L1160: `implies` and the
check of the filters are gone. The plan of the query now applies all the
filters that the scan builder got to the rows of each read of the relation
(`StreamingV2ScanPushDown.withPushedFilters`, a `Filter` over the relation,
which the optimizer merges with those of the query), so whatever a micro-batch
folds, in full or in part, its plan still applies every filter that the scan
may apply. The new test "plan the micro-batches whose filters fold the pushed
filters in part" runs your three shapes: `owner = current_user() AND (owner =
'' OR id > 5)`, `d = to_date(now()) AND (d < DATE'2000-01-01' OR id > 5)`, a
date of the past standing for the day after the start, and `a > 1 OR b > 2`
with `b = 0` inferred through a stream-static join.
##########
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:
Done in becbceb: the pruning of streaming scans is opt-in, with the new
`TableCapability.STREAMING_COLUMN_PRUNING`, by which a table declares that the
partition readers of the streams of its scans return the columns of
`Scan.readSchema()`. A table without it gets `pruneColumns` with all the
columns of its output, as before this PR, so a connector whose stream readers
were written for the full schema keeps working after the upgrade. I put it on
the table, as `SCAN_MERGING` is, next to the other capabilities of streaming
reads (`MICRO_BATCH_READ`, `CONTINUOUS_READ`), so that it is declared once for
both executions. The in-memory test table and `ColumnarTable`, whose streams
project the batches of their readers, declare it. The migration guide entries
are gone, since nothing changes for an existing source. New tests: "read all
the columns with no filter from a table without the capabilities" and "push
down only what the capabilities of the table allow".
##########
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:
Done in becbceb, with your second option: each micro-batch applies the
pushed filters explicitly, with a `Filter` of `pushedFilters` over the
relation, so the scan cannot drop a row which the plan keeps, and the check,
`implies` and `isTrue` are gone. The filter is added where the executions put
the scan relations into the plan of the query, at each read of the relation,
and it holds the filters that the builder got, not only those that it took, so
the builder may apply any of them in any form. As you say, this relies on the
derivation at the start, which keeps its exclusions (placeholders of the time
values, no inferred filters). The planner still checks the columns, which no
rewrite of a micro-batch can add.
##########
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:
Done in becbceb: the fallback now covers every call of the pushdown:
`pushFilters` / `pushPredicates`, `pruneColumns`, `build()`, `readSchema()` and
the creation of the stream all run in a `Try`, and any failure, or a read
schema that does not follow `SupportsPushDownRequiredColumns`, builds a scan of
all the columns with no filter from a new scan builder, with a warning, as
without the pushdowns. With the opt-in of L248 and L27, only a table that
declares a capability gets these calls in the first place. New test: "read all
the columns with no filter when the scan fails with the pushdowns", for a
builder that fails in each of the four calls.
--
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]
