HyukjinKwon commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4470803207
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -1216,15 +1251,29 @@ class BufferedRows(val key: Seq[Any], val schema:
StructType)
def clear(): Unit = rows.clear()
}
+/**
+ * @param dataOrdinals the positions in the rows of the table of the data
columns to read
+ * @param readsAllColumns whether these are all the columns of the table, in
their order
+ */
private class InMemoryMicroBatchReaderFactory(
- metaNames: Array[String]) extends PartitionReaderFactory with Serializable
{
+ metaNames: Array[String],
+ dataOrdinals: Array[Int],
+ dataTypes: Array[DataType],
+ readsAllColumns: Boolean) extends PartitionReaderFactory with Serializable
{
override def createReader(partition: InputPartition):
PartitionReader[InternalRow] = {
val rows = partition.asInstanceOf[InMemoryMicroBatchPartition].rows
new PartitionReader[InternalRow] {
private var idx = -1
override def next(): Boolean = { idx += 1; idx < rows.size }
override def get(): InternalRow = {
- val rawRow = rows(idx)
+ val tableRow = rows(idx)
+ val rawRow = if (readsAllColumns) {
+ tableRow
+ } else {
+ new GenericInternalRow(dataOrdinals.indices.map { i =>
Review Comment:
Done in 3a71891: both readers project their rows with
`ProjectingInternalRow`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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
+import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, AttributeSet, Cast, Expression, ExpressionSet,
SubqueryExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.optimizer.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.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,
PushDownUtils}
+import org.apache.spark.sql.execution.datasources.v2.python.PythonScanBuilder
+import
org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata,
OffsetSeqMetadataBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters
+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
still applies the filters
+ * that the scan got.
+ *
+ * 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 got, which the plan of
each micro-batch must
+ * still apply over the relation
+ */
+ case class StreamingScan[S <: SparkDataStream](
Review Comment:
Done in 3a71891: `StreamingScan` and `buildScan` take a `SparkDataStream`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -1101,6 +1103,41 @@ 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 apply 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(ExpressionSet(filters).contains)
Review Comment:
Fixed in 3a71891, see dongjoon-hyun's comment on
`StreamingV2ScanPushDown.scala:126`: the check now accepts the pushed filters
that the filters over the relation imply, also once the values of their
equalities are put in place of the columns, so `id > 10` holds with `id = 20`.
"plan the micro-batches whose filters imply the pushed filters in another form"
runs a stream-static join and a stream-stream join with such an equality, in
micro-batches where both sources have new rows with the matching key; with the
containment check, they failed as you described.
--
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]
