voonhous commented on code in PR #19162: URL: https://github.com/apache/hudi/pull/19162#discussion_r3886099325
########## 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: Both CTAS tests are dropped in 39aaae78dff6 and the scaladoc no longer mentions `HoodieStagedTable`. A staged-table test needs `HoodieCatalog` registered under a second catalog name with its delegate wired (the shape `TestPolarisHoodieCatalogDelegation` uses), so it is left out of this PR; the abort-deletes-existing-data path noted above remains unverified. ########## 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: Dropped in 39aaae78dff6; the explicit-`location` variant now lives inside `TestCreateTable` "Test CTAS using an illegal definition" next to the managed case. -- 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]
