[WIP] Sequential Union PoC

sql/api/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -1960,6 +1960,33 @@ abstract class Dataset[T] extends Serializable {
    */
   def unionByName(other: Dataset[T], allowMissingColumns: Boolean): Dataset[T]
 
+  /**
+   * Returns a Dataset containing rows from this Dataset followed sequentially by rows from
+   * another Dataset. Unlike `union` which processes both datasets concurrently, this method
+   * processes this Dataset completely before starting the other Dataset.
+   *
+   * This is useful for scenarios like processing historical data followed by live streaming data.
+   * For example:
+   * {{{
+   *   val historical = spark.readStream.format("json")
+   *     .option("maxFilesPerTrigger", "100")  // Makes it bounded
+   *     .load("/historical-data")
+   *
+   *   val live = spark.readStream.format("kafka")
+   *     .option("subscribe", "events")
+   *     .load()
+   *
+   *   val sequential = historical.followedBy(live)
+   *   // Processes all historical files first, then transitions to live Kafka
+   * }}}
+   *
+   * @param other Another Dataset to append after this one completes
+   * @return A new Dataset with sequential union semantics
+   * @group typedrel
+   * @since 4.0.0
+   */
+  def followedBy(other: Dataset[T]): Dataset[T]
+
   /**
    * Returns a new Dataset containing rows only in both this Dataset and another Dataset. This is
    * equivalent to `INTERSECT` in SQL.

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/streaming/SupportsSequentialExecution.java

@@ -0,0 +1,42 @@
+/*
+ * 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.catalyst.streaming;
+
+/**
+ * Interface for streaming sources that can signal completion for sequential execution.
+ * This enables sources to indicate when they have finished processing their bounded data,
+ * allowing sequential union operations to transition to the next source.
+ * 
+ * Example use cases:
+ * - File streams with maxFilesPerTrigger option that need to signal when all files are processed
+ * - Bounded Kafka streams that process a specific range of offsets
+ * - Any source that has a finite amount of data to process
+ */
+public interface SupportsSequentialExecution {
+
+  /**
+   * Returns true if this source has finished processing all available data.
+   * This is typically used for bounded sources (like file streams with maxFilesPerTrigger)
+   * to indicate when they should be considered complete in sequential execution scenarios.
+   * 
+   * For unbounded sources (like continuous Kafka streams), this should always return false.
+   * 
+   * @return true if the source has completed processing all its data, false otherwise
+   */
+  boolean isSourceComplete();
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveUnion.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.analysis
 import scala.collection.mutable
 
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, Union}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, SequentialUnion, Union}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.UNION
 import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -210,5 +210,11 @@ object ResolveUnion extends Rule[LogicalPlan] {
         checkColumnNames(left, right)
         unionTwoSides(left, right, allowMissingCol)
       }
+
+    case SequentialUnion(children, byName, allowMissingCol) if byName =>
+      children.reduceLeft { (left, right) =>
+        checkColumnNames(left, right)
+        unionTwoSides(left, right, allowMissingCol)
+      }
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -635,6 +635,74 @@ case class Union(
     copy(children = newChildren)
 }
 
