From 6170a601144f9babced3d77283fcfbe1a9af130c Mon Sep 17 00:00:00 2001
From: You Zhou <you.zhou@databricks.com>
Date: Thu, 28 May 2026 18:51:40 +0000
Subject: [PATCH] Revert "[SPARK-56975][SS] Reject user-specified schema in
 DataStreamReader.table()"

This reverts commit 05b4d81f3f938ff140886d6f66ad66d08c66d5b2.
---
 .../spark/sql/classic/DataStreamReader.scala     |  1 -
 .../streaming/test/DataStreamTableAPISuite.scala | 16 ----------------
 2 files changed, 17 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
index c8df93768808e..eb3120cac05aa 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
@@ -102,7 +102,6 @@ final class DataStreamReader private[sql](sparkSession: SparkSession)
   /** @inheritdoc */
   def table(tableName: String): DataFrame = {
     require(tableName != null, "The table name can't be null")
-    assertNoSpecifiedSchema("table")
     val identifier = sparkSession.sessionState.sqlParser.parseMultipartIdentifier(tableName)
     val unresolved = UnresolvedRelation(
       identifier,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
index 3930beec084dc..e2c74533e7f3c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
@@ -84,22 +84,6 @@ class DataStreamTableAPISuite extends StreamTest with BeforeAndAfter {
     checkErrorTableNotFound(e, "`non_exist_table`")
   }
 
-  test("read: user-specified schema is not allowed with table API") {
-    val tblName = "my_table"
-    withTable(tblName) {
-      spark.range(3).write.format("parquet").saveAsTable(tblName)
-      val e = intercept[AnalysisException] {
-        spark.readStream
-          .schema(new StructType().add("a", IntegerType))
-          .table(tblName)
-      }
-      checkError(
-        exception = e,
-        condition = "_LEGACY_ERROR_TEMP_1189",
-        parameters = Map("operation" -> "table"))
-    }
-  }
-
   test("read: stream table API with temp view") {
     val tblName = "my_table"
     val stream = MemoryStream[Int]