HyukjinKwon commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4488752312
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -288,15 +289,16 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
}
val microBatchStream = r.stream.asInstanceOf[MicroBatchStream]
new RealTimeStreamScanExec(
- r.output,
+ DataSourceV2Strategy.streamingScanOutput(r, p, f),
Review Comment:
Done in 0072b09: the new `StreamingV2ScanPushDownRealTimeSuite` (on
`StreamRealTimeModeSuiteBase`) runs `.where("extra > 10").select("data")` in
real-time mode over a table with `STREAMING_COLUMN_PRUNING`, whose stream is a
`LowLatencyMemoryStream` with readers that return the columns of the read
schema. It checks the answer, that the scan reads `_2` and `_3`, and that
`RealTimeStreamScanExec` outputs `data` and `extra`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6255,6 +6255,21 @@ object SQLConf {
.stringConf
.createWithDefault("")
+ val STREAMING_V2_COLUMN_PRUNING_ENABLED =
+ buildConf("spark.sql.streaming.v2ColumnPruning.enabled")
+ .internal()
Review Comment:
Done in 0072b09: `spark.sql.streaming.v2ColumnPruning.enabled` is no longer
internal, so the generated docs list it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -478,28 +478,45 @@ case class StreamingDataSourceV2Relation(
/**
* A specialization of [[DataSourceV2ScanRelation]] with the streaming bit set
to true, as well
* as start and end offsets for Microbatch processing.
+ *
+ * @param output the columns of the source that the query references, which
the plan of each
+ * micro-batch is optimized with
+ * @param scanOutput the columns of `output` that the scan reads, in the order
of its read schema,
+ * unless these are all the columns of `output` in their
order. The scan node of
+ * the physical plan outputs these, so the plan must not
read the other columns
+ * of `output`.
*/
case class StreamingDataSourceV2ScanRelation(
relation: StreamingDataSourceV2Relation,
scan: Scan,
output: Seq[AttributeReference],
stream: SparkDataStream,
startOffset: Option[Offset] = None,
- endOffset: Option[Offset] = None)
+ endOffset: Option[Offset] = None,
+ scanOutput: Option[Seq[AttributeReference]] = None)
Review Comment:
Done in 0072b09: the scaladoc of `scanOutput` now says that the readers of
the stream return these columns, so a planner must build the scan node from
them rather than from `output`, and check that the plan reads no other column
of `output`, as `DataSourceV2Strategy` does.
--
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]