+/**
+ * Sequential Union processes children sequentially rather than concurrently.
+ * Each child must complete before the next child begins execution.
+ *
+ * Unlike Union which processes all children in parallel, SequentialUnion is designed
+ * for streaming scenarios where you want to process bounded data first (e.g., historical files)
+ * followed by unbounded data (e.g., live streams).
+ *
+ * Schema validation and resolution follow the same rules as Union.
+ */
+case class SequentialUnion(
+    children: Seq[LogicalPlan],
+    byName: Boolean = false,
+    allowMissingCol: Boolean = false) extends UnionBase {
+
+  require(children.length >= 2, "SequentialUnion requires at least two children")
+  assert(!allowMissingCol || byName, "`allowMissingCol` can be true only if `byName` is true.")
+
+  override def maxRows: Option[Long] = {
+    // Sequential execution means sum of all children's rows
+    var sum = BigInt(0)
+    children.foreach { child =>
+      if (child.maxRows.isDefined) {
+        sum += child.maxRows.get
+        if (!sum.isValidLong) {
+          return None
+        }
+      } else {
+        return None
+      }
+    }
+    Some(sum.toLong)
+  }
+
+  final override val nodePatterns: Seq[TreePattern] = Seq(UNION)
+
+  override def maxRowsPerPartition: Option[Long] = {
+    // For sequential execution, only one child is active at a time,
+    // so maxRowsPerPartition is the maximum among all children
+    val childMaxRows = children.flatMap(_.maxRowsPerPartition)
+    if (childMaxRows.length == children.length) {
+      Some(childMaxRows.max)
+    } else {
+      None
+    }
+  }
+
+  private def duplicatesResolvedPerBranch: Boolean =
+    children.forall(child => child.outputSet.size == child.output.size)
+
+  def duplicatesResolvedBetweenBranches: Boolean = {
+    children.map(_.outputSet.size).sum ==
+      AttributeSet.fromAttributeSets(children.map(_.outputSet)).size
+  }
+
+  override lazy val resolved: Boolean = {
+    children.length > 1 &&
+    !(byName || allowMissingCol) &&
+    childrenResolved &&
+    allChildrenCompatible &&
+    (!conf.unionIsResolvedWhenDuplicatesPerChildResolved || duplicatesResolvedPerBranch)
+  }
+
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[LogicalPlan]): SequentialUnion =
+    copy(children = newChildren)
+}
+
 object Join {
   def computeOutput(
     joinType: JoinType,

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/Dataset.scala

@@ -675,6 +675,14 @@ class Dataset[T] private[sql] (
     }
   }
 
+  /** @inheritdoc */
+  def followedBy(other: sql.Dataset[T]): Dataset[T] = {
+    // For Spark Connect, this would need proper protocol support
+    // For now, throw an exception indicating it's not supported
+    throw new UnsupportedOperationException(
+      "followedBy is not yet supported in Spark Connect")
+  }
+
   /** @inheritdoc */
   def intersect(other: sql.Dataset[T]): Dataset[T] = {
     buildSetOp(other, proto.SetOperation.SetOpType.SET_OP_TYPE_INTERSECT) { builder =>

sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala

@@ -1157,6 +1157,11 @@ class Dataset[T] private[sql](
     }
   }
 
+  /** @inheritdoc */
+  def followedBy(other: sql.Dataset[T]): Dataset[T] = withSetOperator {
+    SequentialUnion(logicalPlan :: other.logicalPlan :: Nil)
+  }
+
   /** @inheritdoc */
   def intersect(other: sql.Dataset[T]): Dataset[T] = withSetOperator {
     Intersect(logicalPlan, other.logicalPlan, isAll = false)

sql/core/src/main/scala/org/apache/spark/sql/execution/SequentialUnionExec.scala

@@ -0,0 +1,97 @@
+/*
+ * 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
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, UnknownPartitioning}
+import org.apache.spark.sql.execution.metric.SQLMetrics
+
+/**
+ * Physical operator for sequential union that processes children sequentially rather than
+ * concurrently. Unlike UnionExec which processes all children in parallel, SequentialUnionExec
+ * processes only one child at a time, making sequential execution a first-class citizen.
+ *
+ * This operator maintains internal state to track which child is currently active and transitions
+ * between children based on completion status (for streaming) or processes them sequentially
+ * (for batch).
+ */
+case class SequentialUnionExec(children: Seq[SparkPlan]) extends SparkPlan {
+
+  override def output: Seq[Attribute] = children.head.output
+
+  override def outputPartitioning: Partitioning = UnknownPartitioning(0)
+
+  override lazy val metrics = Map(
+    "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"),
+    "activeChild" -> SQLMetrics.createMetric(sparkContext, "active child index")
+  )
+
+  /**
+   * For true sequential processing, we need to determine which child should be active.
+   * This is a simplified version that executes the first child with data.
+   * In a production implementation, this would coordinate with the streaming execution
+   * to determine the active child based on source completion status.
+   */
+  private def findActiveChild(): Int = {
+    // For now, find the first child that would produce data
+    // In streaming context, SequentialUnionManager ensures only one child has data
+    for (i <- children.indices) {
+      // We could add more sophisticated logic here to peek at data availability
+      // For now, return the first child
+      return i
+    }
+    0 // Default to first child
+  }
+
+  protected override def doExecute(): RDD[InternalRow] = {
+    val numOutputRowsMetric = longMetric("numOutputRows")
+    val activeChildMetric = longMetric("activeChild")
+
+    val activeIndex = findActiveChild()
+    activeChildMetric.set(activeIndex)
+
+    // Execute only the active child - this is true sequential processing
+    val activeChild = if (activeIndex < children.length) {
+      children(activeIndex)
+    } else {
+      // Fallback to first child if something goes wrong
+      children.head
+    }
+
+    val activeRDD = activeChild.execute()
+
+    // Add metric tracking
+    activeRDD.mapPartitions { iter =>
+      val wrappedIter = iter.map { row =>
+        numOutputRowsMetric += 1
+        row
+      }
+      wrappedIter
+    }
+  }
+
+  override def verboseString(maxFields: Int): String = {
+    val activeIndex = findActiveChild()
+    s"SequentialUnion(activeChild=$activeIndex/${children.length})"
+  }
+
+  override protected def withNewChildrenInternal(newChildren: IndexedSeq[SparkPlan]): SparkPlan =
+    copy(children = newChildren)
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

@@ -1049,6 +1049,10 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
         GlobalLimitExec(child = planLater(child), offset = offset) :: Nil
       case union: logical.Union =>
         execution.UnionExec(union.children.map(planLater)) :: Nil
+      case sequentialUnion: logical.SequentialUnion =>
+        // For now, treat SequentialUnion like regular Union at physical level
+        // The sequential semantics are handled at the streaming execution level
+        execution.UnionExec(sequentialUnion.children.map(planLater)) :: Nil
       case u @ logical.UnionLoop(id, anchor, recursion, _, limit, maxDepth) =>
         execution.UnionLoopExec(id, anchor, recursion, u.output, limit, maxDepth) :: Nil
       case g @ logical.Generate(generator, _, outer, _, _, child) =>

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/FileStreamSource.scala

@@ -29,6 +29,7 @@ import org.apache.hadoop.fs.{FileStatus, FileSystem, GlobFilter, Path}
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.internal.{Logging, LogKeys}
 import org.apache.spark.paths.SparkPath
+import org.apache.spark.sql.catalyst.streaming.SupportsSequentialExecution
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.classic.{DataFrame, Dataset, SparkSession}
 import org.apache.spark.sql.connector.read.streaming
@@ -55,6 +56,7 @@ class FileStreamSource(
     options: Map[String, String])
   extends SupportsAdmissionControl
   with SupportsTriggerAvailableNow
+  with SupportsSequentialExecution
   with Source
   with Logging {
 
@@ -304,6 +306,8 @@ class FileStreamSource(
 
     assert(startOffset <= endOffset)
     val files = metadataLog.get(Some(startOffset + 1), Some(endOffset)).flatMap(_._2)
+    logError(s"### FileStreamSource.getBatch: Processing ${files.length} files from " +
+      s"$startOffset to $endOffset")
     logInfo(log"Processing ${MDC(LogKeys.NUM_FILES, files.length)} files from " +
       log"${MDC(LogKeys.FILE_START_OFFSET, startOffset + 1)}:" +
       log"${MDC(LogKeys.FILE_END_OFFSET, endOffset)}")
@@ -431,6 +435,35 @@ class FileStreamSource(
   }
 
   override def stop(): Unit = sourceCleaner.foreach(_.stop())
+
+  /**
+   * Implementation of SupportsSequentialExecution interface.
+   * Returns true if this file source has finished processing all available files
+   * and can be considered "done" for sequential execution purposes.
+   */
+  override def isSourceComplete(): Boolean = {
+    sourceOptions.maxFilesPerTrigger match {
+      case Some(_) =>
+        // For bounded file sources with maxFilesPerTrigger, check if we've processed all
+        // available files. A source is complete when:
+        // 1. We have discovered all available files in the directory
+        // 2. We have processed all of them (seenFiles contains all available files)
+        val availableFiles = fetchAllFiles()
+        val processedFiles = seenFiles.size
+
+        logDebug(s"FileStreamSource.isSourceComplete: " +
+          s"processedFiles=$processedFiles, availableFiles=${availableFiles.length}")
+
+        // Source is complete if we've processed all available files AND there are files to process
+        // (we don't want to consider an empty directory as "complete")
+        availableFiles.nonEmpty && processedFiles >= availableFiles.length
+
+      case None =>
+        // Unbounded file source (no maxFilesPerTrigger) - never done
+        // This represents continuous file monitoring
+        false
+    }
+  }
 }