You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@spark.apache.org by "johanl-db (via GitHub)" <gi...@apache.org> on 2026/04/09 13:29:01 UTC

[PR] [PROTOTYPE][DO_NOT_MERGE] Streaming write schema evolution [spark]

johanl-db opened a new pull request, #55279:
URL: https://github.com/apache/spark/pull/55279

   
   ### What changes were proposed in this pull request?
   Path-finding: attempting to automatically evolve the schema of the sink during streaming writes when the sink table reports capability `AUTOMATIC_SCHEMA_EVOLUTION`
   
   This is a crude prototype and doesn't consider implications that evolving the sink schema may have on streaming semantics.
   
   There's also no user-surface to enable schema evolution on a stream today. Delta uses for example `.option("mergeSchema", "true")`, but Spark itself doesn't provide an a user surface for schema evolution. In this change, we assume that a connector implementation will intercept the `WriteToMicroBatchDataSource` node to enable schema evolution on it as needed
   


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-56470][SQL] Schema evolution for DSv2 sinks during streaming write [spark]

Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.

szehon-ho commented on code in PR #55279:
URL: https://github.com/apache/spark/pull/55279#discussion_r3089424205


##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -8192,6 +8192,23 @@
     ],
     "sqlState" : "0A000"
   },
+  "UNSUPPORTED_STREAMING_SCHEMA_EVOLUTION" : {

Review Comment:
   Missing `sqlState`. Other `UNSUPPORTED_*` errors use `"sqlState" : "0A000"`.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/WriteToMicroBatchDataSource.scala:
##########
@@ -29,19 +32,59 @@ import org.apache.spark.sql.streaming.OutputMode
  * Note that this logical plan does not have a corresponding physical plan, as it will be converted
  * to [[org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2 WriteToDataSourceV2]]
  * with [[MicroBatchWrite]] before execution.
+ *
+ * @param withSchemaEvolution Whether to evolve the sink table schema to match the source.
  */
 case class WriteToMicroBatchDataSource(
     relation: Option[DataSourceV2Relation],
-    table: SupportsWrite,
+    sinkTable: SupportsWrite,
     query: LogicalPlan,
     queryId: String,
     writeOptions: Map[String, String],
     outputMode: OutputMode,
+    override val withSchemaEvolution: Boolean,
     batchId: Option[Long] = None)
-  extends UnaryNode {
+  extends UnaryNode with WriteWithSchemaEvolution {
   override def child: LogicalPlan = query
   override def output: Seq[Attribute] = Nil
 
+  final override val nodePatterns = Seq(COMMAND)
+
+  override def table: LogicalPlan = relation.getOrElse {
+    throw new IllegalStateException(
+      "Cannot access table for schema evolution: no DataSourceV2Relation is set.")
+  }
+
+  override lazy val schemaEvolutionReady: Boolean =
+    relation.exists(_.resolved) && query.resolved
+
+  override def pendingSchemaChanges: Seq[TableChange] = {

Review Comment:
   After `ResolveSchemaEvolution` fires on the first batch, the updated plan (with the new `relation` and `sinkTable`) is only used for that analysis pass. The base `MicroBatchExecution.logicalPlan` (a `lazy val`) still holds the original stale references, so subsequent batches derive from the old plan.
   
   `pendingSchemaChanges` correctly reloads from catalog and returns empty — no duplicate evolution. But the `sinkTable` used by `V2Writes` is still the old instance. For `InMemoryTableCatalog` this means the second batch writes to the old table object, which the test acknowledges.
   
   How does this work for production catalogs? If the assumption is that the `SupportsWrite` object is stateless/shared-storage-backed, it would be worth documenting that here.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/WriteToMicroBatchDataSource.scala:
##########
@@ -29,19 +32,59 @@ import org.apache.spark.sql.streaming.OutputMode
  * Note that this logical plan does not have a corresponding physical plan, as it will be converted
  * to [[org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2 WriteToDataSourceV2]]
  * with [[MicroBatchWrite]] before execution.
+ *
+ * @param withSchemaEvolution Whether to evolve the sink table schema to match the source.
  */
 case class WriteToMicroBatchDataSource(
     relation: Option[DataSourceV2Relation],
-    table: SupportsWrite,
+    sinkTable: SupportsWrite,
     query: LogicalPlan,
     queryId: String,
     writeOptions: Map[String, String],
     outputMode: OutputMode,
+    override val withSchemaEvolution: Boolean,
     batchId: Option[Long] = None)
-  extends UnaryNode {
+  extends UnaryNode with WriteWithSchemaEvolution {
   override def child: LogicalPlan = query
   override def output: Seq[Attribute] = Nil
 
+  final override val nodePatterns = Seq(COMMAND)
+
+  override def table: LogicalPlan = relation.getOrElse {

Review Comment:
   This `IllegalStateException` is user-reachable if someone calls `.withSchemaEvolution()` with a path-based write (`.format("delta").start(path)`) instead of `.toTable()` — `catalogAndIdent` is `None`, so `relation` is `None`.
   
   In practice, short-circuit evaluation in `pendingSchemaChanges` currently prevents this (it checks `relation.isEmpty` first). But `schemaEvolutionEnabled` is a public `lazy val` on the trait that any future code could access directly. Safer to handle `relation.isEmpty` gracefully here — either return a sentinel that makes `schemaEvolutionEnabled` false, or throw a proper Spark error class telling the user schema evolution requires `.toTable()`. A test for this path would be good too.



##########
sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala:
##########
@@ -216,14 +217,21 @@ class StreamingQueryManager private[sql] (
       trigger,
       analyzedPlan,
       catalogAndIdent,
-      catalogTable)
+      catalogTable,
+      withSchemaEvolution)
 
     val analyzedStreamWritePlan =
       sparkSession.sessionState.executePlan(dataStreamWritePlan).analyzed
         .asInstanceOf[WriteToStream]
 
     (sink, trigger) match {
       case (_: SupportsWrite, trigger: ContinuousTrigger) =>
+        if (withSchemaEvolution) {

Review Comment:
   This error is thrown synchronously (user gets it from `.toTable()`), but the `NOT_V2_TABLE` error in `MicroBatchExecution.logicalPlan` surfaces asynchronously as a wrapped `StreamingQueryException`. Could the V1 `Sink` check be moved here too? The `sink` object is already available.



##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala:
##########
@@ -170,6 +170,19 @@ abstract class DataStreamWriter[T] extends WriteConfigMethods[DataStreamWriter[T
     foreachBatch((batchDs: Dataset[T], batchId: Long) => function.call(batchDs, batchId))
   }
 
+  /**
+   * Enables automatic schema evolution for the streaming write. When enabled, if the source
+   * schema has columns not present in the sink table (or type changes), the sink table schema
+   * will be evolved to accommodate the new schema before data is written. The sink table must
+   * support the `AUTOMATIC_SCHEMA_EVOLUTION` capability.
+   *
+   * Schema evolution is applied at query analysis time: when the streaming query is started
+   * (or restarted after failure), the table schema is evolved if needed.
+   *
+   * @since 4.2.0
+   */
+  def withSchemaEvolution(): this.type

Review Comment:
   Worth considering `@Evolving` annotation since the batch equivalent was added recently (#54488) and behavior may still evolve.



##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StreamingSchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,696 @@
+/*
+ * 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.connector
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.SparkUnsupportedOperationException
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.InMemoryTableCatalog
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.streaming.{OutputMode, StreamTest, Trigger}
+import org.apache.spark.sql.types._
+
+/**
+ * Tests for schema evolution in streaming writes using DataSourceV2.
+ *
+ * Schema evolution happens at query analysis time: when a streaming query is started (or
+ * restarted) and the source schema has columns not present in the sink table, the table is
+ * evolved to include those columns before any data is written.
+ */
+class StreamingSchemaEvolutionSuite
+    extends StreamTest with BeforeAndAfter {
+
+  import testImplicits._
+
+  private val catalogName = "testcat"
+  private val namespace = "ns"
+  private val tableIdent = s"$catalogName.$namespace.test_table"
+
+  before {
+    spark.conf.set(
+      s"spark.sql.catalog.$catalogName", classOf[InMemoryTableCatalog].getName)
+    sql(s"CREATE NAMESPACE IF NOT EXISTS $catalogName.$namespace")
+  }
+
+  after {
+    spark.sessionState.catalogManager.reset()
+    spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$catalogName")
+  }
+
+  test("streaming write with extra source column adds column to table") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input.addData((1, "a", 10.0), (2, "b", 20.0))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a", 10.0), Row(2, "b", 20.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("streaming write with matching schema - no evolution needed") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String)]
+        val df = input.toDF().toDF("id", "data")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input.addData((1, "a"), (2, "b"))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a"), Row(2, "b")))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType))))
+      }
+    }
+  }
+
+  test("streaming write evolves schema then processes multiple batches") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          // First batch triggers schema evolution.
+          input.addData((1, "a", 10.0))
+          query.processAllAvailable()
+
+          // Second batch should work without re-triggering evolution.
+          // Note: InMemoryTableCatalog creates new table instances on alterTable,
+          // so the second batch writes to the old instance. We just verify
+          // that the second batch completes without errors.

Review Comment:
   This test only verifies the second batch completes without errors but doesn't check the written data. If the stale-table concern is addressed, this should be strengthened to verify data end-to-end.



##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StreamingSchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,696 @@
+/*
+ * 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.connector
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.SparkUnsupportedOperationException
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.InMemoryTableCatalog
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.streaming.{OutputMode, StreamTest, Trigger}
+import org.apache.spark.sql.types._
+
+/**
+ * Tests for schema evolution in streaming writes using DataSourceV2.
+ *
+ * Schema evolution happens at query analysis time: when a streaming query is started (or
+ * restarted) and the source schema has columns not present in the sink table, the table is
+ * evolved to include those columns before any data is written.
+ */
+class StreamingSchemaEvolutionSuite
+    extends StreamTest with BeforeAndAfter {
+
+  import testImplicits._
+
+  private val catalogName = "testcat"
+  private val namespace = "ns"
+  private val tableIdent = s"$catalogName.$namespace.test_table"
+
+  before {
+    spark.conf.set(
+      s"spark.sql.catalog.$catalogName", classOf[InMemoryTableCatalog].getName)
+    sql(s"CREATE NAMESPACE IF NOT EXISTS $catalogName.$namespace")
+  }
+
+  after {
+    spark.sessionState.catalogManager.reset()
+    spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$catalogName")
+  }
+
+  test("streaming write with extra source column adds column to table") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input.addData((1, "a", 10.0), (2, "b", 20.0))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a", 10.0), Row(2, "b", 20.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("streaming write with matching schema - no evolution needed") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String)]
+        val df = input.toDF().toDF("id", "data")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input.addData((1, "a"), (2, "b"))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a"), Row(2, "b")))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType))))
+      }
+    }
+  }
+
+  test("streaming write evolves schema then processes multiple batches") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          // First batch triggers schema evolution.
+          input.addData((1, "a", 10.0))
+          query.processAllAvailable()
+
+          // Second batch should work without re-triggering evolution.
+          // Note: InMemoryTableCatalog creates new table instances on alterTable,
+          // so the second batch writes to the old instance. We just verify
+          // that the second batch completes without errors.
+          input.addData((2, "b", 20.0))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("streaming write with multiple extra columns") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input.addData((1, "a", 10.0))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a", 10.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("table without AUTOMATIC_SCHEMA_EVOLUTION capability - schema not evolved") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(
+          s"""CREATE TABLE $tableIdent (id INT, data STRING)
+             |TBLPROPERTIES ('auto-schema-evolution' = 'false')""".stripMargin)
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input.addData((1, "a", 10.0))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        // Schema should NOT have been evolved since the table lacks the capability.
+        val result = spark.table(tableIdent)
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType))))
+      }
+    }
+  }
+
+  test("streaming write without withSchemaEvolution - schema not evolved") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+
+        // No .withSchemaEvolution() call.
+        val query = df.writeStream
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input.addData((1, "a", 10.0))
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        // Schema should NOT have been evolved since withSchemaEvolution was not called.
+        val result = spark.table(tableIdent)
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType))))
+      }
+    }
+  }
+
+  test("streaming restart after schema evolution preserves data") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        // First query: write with matching schema.
+        val input1 = MemoryStream[(Int, String)]
+        val df1 = input1.toDF().toDF("id", "data")
+
+        val query1 = df1.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input1.addData((1, "a"), (2, "b"))
+          query1.processAllAvailable()
+        } finally {
+          query1.stop()
+        }
+
+        checkAnswer(spark.table(tableIdent), Seq(Row(1, "a"), Row(2, "b")))
+
+        // Second query: write with extra column, triggers schema evolution.
+        val input2 = MemoryStream[(Int, String, Double)]
+        val df2 = input2.toDF().toDF("id", "data", "amount")
+
+        val query2 = df2.writeStream
+          .withSchemaEvolution()
+          .option(
+            "checkpointLocation",
+            s"${checkpointDir.getCanonicalPath}_2")
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input2.addData((3, "c", 30.0))
+          query2.processAllAvailable()
+        } finally {
+          query2.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(
+          Row(1, "a", null),
+          Row(2, "b", null),
+          Row(3, "c", 30.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("schema evolution on restart after sink table altered between batches") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        // First query: write with matching schema.
+        val input1 = MemoryStream[(Int, String)]
+        val df1 = input1.toDF().toDF("id", "data")
+
+        val query1 = df1.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input1.addData((1, "a"))
+          query1.processAllAvailable()
+        } finally {
+          query1.stop()
+        }
+
+        // Alter the table externally to add a column, simulating a schema change
+        // that happened while the query was down.
+        sql(s"ALTER TABLE $tableIdent ADD COLUMN amount DOUBLE")
+
+        val input2 = MemoryStream[(Int, String, Double)]
+        val df2 = input2.toDF().toDF("id", "data", "amount")
+
+        val query2 = df2.writeStream
+          .withSchemaEvolution()
+          .option(
+            "checkpointLocation",
+            s"${checkpointDir.getCanonicalPath}_2")
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input2.addData((2, "b", 20.0))
+          query2.processAllAvailable()
+        } finally {
+          query2.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("schema evolution with Trigger.Once") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+        input.addData((1, "a", 10.0), (2, "b", 20.0))
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .trigger(Trigger.Once())
+          .toTable(tableIdent)
+
+        try {
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a", 10.0), Row(2, "b", 20.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("schema evolution with Trigger.AvailableNow") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        val input = MemoryStream[(Int, String, Double)]
+        val df = input.toDF().toDF("id", "data", "amount")
+        input.addData((1, "a", 10.0))
+
+        val query = df.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .trigger(Trigger.AvailableNow())
+          .toTable(tableIdent)
+
+        try {
+          query.processAllAvailable()
+        } finally {
+          query.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a", 10.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("incremental schema evolution across multiple restarts") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT)")
+
+        // First query: add "data" column.
+        val input1 = MemoryStream[(Int, String)]
+        val df1 = input1.toDF().toDF("id", "data")
+
+        val query1 = df1.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input1.addData((1, "a"))
+          query1.processAllAvailable()
+        } finally {
+          query1.stop()
+        }
+
+        assert(spark.table(tableIdent).schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType))))
+
+        // Second query: add "amount" column on top of the already-evolved schema.
+        val input2 = MemoryStream[(Int, String, Double)]
+        val df2 = input2.toDF().toDF("id", "data", "amount")
+
+        val query2 = df2.writeStream
+          .withSchemaEvolution()
+          .option(
+            "checkpointLocation",
+            s"${checkpointDir.getCanonicalPath}_2")
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input2.addData((2, "b", 20.0))
+          query2.processAllAvailable()
+        } finally {
+          query2.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(
+          Row(1, "a", null),
+          Row(2, "b", 20.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+
+        // Third query: no new columns - schema should stay the same.
+        val input3 = MemoryStream[(Int, String, Double)]
+        val df3 = input3.toDF().toDF("id", "data", "amount")
+
+        val query3 = df3.writeStream
+          .withSchemaEvolution()
+          .option(
+            "checkpointLocation",
+            s"${checkpointDir.getCanonicalPath}_3")
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input3.addData((3, "c", 30.0))
+          query3.processAllAvailable()
+        } finally {
+          query3.stop()
+        }
+
+        val result2 = spark.table(tableIdent)
+        assert(result2.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+  test("stop and restart same query - schema evolved on restart") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT, data STRING)")
+
+        // First run: matching schema, no evolution.
+        val input1 = MemoryStream[(Int, String)]
+        val df1 = input1.toDF().toDF("id", "data")
+
+        val query1 = df1.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input1.addData((1, "a"))
+          query1.processAllAvailable()
+          input1.addData((2, "b"))
+          query1.processAllAvailable()
+        } finally {
+          query1.stop()
+        }
+
+        checkAnswer(spark.table(tableIdent), Seq(Row(1, "a"), Row(2, "b")))
+
+        // Second run: new source schema with extra column.
+        // Uses a different checkpoint since MemoryStream state can't be
+        // reused across restarts with a different schema.
+        val input2 = MemoryStream[(Int, String, Double)]
+        val df2 = input2.toDF().toDF("id", "data", "amount")
+
+        val query2 = df2.writeStream
+          .withSchemaEvolution()
+          .option(
+            "checkpointLocation",
+            s"${checkpointDir.getCanonicalPath}_2")
+          .outputMode(OutputMode.Append())
+          .toTable(tableIdent)
+
+        try {
+          input2.addData((3, "c", 30.0))
+          query2.processAllAvailable()
+        } finally {
+          query2.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(
+          Row(1, "a", null),
+          Row(2, "b", null),
+          Row(3, "c", 30.0)))
+      }
+    }
+  }
+
+  test("schema evolution with Trigger.Once across restart") {
+    withTable(tableIdent) {
+      withTempDir { checkpointDir =>
+        sql(s"CREATE TABLE $tableIdent (id INT)")
+
+        // First run: Trigger.Once with extra column.
+        val input1 = MemoryStream[(Int, String)]
+        val df1 = input1.toDF().toDF("id", "data")
+        input1.addData((1, "a"))
+
+        val query1 = df1.writeStream
+          .withSchemaEvolution()
+          .option("checkpointLocation", checkpointDir.getCanonicalPath)
+          .trigger(Trigger.Once())
+          .toTable(tableIdent)
+
+        try {
+          query1.processAllAvailable()
+        } finally {
+          query1.stop()
+        }
+
+        assert(spark.table(tableIdent).schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType))))
+
+        // Second run: Trigger.Once with yet another extra column.
+        val input2 = MemoryStream[(Int, String, Double)]
+        val df2 = input2.toDF().toDF("id", "data", "amount")
+        input2.addData((2, "b", 20.0))
+
+        val query2 = df2.writeStream
+          .withSchemaEvolution()
+          .option(
+            "checkpointLocation",
+            s"${checkpointDir.getCanonicalPath}_2")
+          .trigger(Trigger.Once())
+          .toTable(tableIdent)
+
+        try {
+          query2.processAllAvailable()
+        } finally {
+          query2.stop()
+        }
+
+        val result = spark.table(tableIdent)
+        checkAnswer(result, Seq(Row(1, "a", null), Row(2, "b", 20.0)))
+        assert(result.schema == StructType(Seq(
+          StructField("id", IntegerType),
+          StructField("data", StringType),
+          StructField("amount", DoubleType))))
+      }
+    }
+  }
+
+
+  test("withSchemaEvolution rejected with continuous trigger") {

Review Comment:
   nit: extra blank line



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org