voonhous commented on code in PR #19163: URL: https://github.com/apache/hudi/pull/19163#discussion_r3665007111
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestAlterTableColumnCoverage.scala: ########## @@ -0,0 +1,111 @@ +/* + * 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.ddl + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Column add / rename / drop / position / comment coverage for + * [[org.apache.spark.sql.hudi.command.AlterTableCommand]] driven through the schema-on-read + * (schema evolution) path. Uses an unpartitioned table so the `commitWithSchema` data-schema + * derivation takes the no-partition-column branch, complementing the partitioned coverage in + * [[TestSpark3DDL]]. + */ +class TestAlterTableColumnCoverage extends HoodieSparkSqlTestBase { + + test("Test alter table add/rename/drop/position/comment on unpartitioned table") { + withSQLConf("hoodie.schema.on.read.enable" -> "true") { + withTempDir { tmp => + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | price double, + | ts long + |) using hudi + | location '${tmp.getCanonicalPath}' + | tblproperties (type = 'cow', primaryKey = 'id', preCombineField = 'ts') + """.stripMargin) + spark.sql(s"insert into $tableName values (1, 'a1', 10.0, 1000)") + + // ADD columns (first alter -> empty history schema branch), with a comment and an + // explicit position. + spark.sql(s"alter table $tableName add columns (age int comment 'the age' after name)") + spark.sql(s"insert into $tableName values (2, 'a2', 25, 20.0, 2000)") + checkAnswer(s"select id, name, age, price, ts from $tableName order by id")( + Seq(1, "a1", null, 10.0, 1000L), + Seq(2, "a2", 25, 20.0, 2000L) + ) + val addedSchema = spark.sessionState.catalog + .getTableMetadata(TableIdentifier(tableName)).schema + assert(addedSchema.exists(_.name == "age")) + assert(addedSchema.find(_.name == "age").get.getComment().contains("the age")) + + // ALTER column type widening (int -> long); second alter -> non-empty history branch. + spark.sql(s"alter table $tableName alter column age type long") + assert(spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName)) + .schema.find(_.name == "age").get.dataType.typeName == "long") + + checkAnswer(s"select id, name, age from $tableName order by id")( + Seq(1, "a1", null), + Seq(2, "a2", 25L) + ) + + // ALTER column comment on an existing column. + spark.sql(s"alter table $tableName alter column price comment 'unit price'") + assert(spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName)) + .schema.find(_.name == "price").get.getComment().contains("unit price")) + + // NOTE: RENAME COLUMN and DROP COLUMN (applyDeleteAction) are not exercised here. On this Review Comment: **Correctness / blocker.** This NOTE attributes the RENAME/DROP failure to a missing "table-service config recipe", but the branch history says these tests were removed because they *failed*: `c93ab6c62dc8` dropped RENAME with the message "fails across Spark versions", then `873303549cb3` dropped DROP. The cause looks like a live regression, not a config requirement. `#13595` (`0fe119a0cf1e`, an ancestor of this branch) added a bare `hoodieTable.validateSchema()` to `AlterTableCommand.commitWithSchema`, and **in the same commit** added `hoodie.datasource.write.schema.allow.auto.evolution.column.drop -> true` to four pre-existing `TestSpark3DDL` rename/drop tests. Pre-existing tests needing a new config to keep passing is the signature of a behaviour change. Net effect: plain `ALTER TABLE ... RENAME COLUMN` / `DROP COLUMN` on a schema-on-read table fails with `MissingSchemaFieldException` under shipped defaults (`hoodie.avro.schema.validate=false`, `allow.auto.evolution.column.drop=false`) since 2025-07-25. Please file a GitHub issue against `AlterTableCommand.commitWithSchema` / `#13595` and link it from this PR, and delete this NOTE instead of merging it. As written it records a product bug as intended behaviour, which is what will keep it from being found. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestAlterTableColumnCoverage.scala: ########## @@ -0,0 +1,111 @@ +/* + * 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.ddl + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Column add / rename / drop / position / comment coverage for + * [[org.apache.spark.sql.hudi.command.AlterTableCommand]] driven through the schema-on-read + * (schema evolution) path. Uses an unpartitioned table so the `commitWithSchema` data-schema Review Comment: **Cleanliness / major.** This file's stated reason to exist is not accurate. The scaladoc says it "Uses an unpartitioned table ... complementing the partitioned coverage in `TestSpark3DDL`", but `TestSpark3DDL:584` ("Test alter column by add rename and drop") is **already unpartitioned** -- there is no `partitioned by` in its DDL -- with the identical `(id int, name string, price double, ts long)` schema, and it runs over both `cow` and `mor` while this file is COW-only. It also covers strictly more: `alter column id type long` (`:611`), `add columns(ext1 string comment ... after name)` (`:616`), plus rename, drop, and `validateInternalSchema` assertions. The second test here is a subset too (see the comment on line 87). Also, `AlterTableCommand.scala:306` is a `.filter` over the partition-column list, not a branch, so the "no-partition-column branch" this file claims to reach does not exist. Suggest deleting this file. Nothing in it is uncovered. (The regression noted on line 77 still needs its own issue regardless.) ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeIntoWriteCoverage.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.dml.others + +import org.apache.hudi.DataSourceWriteOptions.SPARK_SQL_OPTIMIZED_WRITES + +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Focused coverage for the MERGE INTO write path implemented by + * [[org.apache.spark.sql.hudi.command.payload.ExpressionPayload]]. + * + * The scenarios below intentionally combine, within a single statement, all of: + * - conditional matched UPDATE clauses (the update-condition evaluation loop), + * - matched DELETE clauses (both conditional and unconditional delete markers), + * - conditional NOT-MATCHED INSERT clauses (including a not-matched row that + * matches no insert condition and is therefore filtered), + * so both the matched and not-matched evaluators, the delete-marker branch, and the + * record-merge branch are exercised, against partitioned COW and MOR tables with the + * row-writer path both enabled and disabled. + * + * Every table here declares a primary key, so `buildMergeIntoConfig` selects + * [[org.apache.spark.sql.hudi.command.SqlKeyGenerator]]. + * [[org.apache.spark.sql.hudi.command.MergeIntoKeyGenerator]] is the primary-keyless + * path and is covered by [[TestMergeIntoTableWithNonRecordKeyField]]. + */ +class TestMergeIntoWriteCoverage extends HoodieSparkSqlTestBase { + + test("Test MergeInto conditional update, delete-marker and insert on partitioned COW") { + Seq(true, false).foreach { optimizedWrites => Review Comment: **Correctness / major.** This `Seq(true, false)` loop runs the suite's heaviest test twice with no behavioural difference, so it doubles the cost for zero discriminating power. `SPARK_SQL_OPTIMIZED_WRITES` reaches the MERGE path in exactly one place, `MergeIntoHoodieTableCommand:200`, and that branch is gated on `resolving.isEmpty`. All four tests match `on t.id = s.id` against `primaryKey = 'id'`, so `resolving` is never empty and the flag is inert under both values. Every other use of the config is in `UpdateHoodieTableCommand` / `DeleteHoodieTableCommand`, not MERGE. The scaladoc also describes this config as "the row-writer path". It is documented as "Controls whether spark sql prepped update and delete are enabled" (`DataSourceOptions.scala:806`); `ENABLE_ROW_WRITER` (`hoodie.datasource.write.row.writer.enable`) is a different config. Drop the loop and the "row-writer path both enabled and disabled" sentence from the scaladoc. If you want genuine coverage of the flag, add a test whose ON clause omits a record-key column and assert the `HoodieAnalysisException` fires only when it is `true`. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestBulkInsertRowWriterCommitCoverage.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.dml.insert + +import org.apache.hudi.HoodieCLIUtils +import org.apache.hudi.common.util.{Option => HOption} + +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Focused coverage for the row-writer bulk-insert commit executors in + * `org.apache.hudi.commit`: [[org.apache.hudi.commit.BaseDatasetBulkInsertCommitActionExecutor]] + * and [[org.apache.hudi.commit.DatasetBulkInsertOverwriteCommitActionExecutor]]. + * + * All tests set `hoodie.spark.sql.insert.into.operation = bulk_insert` and keep the default + * row-writer path enabled, so INSERT / INSERT OVERWRITE flow through the Dataset-based executors. + */ +class TestBulkInsertRowWriterCommitCoverage extends HoodieSparkSqlTestBase { + + test("Test row-writer bulk_insert into partitioned table") { + withSQLConf( + "hoodie.metadata.enable" -> "false", + "hoodie.spark.sql.insert.into.operation" -> "bulk_insert", + "hoodie.bulkinsert.shuffle.parallelism" -> "1") { + Seq("cow", "mor").foreach { tableType => + withTempDir { tmp => + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | price double, + | dt string + |) using hudi + | partitioned by (dt) + | location '${tmp.getCanonicalPath}' + | tblproperties (type = '$tableType', primaryKey = 'id') + """.stripMargin) + + spark.sql( + s"""insert into $tableName values + | (1, 'a1', 10.0, '2024-01-01'), + | (2, 'a2', 20.0, '2024-01-01'), + | (3, 'a3', 30.0, '2024-01-02') + """.stripMargin) + + checkAnswer(s"select id, name, price, dt from $tableName order by id")( + Seq(1, "a1", 10.0, "2024-01-01"), + Seq(2, "a2", 20.0, "2024-01-01"), + Seq(3, "a3", 30.0, "2024-01-02") + ) + } + } + } + } + + test("Test row-writer insert overwrite with dynamic partitions") { + // Dynamic overwrite mode is required so that only the partitions present in the + // incoming data are replaced; the default (static) mode overwrites the whole table. + withSQLConf( + "hoodie.metadata.enable" -> "false", + "hoodie.spark.sql.insert.into.operation" -> "bulk_insert", + "hoodie.datasource.overwrite.mode" -> "dynamic") { + withTempDir { tmp => + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | dt string + |) using hudi + | partitioned by (dt) + | location '${tmp.getCanonicalPath}' + | tblproperties (type = 'cow', primaryKey = 'id') + """.stripMargin) + + spark.sql( + s"""insert into $tableName values + | (1, 'a1', '2024-01-01'), + | (2, 'a2', '2024-01-02') + """.stripMargin) + + // Dynamic insert overwrite: only the partitions present in the incoming data + // (2024-01-01) are replaced; 2024-01-02 must survive. + spark.sql( + s"""insert overwrite table $tableName partition (dt) + | select 1 as id, 'a1_new' as name, '2024-01-01' as dt union all + | select 3 as id, 'a3' as name, '2024-01-01' as dt + """.stripMargin) + + checkAnswer(s"select id, name, dt from $tableName order by id")( + Seq(1, "a1_new", "2024-01-01"), + Seq(2, "a2", "2024-01-02"), + Seq(3, "a3", "2024-01-01") + ) + } + } + } + + test("Test row-writer insert overwrite with static partition") { Review Comment: **Cleanliness / major.** This duplicates `TestInsertTable2:644`, which sets the same `SPARK_SQL_INSERT_INTO_OPERATION -> BULK_INSERT`, seeds two partitions, runs `insert overwrite table t partition(dt = '...')`, and asserts the untouched partition survives -- and additionally loops cow+mor and asserts the `INSERT_OVERWRITE` operation type. This version is COW-only and drops the op-type assertion. Suggest deleting this test. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestBulkInsertRowWriterCommitCoverage.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.dml.insert + +import org.apache.hudi.HoodieCLIUtils +import org.apache.hudi.common.util.{Option => HOption} + +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Focused coverage for the row-writer bulk-insert commit executors in + * `org.apache.hudi.commit`: [[org.apache.hudi.commit.BaseDatasetBulkInsertCommitActionExecutor]] + * and [[org.apache.hudi.commit.DatasetBulkInsertOverwriteCommitActionExecutor]]. + * + * All tests set `hoodie.spark.sql.insert.into.operation = bulk_insert` and keep the default + * row-writer path enabled, so INSERT / INSERT OVERWRITE flow through the Dataset-based executors. + */ +class TestBulkInsertRowWriterCommitCoverage extends HoodieSparkSqlTestBase { + + test("Test row-writer bulk_insert into partitioned table") { + withSQLConf( + "hoodie.metadata.enable" -> "false", + "hoodie.spark.sql.insert.into.operation" -> "bulk_insert", + "hoodie.bulkinsert.shuffle.parallelism" -> "1") { + Seq("cow", "mor").foreach { tableType => + withTempDir { tmp => + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | price double, + | dt string + |) using hudi + | partitioned by (dt) + | location '${tmp.getCanonicalPath}' + | tblproperties (type = '$tableType', primaryKey = 'id') + """.stripMargin) + + spark.sql( + s"""insert into $tableName values + | (1, 'a1', 10.0, '2024-01-01'), + | (2, 'a2', 20.0, '2024-01-01'), + | (3, 'a3', 30.0, '2024-01-02') + """.stripMargin) + + checkAnswer(s"select id, name, price, dt from $tableName order by id")( Review Comment: **Correctness / major.** This asserts row contents only, so it passes unchanged whether the write went through `bulk_insert`, `insert`, or `upsert`. The suite's whole stated value is covering the row-writer commit executors, and nothing pins that they ran. `TestInsertTable2:230` covers this same scenario (same DDL, same cow/mor loop) and does pin it, via `getLastCommitMetadata`. Add the operation-type assertion; the helper already exists at `HoodieSparkSqlTestBase:405`: ```suggestion assertResult(WriteOperationType.BULK_INSERT) { getLastCommitMetadata(spark, tmp.getCanonicalPath).getOperationType } checkAnswer(s"select id, name, price, dt from $tableName order by id")( ``` ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestBulkInsertRowWriterCommitCoverage.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.dml.insert + +import org.apache.hudi.HoodieCLIUtils +import org.apache.hudi.common.util.{Option => HOption} + +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Focused coverage for the row-writer bulk-insert commit executors in + * `org.apache.hudi.commit`: [[org.apache.hudi.commit.BaseDatasetBulkInsertCommitActionExecutor]] + * and [[org.apache.hudi.commit.DatasetBulkInsertOverwriteCommitActionExecutor]]. + * + * All tests set `hoodie.spark.sql.insert.into.operation = bulk_insert` and keep the default + * row-writer path enabled, so INSERT / INSERT OVERWRITE flow through the Dataset-based executors. + */ +class TestBulkInsertRowWriterCommitCoverage extends HoodieSparkSqlTestBase { + + test("Test row-writer bulk_insert into partitioned table") { Review Comment: **Cleanliness / major.** This duplicates `TestInsertTable2:230` ("Test bulk insert with insert into for single partitioned table"): same `(id int, name string, price double, dt string)` DDL, same `partitioned by (dt)`, same `Seq("cow","mor")` loop -- and that one additionally asserts `WriteOperationType.BULK_INSERT`. The one thing this version touches that the existing test does not is the config entry point: `hoodie.spark.sql.insert.into.operation` goes through `ProvidesHoodieConfig:231` while `hoodie.sql.bulk.insert.enable` goes through `:228`. But with only a `checkAnswer` this test cannot detect a regression in either branch, so that distinction buys nothing as written. Suggest deleting this test. If you keep it, it needs the assertion suggested on line 65 to be worth its runtime. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestBulkInsertRowWriterCommitCoverage.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.dml.insert + +import org.apache.hudi.HoodieCLIUtils +import org.apache.hudi.common.util.{Option => HOption} + +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Focused coverage for the row-writer bulk-insert commit executors in + * `org.apache.hudi.commit`: [[org.apache.hudi.commit.BaseDatasetBulkInsertCommitActionExecutor]] + * and [[org.apache.hudi.commit.DatasetBulkInsertOverwriteCommitActionExecutor]]. + * + * All tests set `hoodie.spark.sql.insert.into.operation = bulk_insert` and keep the default + * row-writer path enabled, so INSERT / INSERT OVERWRITE flow through the Dataset-based executors. + */ +class TestBulkInsertRowWriterCommitCoverage extends HoodieSparkSqlTestBase { + + test("Test row-writer bulk_insert into partitioned table") { + withSQLConf( + "hoodie.metadata.enable" -> "false", + "hoodie.spark.sql.insert.into.operation" -> "bulk_insert", + "hoodie.bulkinsert.shuffle.parallelism" -> "1") { + Seq("cow", "mor").foreach { tableType => + withTempDir { tmp => + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | price double, + | dt string + |) using hudi + | partitioned by (dt) + | location '${tmp.getCanonicalPath}' + | tblproperties (type = '$tableType', primaryKey = 'id') + """.stripMargin) + + spark.sql( + s"""insert into $tableName values + | (1, 'a1', 10.0, '2024-01-01'), + | (2, 'a2', 20.0, '2024-01-01'), + | (3, 'a3', 30.0, '2024-01-02') + """.stripMargin) + + checkAnswer(s"select id, name, price, dt from $tableName order by id")( + Seq(1, "a1", 10.0, "2024-01-01"), + Seq(2, "a2", 20.0, "2024-01-01"), + Seq(3, "a3", 30.0, "2024-01-02") + ) + } + } + } + } + + test("Test row-writer insert overwrite with dynamic partitions") { + // Dynamic overwrite mode is required so that only the partitions present in the + // incoming data are replaced; the default (static) mode overwrites the whole table. + withSQLConf( + "hoodie.metadata.enable" -> "false", + "hoodie.spark.sql.insert.into.operation" -> "bulk_insert", + "hoodie.datasource.overwrite.mode" -> "dynamic") { + withTempDir { tmp => + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | dt string + |) using hudi + | partitioned by (dt) + | location '${tmp.getCanonicalPath}' + | tblproperties (type = 'cow', primaryKey = 'id') + """.stripMargin) + + spark.sql( + s"""insert into $tableName values + | (1, 'a1', '2024-01-01'), + | (2, 'a2', '2024-01-02') + """.stripMargin) + + // Dynamic insert overwrite: only the partitions present in the incoming data + // (2024-01-01) are replaced; 2024-01-02 must survive. + spark.sql( + s"""insert overwrite table $tableName partition (dt) + | select 1 as id, 'a1_new' as name, '2024-01-01' as dt union all + | select 3 as id, 'a3' as name, '2024-01-01' as dt + """.stripMargin) + + checkAnswer(s"select id, name, dt from $tableName order by id")( + Seq(1, "a1_new", "2024-01-01"), + Seq(2, "a2", "2024-01-02"), + Seq(3, "a3", "2024-01-01") + ) + } + } + } + + test("Test row-writer insert overwrite with static partition") { + withSQLConf( + "hoodie.metadata.enable" -> "false", + "hoodie.spark.sql.insert.into.operation" -> "bulk_insert") { + withTempDir { tmp => + val tableName = generateTableName + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | dt string + |) using hudi + | partitioned by (dt) + | location '${tmp.getCanonicalPath}' + | tblproperties (type = 'cow', primaryKey = 'id') + """.stripMargin) + + spark.sql( + s"""insert into $tableName values + | (1, 'a1', '2024-01-01'), + | (2, 'a2', '2024-01-01'), + | (3, 'a3', '2024-01-02') + """.stripMargin) + + // Static partition spec -> STATIC_OVERWRITE_PARTITION_PATHS drives the static + // branch of getPartitionToReplacedFileIds. Only 2024-01-01 is replaced. + spark.sql( + s"""insert overwrite table $tableName partition (dt = '2024-01-01') + | select 9 as id, 'a9' as name + """.stripMargin) + + checkAnswer(s"select id, name, dt from $tableName order by id")( + Seq(3, "a3", "2024-01-02"), + Seq(9, "a9", "2024-01-01") + ) + } + } + } + + test("Test row-writer insert overwrite rejected when overlapping pending clustering") { Review Comment: **Correctness / major (coverage gap on a fragile branch).** This is the most valuable test in the PR -- it is the only coverage anywhere for `rejectIfOverlappingPendingClustering` on the Dataset/row-writer path. `#18829` (`e2cf7216fddf`) shipped that path together with `TestInsertOverwriteWithClustering.java`, but that suite only drives `client.insertOverwrite(...)`, the RDD path (no match for `DatasetBulkInsert` / `ENABLE_ROW_WRITER` in it). Worth keeping. But as written it reaches only the narrowest slice of the code it guards: 1. **Static mode only.** No `hoodie.datasource.overwrite.mode` is set, so Spark defaults to STATIC and `deduceOverwriteConfig` populates `STATIC_OVERWRITE_PARTITION_PATHS`, taking the static arm of `DatasetBulkInsertOverwriteCommitActionExecutor.resolveTargetPartitions`. The dynamic arm -- which reads `_hoodie_partition_path` off `preparedRecords` and carries a production comment warning about invocation ordering vs `prepareForBulkInsert` -- is never executed with a pending clustering plan. 2. **Single partition.** The table has only `2024-01-01`, so the test passes even if the overlap check returned every file group in the table. The RDD suite guards this with `TestInsertOverwriteWithClustering.java:338` (`testInsertOverwriteNonOverlappingPartitionWithPendingClustering`); there is no row-writer counterpart. 3. **Partitioned only.** The unpartitioned arm of `resolveTargetPartitions`, also added by `#18829` four weeks ago, has zero coverage repo-wide. Please add a second partition plus an assertion that overwriting the non-clustered partition still succeeds, and parameterize over `hoodie.datasource.overwrite.mode` static/dynamic. That converts this from "reaches the code" into "would catch a regression". ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala: ########## @@ -119,6 +119,11 @@ class HoodieSparkSqlTestBase extends FunSuite with BeforeAndAfterAll { catalog.dropTable(table, true, true) } } + // The INMEMORY index keeps a JVM-static record-location map; reset it after every test so + // stale keys from an earlier test cannot misroute writes in a later one. withRecordType + // clears it between record-type iterations, but only on success and only for tests that use + // it, so a throwing or non-withRecordType INMEMORY test would otherwise leak state here. + HoodieInMemoryHashIndex.clear() Review Comment: **Correctness / nit, optional.** The reset sits after the catalog-drop loop inside the same `finally`. If `catalog.listDatabases()` or `dropTable` throws, the reset is skipped -- which is exactly the leak-on-failure case the comment above says it closes. I could not construct a case where those actually throw for these suites (tables use an explicit `location`, so Spark treats them as EXTERNAL and skips data deletion), so this is defensive only. Moving `HoodieInMemoryHashIndex.clear()` to the first statement in the `finally`, or wrapping the drop loop in its own `try`/`finally`, would make it hold unconditionally. For what it is worth, I checked this change fairly hard and it is safe: `clear()` is null-guarded (`HoodieInMemoryHashIndex.java:130`), suites run sequentially (`forkMode=once`, no `<parallel>` anywhere, no `ParallelTestExecution`), and no suite relies on index state surviving across `test()` blocks. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/others/TestMergeIntoWriteCoverage.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.dml.others + +import org.apache.hudi.DataSourceWriteOptions.SPARK_SQL_OPTIMIZED_WRITES + +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Focused coverage for the MERGE INTO write path implemented by Review Comment: **Cleanliness / major.** All four tests in this file already exist elsewhere, in suites that assert strictly more: - `:46` conditional update + delete + filtered insert on COW -> `TestMergeIntoTable:43`, same schema and same optimized-writes loop, plus `validateTableSchema` per step - `:112` unconditional delete marker -> `TestMergeIntoTable:863`, `:875`, `:892`, `:907` - `:152` MOR conditional update/delete/insert -> `TestMergeIntoTable:477` - `:201` `OverwriteWithLatestAvroPayload` -> `TestMergeModeCommitTimeOrdering`, which sets that payload class at `:53` and `:71` and runs `merge into` at `:295`, `:309`, `:340`; `TestMergeIntoTable2:1251` also parametrizes cow/mor x COMMIT_TIME/EVENT_TIME, so it can distinguish the two modes where a single-mode test cannot Suggest deleting this file. Separately, if the goal is `ExpressionPayload` coverage, the branches that have actually broken before are shape-driven -- nested struct (#3379), decimal (#3224), string ordering field (#3099) -- and every test here uses flat scalars only, so none of them are reached. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestAlterTableColumnCoverage.scala: ########## @@ -0,0 +1,111 @@ +/* + * 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.ddl + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase + +/** + * Column add / rename / drop / position / comment coverage for + * [[org.apache.spark.sql.hudi.command.AlterTableCommand]] driven through the schema-on-read + * (schema evolution) path. Uses an unpartitioned table so the `commitWithSchema` data-schema + * derivation takes the no-partition-column branch, complementing the partitioned coverage in + * [[TestSpark3DDL]]. + */ +class TestAlterTableColumnCoverage extends HoodieSparkSqlTestBase { + + test("Test alter table add/rename/drop/position/comment on unpartitioned table") { Review Comment: **Cleanliness / nit, optional.** The test name and the PR description both advertise rename and drop, which the file's own NOTE at line 77 says are not exercised. Same overstated-coverage pattern cshuo caught on `MergeIntoKeyGenerator`. Moot if the file goes away; otherwise: ```suggestion test("Test alter table add/type-widen/comment on unpartitioned table") { ``` -- 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]
