dongjoon-hyun commented on code in PR #59943: URL: https://github.com/apache/spark/pull/59943#discussion_r4489745232
########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,445 @@ +/* + * 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 +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet, Cast, Expression, Generator, NamedExpression, OuterReference, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction +import org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions +import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef, LeafNode, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.trees.TreePattern.{CURRENT_LIKE, RUNTIME_REPLACEABLE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +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.errors.QueryExecutionErrors +import org.apache.spark.sql.execution.QueryExecution +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{AnyTimeType, DataType, DatetimeType, StringType, StructType, VariantType} +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. A scan cannot read other columns + * once its stream exists, so a micro-batch that does fails with STREAMING_SCAN_MISSING_COLUMNS, + * as when the configuration of the optimizer changes in the session of the query after it starts + * so that its plan reads other columns of a source. + * + * 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 { + + /** + * The scan of a streaming source, and its stream. + * + * @param scanOutput the columns of the relation that the scan reads, in the order of its read + * schema, unless these are all its columns in their order + */ + case class StreamingScan( + scan: Scan, + stream: SparkDataStream, + scanOutput: Option[Seq[AttributeReference]]) + + /** + * Returns the columns that each streaming relation of Data Source V2 needs to read, from `plan` + * optimized as the micro-batches are (see `forStartup`). A relation without an entry reads all + * its columns, as all do if the optimization fails. + * + * @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 requiredColumnsOfSources( + session: SparkSession, + plan: LogicalPlan, + offsetLogMetadata: Option[OffsetSeqMetadataBase]): Map[StreamingRelationV2, AttributeSet] = { + val conf = session.sessionState.conf.clone() + offsetLogMetadata.foreach(OffsetSeqMetadata.setSessionConf(_, conf, logChanges = false)) + conf.setConf(SQLConf.CBO_ENABLED, false) + session.withActive { Review Comment: This runs the whole optimizer (plus the `transformUpWithSubqueries` pass below) on every query start, and on every reconfiguration in continuous processing, whenever a source has the capability. Is the extra startup latency measured? If it can be large for big plans, it may be worth documenting. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/columnar/ColumnarTable.scala: ########## @@ -61,8 +61,14 @@ class ColumnarTable( override def schema(): StructType = tableSchema - private lazy val tableCapabilities = - createDataSource(withTableOptions(CaseInsensitiveStringMap.empty())).capabilities() + // The streams of the scans of the table return the columns of their read schemas: their stream + // readers get these columns, and return exactly them, see ColumnarScan. + private lazy val tableCapabilities = { + val capabilities = new util.HashSet[TableCapability]( + createDataSource(withTableOptions(CaseInsensitiveStringMap.empty())).capabilities()) + capabilities.add(TableCapability.STREAMING_COLUMN_PRUNING) Review Comment: This adds `STREAMING_COLUMN_PRUNING` for every source wrapped by `ColumnarTable`, relying on `ColumnarScan` returning the pruned read schema from the stream readers. If a wrapped source's reader returns the full schema, the scan node would see a wrong column count/ordinals. Is there a check (or a test) that guarantees the wrapped readers honor the pruned `readSchema`? ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,445 @@ +/* + * 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 +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet, Cast, Expression, Generator, NamedExpression, OuterReference, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction +import org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions +import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef, LeafNode, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.trees.TreePattern.{CURRENT_LIKE, RUNTIME_REPLACEABLE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +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.errors.QueryExecutionErrors +import org.apache.spark.sql.execution.QueryExecution +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{AnyTimeType, DataType, DatetimeType, StringType, StructType, VariantType} +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. A scan cannot read other columns + * once its stream exists, so a micro-batch that does fails with STREAMING_SCAN_MISSING_COLUMNS, + * as when the configuration of the optimizer changes in the session of the query after it starts + * so that its plan reads other columns of a source. + * + * 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 { + + /** + * The scan of a streaming source, and its stream. + * + * @param scanOutput the columns of the relation that the scan reads, in the order of its read + * schema, unless these are all its columns in their order + */ + case class StreamingScan( + scan: Scan, + stream: SparkDataStream, + scanOutput: Option[Seq[AttributeReference]]) + + /** + * Returns the columns that each streaming relation of Data Source V2 needs to read, from `plan` + * optimized as the micro-batches are (see `forStartup`). A relation without an entry reads all + * its columns, as all do if the optimization fails. + * + * @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 requiredColumnsOfSources( + session: SparkSession, + plan: LogicalPlan, + offsetLogMetadata: Option[OffsetSeqMetadataBase]): Map[StreamingRelationV2, AttributeSet] = { + val conf = session.sessionState.conf.clone() Review Comment: The pruned columns are fixed once at query start from this simulated run, but each micro-batch is planned and optimized for real. Anything the simulation does not model (AQE, conf changes between batches, rules that behave differently per batch, static subtrees) can make a later batch need a column the scan no longer reads, which fails with `STREAMING_SCAN_MISSING_COLUMNS` even though the user changed nothing. Since `spark.sql.streaming.v2ColumnPruning` defaults to enabled, would it be safer to default it to disabled initially, or to fall back to the full scan instead of failing when a column is missing? ########## 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: Nit: `logChanges` adds a flag and a duplicate `case None` branch to shared checkpoint code, only to keep the throwaway conf clone quiet. Would a small helper that applies the metadata to the conf without logging be simpler, so a future branch cannot forget the flag? ########## sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingV2ScanPushDown.scala: ########## @@ -0,0 +1,445 @@ +/* + * 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 +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet, Cast, Expression, Generator, NamedExpression, OuterReference, SubqueryExpression, Unevaluable} +import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction +import org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions +import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef, LeafNode, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2 +import org.apache.spark.sql.catalyst.trees.TreePattern.{CURRENT_LIKE, RUNTIME_REPLACEABLE} +import org.apache.spark.sql.catalyst.types.DataTypeUtils +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +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.errors.QueryExecutionErrors +import org.apache.spark.sql.execution.QueryExecution +import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetSeqMetadata, OffsetSeqMetadataBase} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{AnyTimeType, DataType, DatetimeType, StringType, StructType, VariantType} +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. A scan cannot read other columns + * once its stream exists, so a micro-batch that does fails with STREAMING_SCAN_MISSING_COLUMNS, + * as when the configuration of the optimizer changes in the session of the query after it starts + * so that its plan reads other columns of a source. + * + * 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 { + + /** + * The scan of a streaming source, and its stream. + * + * @param scanOutput the columns of the relation that the scan reads, in the order of its read + * schema, unless these are all its columns in their order + */ + case class StreamingScan( + scan: Scan, + stream: SparkDataStream, + scanOutput: Option[Seq[AttributeReference]]) + + /** + * Returns the columns that each streaming relation of Data Source V2 needs to read, from `plan` + * optimized as the micro-batches are (see `forStartup`). A relation without an entry reads all + * its columns, as all do if the optimization fails. + * + * @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 requiredColumnsOfSources( + session: SparkSession, + plan: LogicalPlan, + offsetLogMetadata: Option[OffsetSeqMetadataBase]): Map[StreamingRelationV2, AttributeSet] = { + 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 { + // The plan of a micro-batch is normalized before it is optimized, see `QueryExecution`. + val normalized = QueryExecution.normalize(session, plan) + val optimized = session.sessionState.optimizer.execute(forStartup(normalized)) + // 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, AttributeSet)] + // 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 ScanOfStream(project, filters, relation: StreamingRelationV2) => + occurrences += relation -> 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 { + // The scan of a relation reads the rows of all its occurrences, so it reads the + // columns that any of them reads. + occurrences.toSeq.groupMap(_._1)(_._2).map { case (relation, columns) => + relation -> AttributeSet.fromAttributeSets(columns) + } + } + } 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 `plan` for the optimization when the query starts, so that the columns that it reads + * from its streaming relations do not depend on when it is optimized: + * - The subtrees without a streaming relation, such as the static side of a join or the plan + * of a static common table expression, and the subtrees of the plans of subqueries without + * an outer reference, become placeholders of their outputs, so that nothing of them folds + * into the plan: a micro-batch applies the cache, which may change after the query starts, + * and their scans are neither built nor planned. + * - The expressions whose values may depend on when they are evaluated or on the state of the + * session, such as the current time or the conversions to dates and timestamps (see + * `makesDateOrTimestamp`), become placeholders that are not deterministic for the optimizer, + * as the current time of a micro-batch, and keep their children. + * Visible for tests. + */ + private[sql] def forStartup(plan: LogicalPlan): LogicalPlan = withOpaqueSubqueries( + plan.transformDown { + // `WithCTE` keeps the definitions of its common table expressions as they are. + case definition: CTERelationDef if !definition.isStreaming => + definition.copy(child = BatchRelation(definition.child.output)) + case node if !node.isStreaming => BatchRelation(node.output) + }).transformUpWithSubqueries { + case node => + // The expressions that ReplaceExpressions replaces with others when the optimization starts + // are replaced first, as the optimizer does, so that the conversions that they stand for, + // such as the cast of to_date('today'), are replaced below. + val replaced = if (node.containsPattern(RUNTIME_REPLACEABLE)) { + node.mapExpressions(ReplaceExpressions.replace) + } else { + node + } + replaced.transformExpressionsUp { + // 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. + case e if e.containsPattern(CURRENT_LIKE) => StartupValue(e, e.children) + case e if makesDateOrTimestamp(e) => StartupValue(e, e.children) + } + } + + /** + * Returns `plan` with the subtrees of the plans of its subqueries that have no outer reference + * replaced by placeholders of their outputs, also in the subqueries of the others. The others + * keep the columns of the plan that they reference, which the plan reads. + */ + private def withOpaqueSubqueries(plan: LogicalPlan): LogicalPlan = plan.transformUp { + case node => node.transformExpressions { + case subquery: SubqueryExpression => + subquery.withNewPlan(withOpaqueSubqueries(subquery.plan.transformDown { + case child if !hasOuterReferences(child) => BatchRelation(child.output) + })) + } + } + + // Whether `plan` references the columns of an outer query, also in its subqueries. + private def hasOuterReferences(plan: LogicalPlan): Boolean = plan.exists { node => + node.expressions.exists(_.exists { + case _: OuterReference => true + case subquery: SubqueryExpression => hasOuterReferences(subquery.plan) + case _ => false + }) + } + + /** + * Matches the projections and the filters over a relation that the planner plans with the scan + * of the relation, as `DataSourceV2Strategy` does for the scans of streaming sources, so that + * the columns that a scan reads are those that its micro-batches plan. + */ + private[sql] object ScanOfStream { + def unapply(plan: LogicalPlan): Option[(Seq[NamedExpression], Seq[Expression], LogicalPlan)] = + PhysicalOperation.unapply(plan) + } + + /** + * Returns whether `expression` may make a date or a timestamp of a value without a date, which + * takes the date of when it is evaluated: a string with only a time such as '12:00:00', a + * special string such as 'today' or 'now', a time, or a variant with such a string. The + * optimizer evaluates such an expression once its children are constant, which they may only + * become during the optimization, as when an alias or an equality propagates a string, so the + * expression counts whatever its children are, if its type has a date or a timestamp, of any + * precision and time zone, at any depth: also the elements of an array, the keys or the values + * of a map, or the fields of a struct, and the type of one of its children has a string, a time + * or a variant at any depth. + * + * These are the casts, and the other conversions, such as `from_json` or `variant_get`, which + * only count if no child has a date or a timestamp: the expressions that have one, such as + * `named_struct`, `element_at` or an alias, pass on the dates and the timestamps of their + * children, whose own conversions are replaced. The expressions that the optimizer does not + * evaluate, such as aggregate functions or generators, which their operators need as they are, + * do not count. + */ + private def makesDateOrTimestamp(expression: Expression): Boolean = expression match { Review Comment: This heuristic is a hand-maintained list of the expressions that must not be folded at startup. Expressions that depend on session state but do not match it (for example `CASE WHEN current_timezone() = 'UTC' THEN a ELSE b END`) can be folded at startup and prune a branch the per-batch plan still uses. Could we avoid folding during this simulation instead (for example, only run ColumnPruning-style rules), so the result does not depend on matching every such expression? If not, please add tests for expressions of this kind. ########## 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: Nit: the source here is `catalog.name +: namespace :+ name`, i.e. one element per identifier part, while `streamingScanInvalidReadSchemaError` is called with `table.name()` as a single name. The same table may therefore be shown as `` `cat`.`db`.`t` `` in one error and as `` `cat.db.t` `` in the other. Could the two errors use the same format? -- 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]
