HyukjinKwon commented on code in PR #59943:
URL: https://github.com/apache/spark/pull/59943#discussion_r4489954272
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -1142,6 +1146,33 @@ private[sql] object DataSourceV2Strategy extends Logging
{
withFilter
}
}
+
+ /**
+ * Returns the output of the scan node of a streaming relation: the columns
that its scan reads,
+ * in the order of its read schema. The projections and the filters over the
relation, which the
+ * scan node is planned with, must 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] = relation.scanOutput
match {
+ case Some(scanOutput) =>
+ val missing = AttributeSet((project ++ filters).flatMap(_.references))
-- scanOutput
+ if (missing.nonEmpty) {
+ val source = (relation.catalog, relation.identifier) match {
+ case (Some(catalog), Some(identifier)) =>
+ (catalog.name +: identifier.namespace.toSeq) :+ identifier.name
Review Comment:
Done, thank you. Both errors now take the name from
`StreamingV2ScanPushDown.sourceName(table, catalog, identifier)`: the catalog,
the namespace and the name of the identifier, one part each, or the table's
name when the relation has no identifier. Both executions pass it to
`buildScan` for `STREAMING_SCAN_INVALID_READ_SCHEMA`, and
`DataSourceV2Strategy` uses it for `STREAMING_SCAN_MISSING_COLUMNS`. So the
same table shows as `` `cat`.`db`.`t` `` in both.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/OffsetSeq.scala:
##########
@@ -343,15 +343,25 @@ object OffsetSeqMetadata extends Logging {
metadata.copy(conf = conf)
}
- /** Set the SparkSession configuration with the values in the metadata */
- def setSessionConf(metadata: OffsetSeqMetadataBase, sessionConf: SQLConf):
Unit = {
+ /**
+ * Set the SparkSession configuration with the values in the metadata.
+ *
+ * @param logChanges whether to log the values that are not in the metadata
or that differ from
+ * those of the session, which a copy of the configuration
of a query that
+ * gets them again later does not need to log twice
+ */
+ def setSessionConf(
+ metadata: OffsetSeqMetadataBase,
+ sessionConf: SQLConf,
+ logChanges: Boolean = true): Unit = {
Review Comment:
Done, thank you. `setSessionConf` is back to master's version, without the
flag. The startup uses a new helper, `OffsetSeqMetadata.setConfWithoutLogging`,
which sets the same values without logging: for the relevant and the rebound
confs, the value in the metadata, or the default of a conf the metadata does
not have.
The new test "setConfWithoutLogging sets the values that setSessionConf
sets" in `OffsetSeqLogSuite` compares the two on the metadata of an old
checkpoint, of a 4.2 checkpoint, and of a new query with a rebound conf,
starting from session values that the metadata replaces or keeps. A later
change to one of them that the other misses fails there.
--
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]
