New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[HUDI-8824] MIT should error out for some assignment clause patterns #12584

Open

Davis-Zhang-Onehouse wants to merge 2 commits into apache:master from Davis-Zhang-Onehouse:HUDI-8824

+331 −56

Contributor

Davis-Zhang-Onehouse commented Jan 7, 2025 •

edited

Loading

Change Logs

For insert and update clause of MIT, in some cases if we didn't set the primary key/precombine field value explicitly, MIT errors out in the query analyze phase.

Also some changes that make sure the precombine key value data type is the same over source and target table (an irrelevant test maintenance)

Impact

Better user facing error message and guard against cases we don't support.

Risk level (write none, low medium or high below)

none

Documentation Update

We track MIT doc update all in https://issues.apache.org/jira/browse/HUDI-8527

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

github-actions bot added the size:L label

Davis-Zhang-Onehouse force-pushed the HUDI-8824 branch 3 times, most recently from 538edc9 to 441d8eb Compare

January 7, 2025 21:47

yihua reviewed

View reviewed changes

...udi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala Outdated

Comment on lines 485 to 491

+                  val isPartialUpdateAction = (targetTableType == MOR_TABLE_TYPE_OPT_VAL
+                    && UPSERT_OPERATION_OPT_VAL == getOperationType(parameters)
+                    && parameters.getOrElse(
+                    ENABLE_MERGE_INTO_PARTIAL_UPDATES.key,
+                    ENABLE_MERGE_INTO_PARTIAL_UPDATES.defaultValue.toString).toBoolean
+                    && updatingActions.nonEmpty)
+                  isPartialUpdateAction

Contributor

yihua Jan 8, 2025

nit: inline the statement

...udi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala Outdated

                     assert(insert.assignments.length <= targetTableSchema.length,
                       s"The number of insert assignments[${insert.assignments.length}] must be less than or equal to the " +
                         s"targetTable field size[${targetTableSchema.length}]"))
+                  // Precombine field and primary key field must be present in the assignment clause of all insert actions.

Contributor

yihua Jan 8, 2025

Suggested change

      
                // Precombine field and primary key field must be present in the assignment clause of all insert actions.
          
                // Precombine field and record key field must be present in the assignment clause of all insert actions.

...udi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala Outdated

+                      sparkSession.sessionState.conf.resolver,
+                      mergeInto.targetTable,
+                      Seq(field),
+                      "pre-combine field",

Contributor

yihua Jan 8, 2025

Suggested change

      
                    "pre-combine field",
          
                    "precombine field",

...udi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala Outdated

                   if (hoodieCatalogTable.preCombineKey.isEmpty && updateActions.nonEmpty) {
-                    logWarning(s"Updates without precombine can have nondeterministic behavior")
+                    logWarning(s"Updates without pre-combine can have nondeterministic behavior")

Contributor

yihua Jan 8, 2025

no need to change

...udi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala Outdated

                     assert(insert.assignments.length <= targetTableSchema.length,
                       s"The number of insert assignments[${insert.assignments.length}] must be less than or equal to the " +
                         s"targetTable field size[${targetTableSchema.length}]"))
+                  // Precombine field and primary key field must be present in the assignment clause of all insert actions.

Contributor

yihua Jan 8, 2025

I assume the insert action should always include all columns, so such check may not be necessary, i.e., missing any column would lead to insert failure already?

Contributor

yihua Jan 9, 2025

Let's keep the checks so it's strict.

yihua reviewed

View reviewed changes

...udi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala Outdated

+                        sparkSession.sessionState.conf.resolver,
+                        mergeInto.targetTable,
+                        hoodieCatalogTable.tableConfig.getRecordKeyFields.orElse(Array.empty),
+                        "primaryKey field",

Contributor

yihua Jan 8, 2025

use record key field to be consistent

...udi-spark/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala

+                  // Precombine field and primary key field must be present in the assignment clause of all insert actions.
+                  // Check has no effect if we don't have such fields in target table or we don't have insert actions
+                  insertActions.foreach(action =>
+                    hoodieCatalogTable.preCombineKey.foreach(

Contributor

yihua Jan 8, 2025

let's only validate precombine if the merge mode is EVENT_TIME_ORDERING.

hudi-spark-datasource/hudi-spark/src/test/resources/sql-statements.sql Outdated

@@ @@ -63,10 +63,10 @@ select id, name, price, cast(dt as string) from h0_p; @@
               # CREATE TABLE
               create table h1 (
-                id bigint,
+                id int,

Contributor

yihua Jan 8, 2025

Why do we need this change?

...urce/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestMergeIntoLogOnlyTable.scala Outdated

                          |  id int,
                          |  name string,
                          |  price double,
-                         |  ts long
+                         |  ts int

Contributor

yihua Jan 8, 2025

Similar here

...-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestMergeIntoTable.scala Outdated

                          |  id int,
                          |  name string,
                          |  value int,
-                         |  ts long
+                         |  ts int

Contributor

yihua Jan 8, 2025

Let's not change the types of precombine field.

...-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestMergeIntoTable.scala Outdated

                            |  id int,
                            |  name string,
                            |  price double,
-                           |  ts long
+                           |  ts int

Contributor

yihua Jan 8, 2025

Since we also have "(int) 0" as the natural ordering, making this integer can mask issues around precombine field and merging logic.

...-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestMergeIntoTable.scala Outdated

                         s"""
                            | merge into $tableName
                            | using (
-                           |  select 1 as id, 'a1' as name, 10 as price, $dataValue as c, '1' as flag
+                           |  select 1 as id, 'a1' as name, 10 as price, cast($dataValue as $dataType) as c, '1' as flag

Contributor

yihua Jan 8, 2025

Use 1L if there is any type mismatch in the tests. The user is expected to provide the data with the correct types.

...datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestMergeIntoTable2.scala Outdated

+                  Seq("cow", "mor").foreach { tableType =>
+                    withRecordType()(withTempDir { tmp =>
+                      log.info(s"Testing table type $tableType")
+                      spark.sql(s"set ${DataSourceWriteOptions.ENABLE_MERGE_INTO_PARTIAL_UPDATES.key} = false")

Contributor

yihua Jan 8, 2025

We should unset the config after the test or set the default of the config so that it does not affect other tests in the same Spark sessions (since this SET statement has global impact).

yihua reviewed

View reviewed changes

...datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestMergeIntoTable2.scala Outdated

+                      spark.sql(s"insert into $tableName values (1, 'a1', 10, 1000, '2021-03-21')")
+                      // Test 1: Update statements where at least one misses primary key assignment
+                      if (tableType.equals("mor")) {

Contributor

yihua Jan 8, 2025

Could we also test the same SQL statements on COW table and make sure they pass with data validation (i.e., the case where the new constraint does not apply)?


          [HUDI-8824] MIT error out on bad assignment clauses

92bb3d2

Davis-Zhang-Onehouse force-pushed the HUDI-8824 branch 3 times, most recently from 3f5a535 to 631494d Compare

January 9, 2025 18:41


          address PR comments

ffad811

Davis-Zhang-Onehouse force-pushed the HUDI-8824 branch from 631494d to ffad811 Compare

January 10, 2025 00:03

hudi-bot commented Jan 10, 2025

CI report:

ffad811 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

size:L