This is an automated email from the ASF dual-hosted git repository.
wombatu-kun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 770e8aabf367 fix(spark): match the staged table, not LogicalWriteInfo,
in BasicStagedTable.newWriteBuilder (#19251)
770e8aabf367 is described below
commit 770e8aabf367b51b56912e4f9120fe0500334198
Author: Vova Kolmakov <[email protected]>
AuthorDate: Mon Jul 13 21:18:35 2026 +0700
fix(spark): match the staged table, not LogicalWriteInfo, in
BasicStagedTable.newWriteBuilder (#19251)
---
.../spark/sql/hudi/catalog/BasicStagedTable.scala | 2 +-
.../spark/sql/hudi/catalog/HoodieCatalog.scala | 20 +++-
.../sql/hudi/catalog/TestBasicStagedTable.scala | 52 ++++++++++
.../catalog/TestHoodieCatalogStagedTable.scala | 112 +++++++++++++++++++++
4 files changed, 182 insertions(+), 4 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/BasicStagedTable.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/BasicStagedTable.scala
index 2cc94c3e1479..ae2ac079187d 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/BasicStagedTable.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/BasicStagedTable.scala
@@ -38,7 +38,7 @@ case class BasicStagedTable(ident: Identifier,
table: Table,
catalog: TableCatalog) extends SupportsWrite with
StagedTable {
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
- info match {
+ table match {
case supportsWrite: SupportsWrite => supportsWrite.newWriteBuilder(info)
case _ => throw new HoodieException(s"Table `${ident.name}` does not
support writes.")
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
index 252760265302..f5ef66a04b92 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
@@ -74,7 +74,7 @@ class HoodieCatalog extends DelegatingCatalogExtension
} else {
BasicStagedTable(
ident,
- super.createTable(ident, schema, partitions, properties),
+ createOrLoadTable(ident, schema, partitions, properties),
this)
}
}
@@ -92,7 +92,7 @@ class HoodieCatalog extends DelegatingCatalogExtension
super.dropTable(ident)
BasicStagedTable(
ident,
- super.createTable(ident, schema, partitions, properties),
+ createOrLoadTable(ident, schema, partitions, properties),
this)
}
}
@@ -115,11 +115,25 @@ class HoodieCatalog extends DelegatingCatalogExtension
}
BasicStagedTable(
ident,
- super.createTable(ident, schema, partitions, properties),
+ createOrLoadTable(ident, schema, partitions, properties),
this)
}
}
+ /**
+ * Creates the table in the delegate catalog and returns it, never null.
+ *
+ * A delegate is allowed to return null from `createTable`:
`V2SessionCatalog` does so deliberately, to save the
+ * `loadTable` round-trip for a `CREATE TABLE` without `AS SELECT`. Load the
table that was just created in that
+ * case, so that the staged table is never backed by a null table. Spark
guards its own staging the same way.
+ */
+ private def createOrLoadTable(ident: Identifier,
+ schema: StructType,
+ partitions: Array[Transform],
+ properties: util.Map[String, String]): Table =
{
+ Option(super.createTable(ident, schema, partitions,
properties)).getOrElse(loadTable(ident))
+ }
+
override def loadTable(ident: Identifier): Table = {
super.loadTable(ident) match {
case V1Table(catalogTable0) if sparkAdapter.isHoodieTable(catalogTable0)
=>
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/catalog/TestBasicStagedTable.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/catalog/TestBasicStagedTable.scala
new file mode 100644
index 000000000000..7efc70a82abf
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/catalog/TestBasicStagedTable.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.hudi.catalog
+
+import org.apache.hudi.exception.HoodieException
+
+import org.apache.spark.sql.connector.catalog.{Identifier, SupportsWrite,
Table, TableCatalog}
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
+import org.junit.jupiter.api.Assertions.{assertSame, assertThrows, assertTrue}
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito.{mock, when}
+
+class TestBasicStagedTable {
+
+ private val ident = Identifier.of(Array("db"), "tbl")
+
+ @Test
+ def testNewWriteBuilderDelegatesToWritableTable(): Unit = {
+ val table = mock(classOf[SupportsWrite])
+ val info = mock(classOf[LogicalWriteInfo])
+ val writeBuilder = mock(classOf[WriteBuilder])
+ when(table.newWriteBuilder(info)).thenReturn(writeBuilder)
+
+ val staged = BasicStagedTable(ident, table, mock(classOf[TableCatalog]))
+
+ assertSame(writeBuilder, staged.newWriteBuilder(info))
+ }
+
+ @Test
+ def testNewWriteBuilderThrowsWhenTableIsNotWritable(): Unit = {
+ val staged = BasicStagedTable(ident, mock(classOf[Table]),
mock(classOf[TableCatalog]))
+
+ val ex = assertThrows(classOf[HoodieException],
+ () => staged.newWriteBuilder(mock(classOf[LogicalWriteInfo])))
+ assertTrue(ex.getMessage.contains("`tbl` does not support writes"))
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogStagedTable.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogStagedTable.scala
new file mode 100644
index 000000000000..c607b434580c
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogStagedTable.scala
@@ -0,0 +1,112 @@
+/*
+ * 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.hudi.catalog
+
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.testutils.HoodieClientTestUtils
+
+import org.apache.spark.api.java.JavaSparkContext
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.catalog.{Identifier, StagedTable,
SupportsWrite, Table, TableCatalog}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+import org.junit.jupiter.api.{AfterAll, BeforeAll, TestInstance}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertSame,
assertThrows, assertTrue}
+import org.junit.jupiter.api.TestInstance.Lifecycle
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+import org.mockito.Mockito.{mock, when}
+
+import java.util
+
+import scala.collection.JavaConverters._
+
+/**
+ * Tests the staging methods of {@link HoodieCatalog} for a non-Hudi table,
which stage through
+ * {@link BasicStagedTable} and hand the write off to the delegate catalog's
table.
+ */
+@TestInstance(Lifecycle.PER_CLASS)
+class TestHoodieCatalogStagedTable {
+
+ private val ident = Identifier.of(Array("default"), "tbl")
+ private val schema = StructType(Seq(StructField("id", IntegerType)))
+ private val partitions = Array.empty[Transform]
+ // Not a Hudi provider, so the staging methods take the BasicStagedTable
branch
+ private val properties: util.Map[String, String] = Map("provider" ->
"parquet").asJava
+
+ private var sparkSession: SparkSession = _
+
+ @BeforeAll
+ def setUp(): Unit = {
+ val jsc = new JavaSparkContext(
+
HoodieClientTestUtils.getSparkConfForTest(classOf[TestHoodieCatalogStagedTable].getName))
+ jsc.setLogLevel("ERROR")
+ // HoodieCatalog resolves SparkSession.active in its constructor
+ sparkSession = SparkSession.builder.config(jsc.getConf).getOrCreate
+ }
+
+ @AfterAll
+ def tearDown(): Unit = {
+ sparkSession.close()
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("stageCreate", "stageReplace",
"stageCreateOrReplace"))
+ def testStagedWriteIsDelegatedToWritableTable(stagingMethod: String): Unit =
{
+ val delegateTable = mock(classOf[SupportsWrite])
+ val info = mock(classOf[LogicalWriteInfo])
+ val writeBuilder = mock(classOf[WriteBuilder])
+ when(delegateTable.newWriteBuilder(info)).thenReturn(writeBuilder)
+ val delegate = mock(classOf[TableCatalog])
+ when(delegate.createTable(ident, schema, partitions,
properties)).thenReturn(delegateTable)
+
+ val staged = stage(stagingMethod, delegate)
+
+ assertSame(writeBuilder,
staged.asInstanceOf[SupportsWrite].newWriteBuilder(info))
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("stageCreate", "stageReplace",
"stageCreateOrReplace"))
+ def testStagedTableIsLoadedWhenDelegateCreateTableReturnsNull(stagingMethod:
String): Unit = {
+ // V2SessionCatalog, the default delegate, returns null from createTable
by design, to save the loadTable call
+ val delegate = mock(classOf[TableCatalog])
+ val loadedTable = mock(classOf[Table])
+ when(loadedTable.schema()).thenReturn(schema)
+ when(delegate.loadTable(ident)).thenReturn(loadedTable)
+
+ val staged = stage(stagingMethod, delegate)
+
+ // The staged table is backed by the table that was just created, rather
than by null
+ assertEquals(schema, staged.schema())
+ // It is not writable, so the write is rejected instead of being delegated
+ val ex = assertThrows(classOf[HoodieException],
+ () =>
staged.asInstanceOf[SupportsWrite].newWriteBuilder(mock(classOf[LogicalWriteInfo])))
+ assertTrue(ex.getMessage.contains("`tbl` does not support writes"))
+ }
+
+ private def stage(stagingMethod: String, delegate: TableCatalog):
StagedTable = {
+ val catalog = new HoodieCatalog()
+ catalog.setDelegateCatalog(delegate)
+ stagingMethod match {
+ case "stageCreate" => catalog.stageCreate(ident, schema, partitions,
properties)
+ case "stageReplace" => catalog.stageReplace(ident, schema, partitions,
properties)
+ case "stageCreateOrReplace" => catalog.stageCreateOrReplace(ident,
schema, partitions, properties)
+ }
+ }
+}
