zikangh commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4485731164
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -261,6 +260,12 @@ class MicroBatchExecution(
}
import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+ // The columns that the query reads from each Data Source V2 source, which
the scan of the
+ // source reads if its table has the capability STREAMING_COLUMN_PRUNING
and its scan builder
+ // supports the pruning, from the plan of the query optimized once as its
micro-batches are.
+ // The plan is only optimized for such a source.
+ lazy val sourceReads = StreamingV2ScanPushDown.reads(
Review Comment:
"reads" is too ambiguous. could you make it more descriptive and explicit so
we can remove the comment above?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -312,7 +313,7 @@ class MicroBatchExecution(
},
sourceIdentifyingName
)
- StreamingDataSourceV2ScanRelation(relation, scan, output, stream)
+ streamingScan.toRelation(relation)
Review Comment:
toRelation just contains one line, it's more readable if we don't create
this method.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,391 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming.runtime
+
+import scala.collection.mutable
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation,
ResolvedInlineTable}
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, AttributeSet, Cast, Expression, Unevaluable}
+import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime,
ReplaceExpressions, SpecialDatetimeValues}
+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, CURRENT_LIKE,
RUNTIME_REPLACEABLE}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.classic.SparkSession
+import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability}
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder,
SupportsPushDownRequiredColumns}
+import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata,
OffsetSeqMetadataBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType,
StructType, TimestampLTZNanosType, TimestampType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Pushes the top-level columns that the optimized plan of a streaming query
reads from each Data
+ * Source V2 streaming source down to the scan of the source, as
`V2ScanRelationPushDown` does for
+ * the scans of batch queries, if the table of the source has the capability
+ * `STREAMING_COLUMN_PRUNING`.
+ *
+ * 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
+ * reads the columns that the planner takes from the optimized plan for the
relation of the source
+ * (see `DataSourceV2Strategy`): those of the projections and the filters over
each occurrence of
+ * the relation.
+ *
+ * The plan of each micro-batch is the one without this pushdown: the
optimizer keeps deciding
+ * which columns every operator gets, and in which order, including the
stateful ones. Only the
+ * scan node of the physical plan outputs the columns that the scan reads, and
the planner checks
+ * that the micro-batch reads no other column of the relation.
+ *
+ * The other operations that the scans of batch queries can apply, such as
aggregates and limits,
+ * apply to the data of all the micro-batches of a streaming query, which the
scan of a micro-batch
+ * does not read, so they are not pushed down.
+ */
+private[sql] object StreamingV2ScanPushDown extends Logging {
+
+ /**
+ * What the optimized plan of a streaming query reads from a streaming
relation.
+ *
+ * @param requiredColumns the columns of the relation that the plan reads
+ */
+ case class Reads(requiredColumns: AttributeSet)
+
+ /**
+ * 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
+ */
+ case class StreamingScan(
+ scan: Scan,
+ stream: SparkDataStream,
+ prunedOutput: Option[Seq[AttributeReference]]) {
+
+ /** Returns the scan relation of the source whose relation is `relation`.
*/
+ def toRelation(
+ relation: StreamingDataSourceV2Relation):
StreamingDataSourceV2ScanRelation = {
+ StreamingDataSourceV2ScanRelation(relation, scan, relation.output,
stream,
+ prunedOutput = prunedOutput)
+ }
+ }
+
+ /**
+ * Returns what the optimized `plan` reads from each of its streaming
relations of Data Source
+ * V2: the columns that the planner takes from the projections and the
filters over each
+ * occurrence of the relation, or all its columns 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, 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:
+ * - The expressions whose values depend on when they are evaluated or on
the state of the
+ * session, such as the current time, the casts of special strings such
as 'today' or 'now'
+ * to dates and timestamps, the casts of strings to timestamps with local
time zone, which take
+ * the current date for a string with only a 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, so the columns
that the plan reads do
+ * not depend on them.
+ * - 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)
+ 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)))
+ 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.")
+ Map.empty
+ } else {
+ combine(occurrences.toSeq)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Could not optimize the streaming query to push down
the columns " +
+ log"that it reads to the scans of its sources, which read all
their columns.", 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.
+ */
+ private def combine(
+ occurrences: Seq[(StreamingRelationV2, Reads)]):
Map[StreamingRelationV2, Reads] = {
+ occurrences.groupBy(_._1).map { case (relation, reads) =>
+ relation ->
Reads(AttributeSet.fromAttributeSets(reads.map(_._2.requiredColumns)))
+ }
+ }
+
+ /**
+ * Returns `plan` for the optimization when the query starts, see `reads`,
with its subqueries.
+ * The expressions that are replaced keep their children, so that the plan
still reads the same
+ * columns. Visible for tests.
+ */
+ private[sql] def forStartup(plan: LogicalPlan): LogicalPlan =
plan.transformUpWithSubqueries {
+ case relation: DataSourceV2Relation => BatchRelation(relation.output)
Review Comment:
reads doesn't apply the cache, so a cached static side is transparent here
but an opaque InMemoryRelation in each micro-batch. e.g. joining with
`dim.withColumn("on", lit(false)).cache()` and selecting `when($"on",
$"data").otherwise($"extra")` prunes data at startup, and every micro-batch
then fails the column check. could forStartup replace each maximal
non-streaming subtree with a BatchRelation instead of only the leaves?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,391 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming.runtime
+
+import scala.collection.mutable
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation,
ResolvedInlineTable}
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, AttributeSet, Cast, Expression, Unevaluable}
+import org.apache.spark.sql.catalyst.optimizer.{ComputeCurrentTime,
ReplaceExpressions, SpecialDatetimeValues}
+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, CURRENT_LIKE,
RUNTIME_REPLACEABLE}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.classic.SparkSession
+import org.apache.spark.sql.connector.catalog.{SupportsRead, TableCapability}
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder,
SupportsPushDownRequiredColumns}
+import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
PushDownUtils, StreamingDataSourceV2Relation, StreamingDataSourceV2ScanRelation}
+import
org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata,
OffsetSeqMetadataBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType,
StructType, TimestampLTZNanosType, TimestampType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Pushes the top-level columns that the optimized plan of a streaming query
reads from each Data
+ * Source V2 streaming source down to the scan of the source, as
`V2ScanRelationPushDown` does for
+ * the scans of batch queries, if the table of the source has the capability
+ * `STREAMING_COLUMN_PRUNING`.
+ *
+ * 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
+ * reads the columns that the planner takes from the optimized plan for the
relation of the source
+ * (see `DataSourceV2Strategy`): those of the projections and the filters over
each occurrence of
+ * the relation.
+ *
+ * The plan of each micro-batch is the one without this pushdown: the
optimizer keeps deciding
+ * which columns every operator gets, and in which order, including the
stateful ones. Only the
+ * scan node of the physical plan outputs the columns that the scan reads, and
the planner checks
+ * that the micro-batch reads no other column of the relation.
+ *
+ * The other operations that the scans of batch queries can apply, such as
aggregates and limits,
+ * apply to the data of all the micro-batches of a streaming query, which the
scan of a micro-batch
+ * does not read, so they are not pushed down.
+ */
+private[sql] object StreamingV2ScanPushDown extends Logging {
+
+ /**
+ * What the optimized plan of a streaming query reads from a streaming
relation.
+ *
+ * @param requiredColumns the columns of the relation that the plan reads
+ */
+ case class Reads(requiredColumns: AttributeSet)
+
+ /**
+ * 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
+ */
+ case class StreamingScan(
+ scan: Scan,
+ stream: SparkDataStream,
+ prunedOutput: Option[Seq[AttributeReference]]) {
+
+ /** Returns the scan relation of the source whose relation is `relation`.
*/
+ def toRelation(
+ relation: StreamingDataSourceV2Relation):
StreamingDataSourceV2ScanRelation = {
+ StreamingDataSourceV2ScanRelation(relation, scan, relation.output,
stream,
+ prunedOutput = prunedOutput)
+ }
+ }
+
+ /**
+ * Returns what the optimized `plan` reads from each of its streaming
relations of Data Source
Review Comment:
most of this comment is impl details, and way too verbose.
--
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]
