1

Author: aakash-db

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/CoreDataflowNodeProcessor.scala

@@ -87,10 +87,8 @@ class CoreDataflowNodeProcessor(rawGraph: DataflowGraph) {
           isStreamingTableOpt = Option(resolvedFlowsToTable.exists(f => f.df.isStreaming))
         )
 
-        // Table will be virtual in either of the following scenarios:
-        // 1. If table is present in context.fullRefreshTables
-        // 2. If table has any virtual inputs (flows or tables)
-        // 3. If the table pre-existing metadata is different from current metadata
+        // We mark all tables as virtual to ensure resolution uses incoming flows
+        // rather than previously materialized tables.
         val virtualTableInput = VirtualTableInput(
           identifier = table.identifier,
           specifiedSchema = table.specifiedSchema,

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala

@@ -37,11 +37,6 @@ case class DataflowGraph(flows: Seq[Flow], tables: Seq[Table], views: Seq[View])
   /** Returns a [[Output]] given its identifier */
   lazy val output: Map[TableIdentifier, Output] = mapUnique(tables, "output")(_.identifier)
 
-  /**
-   * Returns a [[TableInput]], if one is available, that can be read from by downstream flows.
-   */
-  def tableInput(identifier: TableIdentifier): Option[TableInput] = table.get(identifier)
-
   /**
    * Returns [[Flow]]s in this graph that need to get planned and potentially executed when
    * executing the graph. Flows that write to logical views are excluded.

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala

@@ -148,7 +148,7 @@ trait GraphValidations extends Logging {
   }
 
   protected def validateUserSpecifiedSchemas(): Unit = {
-    flows.flatMap(f => tableInput(f.identifier)).foreach { t: TableInput =>
+    flows.flatMap(f => table.get(f.identifier)).foreach { t: TableInput =>
       // The output inferred schema of a table is the declared schema merged with the
       // schema of all incoming flows. This must be equivalent to the declared schema.
       val inferredSchema = SchemaInferenceUtils