voonhous commented on code in PR #19162: URL: https://github.com/apache/hudi/pull/19162#discussion_r3886053756
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = + names.filterNot(_.startsWith("_hoodie")).toSeq + + test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog API") { + withTempDir { tmp => + val catalog = hoodieCatalog + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + val ident = Identifier.of(Array("default"), tableName) + val schema = StructType(Seq( + StructField("id", IntegerType), + StructField("name", StringType), + StructField("ts", LongType))) + val props = Map( + "provider" -> "hudi", + "primaryKey" -> "id", + "preCombineField" -> "ts", + "location" -> tablePath).asJava + + // createTable routes to createHoodieTable(CREATE) and initializes the table on disk. + catalog.createTable(ident, schema, Array.empty[Transform], props) + assertTrue(catalog.tableExists(ident)) + assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists()) + + // loadTable returns a Hudi-backed table exposing the user schema. + val loaded = catalog.loadTable(ident) + assertEquals(Seq("id", "name", "ts"), userFieldNames(loaded.schema().fieldNames)) + + // alterTable: add a column. + catalog.alterTable(ident, TableChange.addColumn(Array("age"), IntegerType, true)) + assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age")) + + // alterTable: update a column comment. + catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"), "the name column")) + val commented = catalog.loadTable(ident).schema().fields.find(_.name == "name").get + assertEquals("the name column", commented.getComment().getOrElse("")) + + // alterTable: changing a column type is rejected by the V2 alter path, which routes to + // AlterHoodieTableChangeColumnCommand and does not support column type changes. + val typeChange = intercept[HoodieAnalysisException] { + catalog.alterTable(ident, TableChange.updateColumnType(Array("age"), LongType)) + } + assertTrue(typeChange.getMessage.contains( + "ALTER TABLE CHANGE COLUMN is not supported for changing column 'age'"), + typeChange.getMessage) + assertEquals(IntegerType, + catalog.loadTable(ident).schema().fields.find(_.name == "age").get.dataType) + + // renameTable moves the catalog entry. + val renamed = Identifier.of(Array("default"), s"${tableName}_renamed") + catalog.renameTable(ident, renamed) + assertFalse(catalog.tableExists(ident)) + assertTrue(catalog.tableExists(renamed)) + + // dropTable removes the Hudi table from the catalog. + assertTrue(catalog.dropTable(renamed)) + assertFalse(catalog.tableExists(renamed)) + } + } + + test("CTAS through the staged table commits managed and partitioned Hudi tables") { + val nonPartitioned = generateTableName + spark.sql( + s""" + |create table $nonPartitioned using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 10 as price, 1000 as ts + |""".stripMargin) + checkAnswer(s"select id, name, price, ts from $nonPartitioned")(Seq(1, "a1", 10, 1000)) + + val partitioned = generateTableName + spark.sql( + s""" + |create table $partitioned using hudi + |partitioned by (dt) + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 1000 as ts, '2024-01-01' as dt + | union all select 2, 'a2', 2000, '2024-01-02' + |""".stripMargin) + checkAnswer(s"select id, name, dt from $partitioned")( + Seq(1, "a1", "2024-01-01"), Seq(2, "a2", "2024-01-02")) + } + + test("A failing CTAS aborts staged changes and cleans up the table path") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + checkExceptionContain( + s""" + |create table $tableName using hudi + |tblproperties (primaryKey = 'id', type = 'cow', hoodie.compact.inline = 'true') + |location '$tablePath' + |as select 1 as id, 'a1' as name, 1000 as ts + |""".stripMargin)("Compaction is not supported on a CopyOnWrite table") + assertFalse(existsPath(tablePath)) + } + } + + test("SHOW CREATE TABLE regenerates Hudi DDL") { + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | ts long, + | dt string + |) using hudi + |partitioned by (dt) + |comment 'a hudi table' + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |""".stripMargin) + + // SHOW CREATE TABLE resolves through Spark's native command for the Hudi V2 table, which + // emits `CREATE TABLE <table>` with a USING/TBLPROPERTIES body. The catalog qualifier differs + // by Spark version (`spark_catalog.default.` on 3.5+, `default.` on 3.4/3.3), so match either. + val ddl = spark.sql(s"show create table $tableName").head().getString(0) Review Comment: This `show create table` resolves to Spark's `ShowCreateTableExec`, not `ShowHoodieCreateTableCommand`: Spark 3.5 `ResolveSessionCatalog` emits the V1 command only for views, `asSerde`, Hive-serde tables or `spark.sql.legacy.useV1Command`, and the Hudi rewrite (`HoodieAnalysis.scala:610-612`) only ever sees that V1 command. The Hudi command emits ``CREATE TABLE IF NOT EXISTS `default`.`t` `` (`quotedString`), which would fail the `default.<t>` check on this line. Could we wrap this in `withSQLConf("spark.sql.legacy.useV1Command" -> "true")` and assert `CREATE TABLE IF NOT EXISTS` plus the backticked name, so the DDL builder is actually covered? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieSqlCommonUtils.scala: ########## @@ -0,0 +1,165 @@ +/* + * 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.common + +import org.apache.hudi.common.table.read.IncrementalQueryAnalyzer + +import org.apache.hadoop.conf.Configuration +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution +import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType} +import org.apache.spark.sql.catalyst.expressions.AttributeReference +import org.apache.spark.sql.hudi.HoodieSqlCommonUtils +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.funsuite.AnyFunSuite + +import java.net.URI +import java.util.TimeZone + +/** + * Unit coverage for the pure helper methods in [[HoodieSqlCommonUtils]] that are otherwise + * only reached through heavier read/write code paths. + */ +class TestHoodieSqlCommonUtils extends AnyFunSuite with BeforeAndAfterAll { + + // Instant formatting is timezone sensitive; pin the JVM default to UTC for the suite and + // restore it afterwards so the change does not leak into other suites in the same JVM. + private var originalTimeZone: TimeZone = _ + + override protected def beforeAll(): Unit = { + super.beforeAll() + originalTimeZone = TimeZone.getDefault + TimeZone.setDefault(TimeZone.getTimeZone("UTC")) + } + + override protected def afterAll(): Unit = { + TimeZone.setDefault(originalTimeZone) + super.afterAll() + } + + private def partitionedTable(partitionCols: Seq[String]): CatalogTable = { + val fields = StructField("id", IntegerType) +: partitionCols.map(StructField(_, StringType)) + CatalogTable( + identifier = TableIdentifier("t", Some("default")), + tableType = CatalogTableType.MANAGED, + storage = CatalogStorageFormat.empty, + schema = StructType(fields), + provider = Some("hudi"), + partitionColumnNames = partitionCols) + } + + private val nonPartitionedTable: CatalogTable = CatalogTable( + identifier = TableIdentifier("t0", Some("default")), + tableType = CatalogTableType.MANAGED, + storage = CatalogStorageFormat.empty, + schema = StructType(Seq(StructField("id", IntegerType))), + provider = Some("hudi")) + + test("formatQueryInstant normalizes supported time formats") { Review Comment: Both instant tests are covered input-for-input by `TestInstantTimeValidation` in this package (#18426), which also checks the 14/17-digit and `.SSS` forms and uses the timezone-independent `matches("\\d{17}")`. Dropping them would also remove the `TimeZone` pin above, which only the epoch assertions need and which is order-fragile: `HoodieSqlCommonUtils.scala:58-62` caches a thread-local `SimpleDateFormat` with whatever TZ was current at first use. Could we drop these two tests and the `beforeAll`/`afterAll` pin? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = + names.filterNot(_.startsWith("_hoodie")).toSeq + + test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog API") { + withTempDir { tmp => + val catalog = hoodieCatalog + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + val ident = Identifier.of(Array("default"), tableName) + val schema = StructType(Seq( + StructField("id", IntegerType), + StructField("name", StringType), + StructField("ts", LongType))) + val props = Map( + "provider" -> "hudi", + "primaryKey" -> "id", + "preCombineField" -> "ts", + "location" -> tablePath).asJava + + // createTable routes to createHoodieTable(CREATE) and initializes the table on disk. + catalog.createTable(ident, schema, Array.empty[Transform], props) + assertTrue(catalog.tableExists(ident)) + assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists()) + + // loadTable returns a Hudi-backed table exposing the user schema. + val loaded = catalog.loadTable(ident) + assertEquals(Seq("id", "name", "ts"), userFieldNames(loaded.schema().fieldNames)) + + // alterTable: add a column. + catalog.alterTable(ident, TableChange.addColumn(Array("age"), IntegerType, true)) + assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age")) + + // alterTable: update a column comment. + catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"), "the name column")) + val commented = catalog.loadTable(ident).schema().fields.find(_.name == "name").get + assertEquals("the name column", commented.getComment().getOrElse("")) + + // alterTable: changing a column type is rejected by the V2 alter path, which routes to + // AlterHoodieTableChangeColumnCommand and does not support column type changes. + val typeChange = intercept[HoodieAnalysisException] { + catalog.alterTable(ident, TableChange.updateColumnType(Array("age"), LongType)) + } + assertTrue(typeChange.getMessage.contains( + "ALTER TABLE CHANGE COLUMN is not supported for changing column 'age'"), + typeChange.getMessage) + assertEquals(IntegerType, + catalog.loadTable(ident).schema().fields.find(_.name == "age").get.dataType) + + // renameTable moves the catalog entry. + val renamed = Identifier.of(Array("default"), s"${tableName}_renamed") + catalog.renameTable(ident, renamed) + assertFalse(catalog.tableExists(ident)) + assertTrue(catalog.tableExists(renamed)) + + // dropTable removes the Hudi table from the catalog. + assertTrue(catalog.dropTable(renamed)) + assertFalse(catalog.tableExists(renamed)) + } + } + + test("CTAS through the staged table commits managed and partitioned Hudi tables") { + val nonPartitioned = generateTableName + spark.sql( + s""" + |create table $nonPartitioned using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 10 as price, 1000 as ts + |""".stripMargin) + checkAnswer(s"select id, name, price, ts from $nonPartitioned")(Seq(1, "a1", 10, 1000)) + + val partitioned = generateTableName + spark.sql( + s""" + |create table $partitioned using hudi + |partitioned by (dt) + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 1000 as ts, '2024-01-01' as dt + | union all select 2, 'a2', 2000, '2024-01-02' + |""".stripMargin) + checkAnswer(s"select id, name, dt from $partitioned")( + Seq(1, "a1", "2024-01-01"), Seq(2, "a2", "2024-01-02")) + } + + test("A failing CTAS aborts staged changes and cleans up the table path") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + checkExceptionContain( + s""" + |create table $tableName using hudi + |tblproperties (primaryKey = 'id', type = 'cow', hoodie.compact.inline = 'true') + |location '$tablePath' + |as select 1 as id, 'a1' as name, 1000 as ts + |""".stripMargin)("Compaction is not supported on a CopyOnWrite table") + assertFalse(existsPath(tablePath)) + } + } + + test("SHOW CREATE TABLE regenerates Hudi DDL") { + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | ts long, + | dt string + |) using hudi + |partitioned by (dt) + |comment 'a hudi table' + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |""".stripMargin) + + // SHOW CREATE TABLE resolves through Spark's native command for the Hudi V2 table, which + // emits `CREATE TABLE <table>` with a USING/TBLPROPERTIES body. The catalog qualifier differs + // by Spark version (`spark_catalog.default.` on 3.5+, `default.` on 3.4/3.3), so match either. + val ddl = spark.sql(s"show create table $tableName").head().getString(0) + assertTrue(ddl.contains("CREATE TABLE") && ddl.contains(s"default.$tableName"), ddl) + assertTrue(ddl.contains("USING hudi"), ddl) + assertTrue(ddl.contains("PARTITIONED BY (dt)"), ddl) + assertTrue(ddl.contains("COMMENT 'a hudi table'"), ddl) + assertTrue(ddl.contains("TBLPROPERTIES"), ddl) + assertTrue(ddl.contains("primaryKey"), ddl) + + intercept[NoSuchTableException] { + ShowHoodieCreateTableCommand(TableIdentifier("does_not_exist_tbl")).run(spark) + } + } + + test("CREATE over an existing location validates conflicting table properties") { + withTempDir { tmp => + val basePath = s"${tmp.getCanonicalPath}/shared" + val first = generateTableName + spark.sql( + s""" + |create table $first (id int, name string, ts long) using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |location '$basePath' + |""".stripMargin) + + // A second table over the same location that keeps the on-disk table config resolves + // to the persisted schema (existing-location reuse path). + val reuse = generateTableName + spark.sql( + s""" + |create table $reuse (id int, name string, ts long) using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |location '$basePath' + |""".stripMargin) + assertEquals(Seq("id", "name", "ts"), userFieldNames(spark.table(reuse).schema.fieldNames)) + + // A conflicting primaryKey against the on-disk table config is rejected with a config-conflict + // error surfaced from the write-path validation (HoodieWriterUtils). + val conflicting = generateTableName + checkExceptionContain( + s""" + |create table $conflicting (id int, name string, ts long) using hudi + |tblproperties (primaryKey = 'name', preCombineField = 'ts') + |location '$basePath' + |""".stripMargin)("hoodie.table.recordkey.fields") + } + } + + test("HoodieInternalV2Table exposes v2 capabilities and handles reads and writes") { + withTempDir { tmp => + withSQLConf("hoodie.schema.on.read.enable" -> "true") { + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + spark.sql( + s""" + |create table $tableName (id int, name string, ts long) using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |location '$tablePath' + |""".stripMargin) + + // With schema evolution enabled, loadTable returns the V2 table directly. + val ident = Identifier.of(Array("default"), tableName) + val loaded = hoodieCatalog.loadTable(ident) + assertTrue(loaded.isInstanceOf[HoodieInternalV2Table]) + val v2 = loaded.asInstanceOf[HoodieInternalV2Table] + assertTrue(v2.capabilities().contains(TableCapability.BATCH_READ)) Review Comment: `capabilities()` declares five entries (`HoodieInternalV2Table.scala:66-68`: `BATCH_READ, V1_BATCH_WRITE, OVERWRITE_BY_FILTER, TRUNCATE, ACCEPT_ANY_SCHEMA`), so checking two of them would not catch a dropped `TRUNCATE` or `OVERWRITE_BY_FILTER`. Also, the comment below about the "V1-fallback write builder" does not hold: `HoodieSpark35Analysis.scala:57-58` swaps the V2 relation for a V1 `LogicalRelation` before any write builder exists. Could we `assertEquals` the full set and reword the comment? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = + names.filterNot(_.startsWith("_hoodie")).toSeq + + test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog API") { + withTempDir { tmp => + val catalog = hoodieCatalog + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + val ident = Identifier.of(Array("default"), tableName) + val schema = StructType(Seq( + StructField("id", IntegerType), + StructField("name", StringType), + StructField("ts", LongType))) + val props = Map( + "provider" -> "hudi", + "primaryKey" -> "id", + "preCombineField" -> "ts", + "location" -> tablePath).asJava + + // createTable routes to createHoodieTable(CREATE) and initializes the table on disk. + catalog.createTable(ident, schema, Array.empty[Transform], props) + assertTrue(catalog.tableExists(ident)) + assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists()) + + // loadTable returns a Hudi-backed table exposing the user schema. + val loaded = catalog.loadTable(ident) + assertEquals(Seq("id", "name", "ts"), userFieldNames(loaded.schema().fieldNames)) + + // alterTable: add a column. + catalog.alterTable(ident, TableChange.addColumn(Array("age"), IntegerType, true)) + assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age")) + + // alterTable: update a column comment. + catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"), "the name column")) + val commented = catalog.loadTable(ident).schema().fields.find(_.name == "name").get + assertEquals("the name column", commented.getComment().getOrElse("")) + + // alterTable: changing a column type is rejected by the V2 alter path, which routes to + // AlterHoodieTableChangeColumnCommand and does not support column type changes. + val typeChange = intercept[HoodieAnalysisException] { + catalog.alterTable(ident, TableChange.updateColumnType(Array("age"), LongType)) + } + assertTrue(typeChange.getMessage.contains( + "ALTER TABLE CHANGE COLUMN is not supported for changing column 'age'"), + typeChange.getMessage) + assertEquals(IntegerType, + catalog.loadTable(ident).schema().fields.find(_.name == "age").get.dataType) + + // renameTable moves the catalog entry. + val renamed = Identifier.of(Array("default"), s"${tableName}_renamed") + catalog.renameTable(ident, renamed) + assertFalse(catalog.tableExists(ident)) + assertTrue(catalog.tableExists(renamed)) + + // dropTable removes the Hudi table from the catalog. + assertTrue(catalog.dropTable(renamed)) Review Comment: For a table created with an explicit `location`, the behavior that matters on drop is that the data survives: `HoodieCatalog.scala:195` passes `purge = false`. Could we add `assertTrue(existsPath(tablePath))` after the drop? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = + names.filterNot(_.startsWith("_hoodie")).toSeq + + test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog API") { + withTempDir { tmp => + val catalog = hoodieCatalog + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + val ident = Identifier.of(Array("default"), tableName) + val schema = StructType(Seq( + StructField("id", IntegerType), + StructField("name", StringType), + StructField("ts", LongType))) + val props = Map( + "provider" -> "hudi", + "primaryKey" -> "id", + "preCombineField" -> "ts", + "location" -> tablePath).asJava + + // createTable routes to createHoodieTable(CREATE) and initializes the table on disk. + catalog.createTable(ident, schema, Array.empty[Transform], props) + assertTrue(catalog.tableExists(ident)) + assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists()) + + // loadTable returns a Hudi-backed table exposing the user schema. + val loaded = catalog.loadTable(ident) + assertEquals(Seq("id", "name", "ts"), userFieldNames(loaded.schema().fieldNames)) + + // alterTable: add a column. + catalog.alterTable(ident, TableChange.addColumn(Array("age"), IntegerType, true)) + assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age")) + + // alterTable: update a column comment. + catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"), "the name column")) + val commented = catalog.loadTable(ident).schema().fields.find(_.name == "name").get + assertEquals("the name column", commented.getComment().getOrElse("")) + + // alterTable: changing a column type is rejected by the V2 alter path, which routes to + // AlterHoodieTableChangeColumnCommand and does not support column type changes. + val typeChange = intercept[HoodieAnalysisException] { + catalog.alterTable(ident, TableChange.updateColumnType(Array("age"), LongType)) + } + assertTrue(typeChange.getMessage.contains( + "ALTER TABLE CHANGE COLUMN is not supported for changing column 'age'"), + typeChange.getMessage) + assertEquals(IntegerType, + catalog.loadTable(ident).schema().fields.find(_.name == "age").get.dataType) + + // renameTable moves the catalog entry. + val renamed = Identifier.of(Array("default"), s"${tableName}_renamed") + catalog.renameTable(ident, renamed) + assertFalse(catalog.tableExists(ident)) + assertTrue(catalog.tableExists(renamed)) + + // dropTable removes the Hudi table from the catalog. + assertTrue(catalog.dropTable(renamed)) + assertFalse(catalog.tableExists(renamed)) + } + } + + test("CTAS through the staged table commits managed and partitioned Hudi tables") { + val nonPartitioned = generateTableName + spark.sql( + s""" + |create table $nonPartitioned using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 10 as price, 1000 as ts + |""".stripMargin) + checkAnswer(s"select id, name, price, ts from $nonPartitioned")(Seq(1, "a1", 10, 1000)) + + val partitioned = generateTableName + spark.sql( + s""" + |create table $partitioned using hudi + |partitioned by (dt) + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 1000 as ts, '2024-01-01' as dt + | union all select 2, 'a2', 2000, '2024-01-02' + |""".stripMargin) + checkAnswer(s"select id, name, dt from $partitioned")( + Seq(1, "a1", "2024-01-01"), Seq(2, "a2", "2024-01-02")) + } + + test("A failing CTAS aborts staged changes and cleans up the table path") { Review Comment: This is the same scenario as `TestCreateTable.scala:1430` ("Test CTAS using an illegal definition -- a COW table with compaction enabled"): same SQL, same message, same `existsPath` assertion; the only delta is the explicit `location`. The happy-path CTAS above also overlaps `TestCreateTable.scala:373` and `:889`, which additionally assert the `BULK_INSERT` operation type. Could we fold the `location` variant into `TestCreateTable:1430` and drop these two? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hive/TestSparkCatalogMetaStoreClient.scala: ########## @@ -203,6 +203,202 @@ class TestSparkCatalogMetaStoreClient extends FunSuite with BeforeAndAfterAll { } } + test("supported client edge cases: empty partitions, no-op setMetaConf, default database location") { + val client = newClient() + + // add_partitions with an empty list is a no-op that returns an empty result. + assertTrue(client.add_partitions(new util.ArrayList[Partition](), false, true).isEmpty) + + // setMetaConf is intentionally a silent no-op, while getMetaConf stays unsupported. + client.setMetaConf("hive.metastore.callerContext", "hudi") + assertThrows[UnsupportedOperationException](client.getMetaConf("hive.metastore.callerContext")) + + // createDatabase without an explicit location falls back to the warehouse path. + val databaseName = generateName("db") + client.createDatabase(new Database(databaseName, "no-location db", null, new util.HashMap[String, String]())) + assertNotNull(client.getDatabase(databaseName).getLocationUri) + assertFalse(client.tableExists(databaseName, "missing_table")) + } + + test("unsupported IMetaStoreClient operations throw UnsupportedOperationException") { + // SparkCatalogMetaStoreClient only implements the subset of IMetaStoreClient exercised by + // HoodieHiveSyncClient/HMSDDLExecutor. Every other method must fail fast rather than return + // a misleading default. This locks in that contract across the delegated surface. + val client = newClient() + + // Connection / config lifecycle. + assertUnsupported(client.isCompatibleWith(null: org.apache.hadoop.hive.conf.HiveConf)) Review Comment: nit, feel free to ignore: 43 `null: org.apache.hadoop.hive.metastore.api.X` casts (29 types) in a file that already imports from that package; could these become imports? The list matches the 140 throwing overrides exactly today, but nothing fails when a new one is added, so a reflective derivation from `IMetaStoreClient` minus the implemented methods would be self-maintaining. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hive/TestSparkCatalogMetaStoreClient.scala: ########## @@ -203,6 +203,202 @@ class TestSparkCatalogMetaStoreClient extends FunSuite with BeforeAndAfterAll { } } + test("supported client edge cases: empty partitions, no-op setMetaConf, default database location") { + val client = newClient() + + // add_partitions with an empty list is a no-op that returns an empty result. + assertTrue(client.add_partitions(new util.ArrayList[Partition](), false, true).isEmpty) + + // setMetaConf is intentionally a silent no-op, while getMetaConf stays unsupported. + client.setMetaConf("hive.metastore.callerContext", "hudi") + assertThrows[UnsupportedOperationException](client.getMetaConf("hive.metastore.callerContext")) + + // createDatabase without an explicit location falls back to the warehouse path. + val databaseName = generateName("db") + client.createDatabase(new Database(databaseName, "no-location db", null, new util.HashMap[String, String]())) + assertNotNull(client.getDatabase(databaseName).getLocationUri) + assertFalse(client.tableExists(databaseName, "missing_table")) + } + + test("unsupported IMetaStoreClient operations throw UnsupportedOperationException") { + // SparkCatalogMetaStoreClient only implements the subset of IMetaStoreClient exercised by + // HoodieHiveSyncClient/HMSDDLExecutor. Every other method must fail fast rather than return + // a misleading default. This locks in that contract across the delegated surface. + val client = newClient() + + // Connection / config lifecycle. + assertUnsupported(client.isCompatibleWith(null: org.apache.hadoop.hive.conf.HiveConf)) + assertUnsupported(client.isSameConfObj(null: org.apache.hadoop.hive.conf.HiveConf)) + assertUnsupported(client.setHiveAddedJars(null: String)) + assertUnsupported(client.isLocalMetaStore()) + assertUnsupported(client.reconnect()) + assertUnsupported(client.close()) Review Comment: This pins `UnsupportedOperationException` for `close()` and (line 268) `getPartition(db, tbl, values)`, but both are on the live path with `hoodie.datasource.hive_sync.use_spark_catalog=true`: `HMSDDLExecutor.dropPartitionsToTable:261` -> `HivePartitionUtil.partitionExists:88` calls `getPartition` and catches only `NoSuchObjectException`/`TException` (`:89`, `:91`), so every partition drop fails; `HoodieHiveSyncClient:707` calls `close()` on every sync and logs a WARN with a stack trace. Could we make `close()` a no-op (same precedent as `setMetaConf`), implement `getPartition` via `externalCatalog.getPartitionOption` (throwing `NoSuchObjectException` when absent), and flip these two assertions? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = + names.filterNot(_.startsWith("_hoodie")).toSeq + + test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog API") { + withTempDir { tmp => + val catalog = hoodieCatalog + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + val ident = Identifier.of(Array("default"), tableName) + val schema = StructType(Seq( + StructField("id", IntegerType), + StructField("name", StringType), + StructField("ts", LongType))) + val props = Map( + "provider" -> "hudi", + "primaryKey" -> "id", + "preCombineField" -> "ts", + "location" -> tablePath).asJava + + // createTable routes to createHoodieTable(CREATE) and initializes the table on disk. + catalog.createTable(ident, schema, Array.empty[Transform], props) + assertTrue(catalog.tableExists(ident)) + assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists()) + + // loadTable returns a Hudi-backed table exposing the user schema. + val loaded = catalog.loadTable(ident) + assertEquals(Seq("id", "name", "ts"), userFieldNames(loaded.schema().fieldNames)) + + // alterTable: add a column. + catalog.alterTable(ident, TableChange.addColumn(Array("age"), IntegerType, true)) + assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age")) + + // alterTable: update a column comment. + catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"), "the name column")) + val commented = catalog.loadTable(ident).schema().fields.find(_.name == "name").get + assertEquals("the name column", commented.getComment().getOrElse("")) + + // alterTable: changing a column type is rejected by the V2 alter path, which routes to + // AlterHoodieTableChangeColumnCommand and does not support column type changes. + val typeChange = intercept[HoodieAnalysisException] { + catalog.alterTable(ident, TableChange.updateColumnType(Array("age"), LongType)) + } + assertTrue(typeChange.getMessage.contains( + "ALTER TABLE CHANGE COLUMN is not supported for changing column 'age'"), + typeChange.getMessage) + assertEquals(IntegerType, + catalog.loadTable(ident).schema().fields.find(_.name == "age").get.dataType) + + // renameTable moves the catalog entry. + val renamed = Identifier.of(Array("default"), s"${tableName}_renamed") + catalog.renameTable(ident, renamed) + assertFalse(catalog.tableExists(ident)) + assertTrue(catalog.tableExists(renamed)) + + // dropTable removes the Hudi table from the catalog. + assertTrue(catalog.dropTable(renamed)) + assertFalse(catalog.tableExists(renamed)) + } + } + + test("CTAS through the staged table commits managed and partitioned Hudi tables") { Review Comment: Neither CTAS test reaches `HoodieStagedTable`: `BaseDefaultSource` is not a `TableProvider`, so Spark's `isV2Provider("hudi")` is false and `ResolveSessionCatalog` builds the V1 `CreateTable`, which `HoodieAnalysis.scala:410-415` turns into `CreateHoodieTableAsSelectCommand`; the cleanup asserted in the next test comes from its `clearTablePath` (`:124`). The staged arms are reachable only through a non-session catalog registration (Spark rejects RTAS on a V1 provider). Could we drop the `HoodieStagedTable` claim from the scaladoc and PR body, or add a staged test through a second catalog name? <details><summary>Why a staged test would be worth having</summary> `HoodieStagedTable.abortStagedChanges` (`:73-81`) deletes the location unconditionally, and `commitStagedChanges` can fail after the path exists: `assert isEmptyPath` at `HoodieCatalog.scala:320-323`, and the CREATE_OR_REPLACE arm runs `saveSourceDF` and then `CreateHoodieTableCommand(tableDesc, false)`, which throws "already exists" (`:64`). Through that route a CTAS into a non-empty location, or a REPLACE of an existing table, deletes pre-existing data. </details> ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = + names.filterNot(_.startsWith("_hoodie")).toSeq + + test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog API") { + withTempDir { tmp => + val catalog = hoodieCatalog + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + val ident = Identifier.of(Array("default"), tableName) + val schema = StructType(Seq( + StructField("id", IntegerType), + StructField("name", StringType), + StructField("ts", LongType))) + val props = Map( + "provider" -> "hudi", + "primaryKey" -> "id", + "preCombineField" -> "ts", + "location" -> tablePath).asJava + + // createTable routes to createHoodieTable(CREATE) and initializes the table on disk. + catalog.createTable(ident, schema, Array.empty[Transform], props) + assertTrue(catalog.tableExists(ident)) + assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists()) + + // loadTable returns a Hudi-backed table exposing the user schema. + val loaded = catalog.loadTable(ident) + assertEquals(Seq("id", "name", "ts"), userFieldNames(loaded.schema().fieldNames)) + + // alterTable: add a column. + catalog.alterTable(ident, TableChange.addColumn(Array("age"), IntegerType, true)) + assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age")) + + // alterTable: update a column comment. + catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"), "the name column")) + val commented = catalog.loadTable(ident).schema().fields.find(_.name == "name").get + assertEquals("the name column", commented.getComment().getOrElse("")) + + // alterTable: changing a column type is rejected by the V2 alter path, which routes to + // AlterHoodieTableChangeColumnCommand and does not support column type changes. + val typeChange = intercept[HoodieAnalysisException] { + catalog.alterTable(ident, TableChange.updateColumnType(Array("age"), LongType)) + } + assertTrue(typeChange.getMessage.contains( + "ALTER TABLE CHANGE COLUMN is not supported for changing column 'age'"), + typeChange.getMessage) + assertEquals(IntegerType, + catalog.loadTable(ident).schema().fields.find(_.name == "age").get.dataType) + + // renameTable moves the catalog entry. + val renamed = Identifier.of(Array("default"), s"${tableName}_renamed") + catalog.renameTable(ident, renamed) + assertFalse(catalog.tableExists(ident)) + assertTrue(catalog.tableExists(renamed)) + + // dropTable removes the Hudi table from the catalog. + assertTrue(catalog.dropTable(renamed)) + assertFalse(catalog.tableExists(renamed)) + } + } + + test("CTAS through the staged table commits managed and partitioned Hudi tables") { + val nonPartitioned = generateTableName + spark.sql( + s""" + |create table $nonPartitioned using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 10 as price, 1000 as ts + |""".stripMargin) + checkAnswer(s"select id, name, price, ts from $nonPartitioned")(Seq(1, "a1", 10, 1000)) + + val partitioned = generateTableName + spark.sql( + s""" + |create table $partitioned using hudi + |partitioned by (dt) + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |as select 1 as id, 'a1' as name, 1000 as ts, '2024-01-01' as dt + | union all select 2, 'a2', 2000, '2024-01-02' + |""".stripMargin) + checkAnswer(s"select id, name, dt from $partitioned")( + Seq(1, "a1", "2024-01-01"), Seq(2, "a2", "2024-01-02")) + } + + test("A failing CTAS aborts staged changes and cleans up the table path") { + withTempDir { tmp => + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + checkExceptionContain( + s""" + |create table $tableName using hudi + |tblproperties (primaryKey = 'id', type = 'cow', hoodie.compact.inline = 'true') + |location '$tablePath' + |as select 1 as id, 'a1' as name, 1000 as ts + |""".stripMargin)("Compaction is not supported on a CopyOnWrite table") + assertFalse(existsPath(tablePath)) + } + } + + test("SHOW CREATE TABLE regenerates Hudi DDL") { + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | ts long, + | dt string + |) using hudi + |partitioned by (dt) + |comment 'a hudi table' + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |""".stripMargin) + + // SHOW CREATE TABLE resolves through Spark's native command for the Hudi V2 table, which + // emits `CREATE TABLE <table>` with a USING/TBLPROPERTIES body. The catalog qualifier differs + // by Spark version (`spark_catalog.default.` on 3.5+, `default.` on 3.4/3.3), so match either. + val ddl = spark.sql(s"show create table $tableName").head().getString(0) + assertTrue(ddl.contains("CREATE TABLE") && ddl.contains(s"default.$tableName"), ddl) + assertTrue(ddl.contains("USING hudi"), ddl) + assertTrue(ddl.contains("PARTITIONED BY (dt)"), ddl) + assertTrue(ddl.contains("COMMENT 'a hudi table'"), ddl) + assertTrue(ddl.contains("TBLPROPERTIES"), ddl) + assertTrue(ddl.contains("primaryKey"), ddl) + + intercept[NoSuchTableException] { + ShowHoodieCreateTableCommand(TableIdentifier("does_not_exist_tbl")).run(spark) + } + } + + test("CREATE over an existing location validates conflicting table properties") { + withTempDir { tmp => + val basePath = s"${tmp.getCanonicalPath}/shared" + val first = generateTableName + spark.sql( + s""" + |create table $first (id int, name string, ts long) using hudi + |tblproperties (primaryKey = 'id', preCombineField = 'ts') + |location '$basePath' + |""".stripMargin) + + // A second table over the same location that keeps the on-disk table config resolves + // to the persisted schema (existing-location reuse path). + val reuse = generateTableName Review Comment: This arm declares the same three columns that are on disk, so the assertion passes whether the persisted or the declared schema wins; `TestCreateTable.scala:1021` and `:1637` already cover the reuse path (the first by omitting the column list). Could we drop this arm and keep only the conflicting-`primaryKey` case, which is new? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hive/TestSparkCatalogMetaStoreClient.scala: ########## @@ -203,6 +203,202 @@ class TestSparkCatalogMetaStoreClient extends FunSuite with BeforeAndAfterAll { } } + test("supported client edge cases: empty partitions, no-op setMetaConf, default database location") { + val client = newClient() + + // add_partitions with an empty list is a no-op that returns an empty result. + assertTrue(client.add_partitions(new util.ArrayList[Partition](), false, true).isEmpty) + + // setMetaConf is intentionally a silent no-op, while getMetaConf stays unsupported. + client.setMetaConf("hive.metastore.callerContext", "hudi") + assertThrows[UnsupportedOperationException](client.getMetaConf("hive.metastore.callerContext")) + + // createDatabase without an explicit location falls back to the warehouse path. + val databaseName = generateName("db") + client.createDatabase(new Database(databaseName, "no-location db", null, new util.HashMap[String, String]())) + assertNotNull(client.getDatabase(databaseName).getLocationUri) Review Comment: `assertNotNull(getLocationUri)` cannot fail (Spark always assigns a database location), so it does not pin the warehouse fallback named in the test title; could we assert the URI contains `warehouseDir.getCanonicalPath`? Also, `getMetaConf` at line 214 is asserted again in the contract test at line 236. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieSqlCommonUtils.scala: ########## @@ -0,0 +1,165 @@ +/* + * 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.common + +import org.apache.hudi.common.table.read.IncrementalQueryAnalyzer + +import org.apache.hadoop.conf.Configuration +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution +import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType} +import org.apache.spark.sql.catalyst.expressions.AttributeReference +import org.apache.spark.sql.hudi.HoodieSqlCommonUtils +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.funsuite.AnyFunSuite + +import java.net.URI +import java.util.TimeZone + +/** + * Unit coverage for the pure helper methods in [[HoodieSqlCommonUtils]] that are otherwise + * only reached through heavier read/write code paths. + */ +class TestHoodieSqlCommonUtils extends AnyFunSuite with BeforeAndAfterAll { + + // Instant formatting is timezone sensitive; pin the JVM default to UTC for the suite and + // restore it afterwards so the change does not leak into other suites in the same JVM. + private var originalTimeZone: TimeZone = _ + + override protected def beforeAll(): Unit = { + super.beforeAll() + originalTimeZone = TimeZone.getDefault + TimeZone.setDefault(TimeZone.getTimeZone("UTC")) + } + + override protected def afterAll(): Unit = { + TimeZone.setDefault(originalTimeZone) + super.afterAll() + } + + private def partitionedTable(partitionCols: Seq[String]): CatalogTable = { + val fields = StructField("id", IntegerType) +: partitionCols.map(StructField(_, StringType)) + CatalogTable( + identifier = TableIdentifier("t", Some("default")), + tableType = CatalogTableType.MANAGED, + storage = CatalogStorageFormat.empty, + schema = StructType(fields), + provider = Some("hudi"), + partitionColumnNames = partitionCols) + } + + private val nonPartitionedTable: CatalogTable = CatalogTable( + identifier = TableIdentifier("t0", Some("default")), + tableType = CatalogTableType.MANAGED, + storage = CatalogStorageFormat.empty, + schema = StructType(Seq(StructField("id", IntegerType))), + provider = Some("hudi")) + + test("formatQueryInstant normalizes supported time formats") { + assertTrue(HoodieSqlCommonUtils.formatQueryInstant("2021-04-01").startsWith("20210401")) + assertTrue(HoodieSqlCommonUtils.formatQueryInstant("2021-04-01 12:30:45").startsWith("20210401123045")) + assertTrue(HoodieSqlCommonUtils.formatQueryInstant("2021-04-01T12:30:45").startsWith("20210401123045")) + // 10-digit epoch seconds (2021-01-01T00:00:00Z) and 13-digit epoch millis. + assertTrue(HoodieSqlCommonUtils.formatQueryInstant("1609459200").startsWith("20210101")) + assertTrue(HoodieSqlCommonUtils.formatQueryInstant("1609459200000").startsWith("20210101")) + intercept[IllegalArgumentException] { + HoodieSqlCommonUtils.formatQueryInstant("abc") + } + } + + test("formatIncrementalInstant passes sentinels through and normalizes real instants") { + assertEquals(IncrementalQueryAnalyzer.START_COMMIT_EARLIEST, + HoodieSqlCommonUtils.formatIncrementalInstant(IncrementalQueryAnalyzer.START_COMMIT_EARLIEST)) + assertEquals("000", HoodieSqlCommonUtils.formatIncrementalInstant("000")) + // A short, zero-prefixed numeric value is treated as a legacy instant and passed through. + assertEquals("0000001", HoodieSqlCommonUtils.formatIncrementalInstant("0000001")) + assertTrue(HoodieSqlCommonUtils.formatIncrementalInstant("2021-04-01").startsWith("20210401")) + } + + test("partition style detectors classify partition paths") { + val t = partitionedTable(Seq("dt")) + + assertTrue(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("dt=2021-04-01"), t)) + assertFalse(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("2021-04-01"), t)) + assertTrue(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("anything"), nonPartitionedTable)) + + assertTrue(HoodieSqlCommonUtils.isUrlEncodeEnabled(Seq("dt=2021-04-01"), t)) Review Comment: `isUrlEncodeEnabled` is `split("/").length == partitionColumnNames.size` (`HoodieSqlCommonUtils.scala:139-145`), so with one partition column these cases only check "no slash". Could we add a two-column case (`Seq("2021%2F04%2F01/12")` true vs `Seq("2021/04/01/12")` false)? Separately, `isSlashSeparatedDatePartitioning` has no production caller (only its definition, from #17787); should the three assertions below be dropped rather than pinning dead code? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = Review Comment: nit, feel free to ignore: `HoodieSqlCommonUtils.removeMetaFields(schema).fieldNames` already does this (`TestAlterTable.scala:99`); could we use it instead of a local helper? ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala: ########## @@ -0,0 +1,243 @@ +/* + * 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.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException +import org.apache.spark.sql.connector.catalog.{Identifier, TableCapability, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand +import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} + +import java.io.File + +import scala.collection.JavaConverters._ + +/** + * DDL-level coverage for [[HoodieCatalog]], [[HoodieStagedTable]], [[HoodieInternalV2Table]] + * and the create/show-create table commands, exercised end-to-end through the V2 session + * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog = HoodieCatalog). + */ +class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase { + + private def hoodieCatalog: HoodieCatalog = + spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog] + + private def userFieldNames(names: Array[String]): Seq[String] = + names.filterNot(_.startsWith("_hoodie")).toSeq + + test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog API") { + withTempDir { tmp => + val catalog = hoodieCatalog + val tableName = generateTableName + val tablePath = s"${tmp.getCanonicalPath}/$tableName" + val ident = Identifier.of(Array("default"), tableName) + val schema = StructType(Seq( + StructField("id", IntegerType), + StructField("name", StringType), + StructField("ts", LongType))) + val props = Map( + "provider" -> "hudi", + "primaryKey" -> "id", + "preCombineField" -> "ts", + "location" -> tablePath).asJava + + // createTable routes to createHoodieTable(CREATE) and initializes the table on disk. + catalog.createTable(ident, schema, Array.empty[Transform], props) + assertTrue(catalog.tableExists(ident)) + assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists()) + + // loadTable returns a Hudi-backed table exposing the user schema. + val loaded = catalog.loadTable(ident) + assertEquals(Seq("id", "name", "ts"), userFieldNames(loaded.schema().fieldNames)) + + // alterTable: add a column. + catalog.alterTable(ident, TableChange.addColumn(Array("age"), IntegerType, true)) + assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age")) + + // alterTable: update a column comment. + catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"), "the name column")) + val commented = catalog.loadTable(ident).schema().fields.find(_.name == "name").get + assertEquals("the name column", commented.getComment().getOrElse("")) + + // alterTable: changing a column type is rejected by the V2 alter path, which routes to + // AlterHoodieTableChangeColumnCommand and does not support column type changes. + val typeChange = intercept[HoodieAnalysisException] { Review Comment: nit, feel free to ignore: since this test drives `HoodieCatalog.alterTable`, one `intercept[UnsupportedOperationException]` on `TableChange.setProperty(...)` asserting the change class name would also expose that `HoodieCatalog.scala:256` formats `${t.getClass}` on a `Class`, so the message is always `class java.lang.Class`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
