This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 8c44ce5a8559 test(spark): add HiveSyncProcedure coverage (#19407)
8c44ce5a8559 is described below
commit 8c44ce5a85597e7137a6c676fe706b9a62a307bb
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Jul 31 04:32:47 2026 -0700
test(spark): add HiveSyncProcedure coverage (#19407)
* test(spark): add HiveSyncProcedure coverage
Add TestHiveSyncProcedure, which invokes the hive_sync stored procedure
with every optional argument set and an unresolvable partition extractor,
covering the argument handling, config assembly, and failure wrapping in the
procedure's call method. It pins that the failure is surfaced as a
HoodieException carrying the "hive sync failed" message and originates from
loading the supplied extractor.
* Assert the exception type and written conf in the hive_sync test
Use assume() so the test reports as skipped, not passed, on Java 17. Assert
a HoodieException appears in the cause chain rather than matching the message
alone. Assert every session and hadoop conf value the procedure writes before
the finally clears them, so dropping any setConfString call is caught. Switch
the manual try/catch to intercept.
---
.../sql/hudi/procedure/TestHiveSyncProcedure.scala | 113 +++++++++++++++++++++
1 file changed, 113 insertions(+)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHiveSyncProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHiveSyncProcedure.scala
new file mode 100644
index 000000000000..26719e78b141
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHiveSyncProcedure.scala
@@ -0,0 +1,113 @@
+/*
+ * 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.procedure
+
+import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hive.{HiveSyncConfig, HiveSyncConfigHolder}
+import org.apache.hudi.sync.common.HoodieSyncConfig
+
+import org.apache.hadoop.hive.conf.HiveConf
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+class TestHiveSyncProcedure extends HoodieSparkProcedureTestBase {
+
+ // Session-conf keys the procedure writes from its optional arguments; reset
after the call so
+ // they do not leak into sibling tests sharing the session.
+ private val sessionSyncConfKeys = Seq(
+ HiveSyncConfig.HIVE_USER.key, HiveSyncConfig.HIVE_PASS.key,
HiveSyncConfig.HIVE_USE_JDBC.key,
+ HiveSyncConfigHolder.HIVE_SYNC_MODE.key,
HoodieSyncConfig.META_SYNC_PARTITION_FIELDS.key,
+ HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key,
+ HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY.key,
HoodieSyncConfig.META_SYNC_INCREMENTAL.key)
+
+ private def createHudiTable(tableName: String, basePath: String): Unit = {
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | options (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts'
+ | )
+ | location '$basePath'
+ """.stripMargin)
+ spark.sql(s"insert into $tableName values(1, 'a1', 10.0, 1000)")
+ }
+
+ test("Test call hive_sync procedure surfaces a sync failure as
HoodieException") {
+ // The procedure constructs a HiveConf; Hive 2.3.7 is compiled with Java
1.8 and its class loader
+ // throws when the Hive APIs are exercised on Java 17. Cancel (report as
skipped, not passed) there.
+ assume(HoodieTestUtils.getJavaVersion < 17)
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ createHudiTable(tableName, basePath)
+
+ try {
+ // Every optional argument is supplied so each argument branch of the
procedure runs. The
+ // unresolvable partition_extractor_class makes the HiveSyncTool
constructor throw while
+ // loading the extractor (in the HoodieSyncClient constructor, before
any metastore
+ // connection is attempted).
+ val thrown = intercept[Throwable] {
+ spark.sql(
+ s"call hive_sync(table => '$tableName', metastore_uri =>
'thrift://localhost:9083'," +
+ s" username => 'hive', password => 'hive', use_jdbc => 'false',
mode => 'hms'," +
+ s" partition_fields => 'name', strategy => 'ALL',
sync_incremental => 'false'," +
+ s" partition_extractor_class =>
'org.apache.hudi.hive.MissingPartitionExtractor')")
+ }
+
+ // The procedure must wrap the failure as HoodieException("hive sync
failed"), and the cause
+ // must be the unresolvable extractor supplied as an argument.
Asserting the exception type,
+ // the wrapping message, and the cause together pins the wrapping
contract and that the
+ // supplied extractor is applied and fails before the metastore.
+ val causes = Iterator.iterate(thrown)(_.getCause).takeWhile(_ !=
null).toList
+ assertTrue(causes.exists(_.isInstanceOf[HoodieException]),
+ "expected a HoodieException in the cause chain, but got: " +
+ causes.map(_.getClass.getName).mkString(" -> "))
+ val chain = causes.flatMap(t => Option(t.getMessage)).mkString(" | ")
+ assertTrue(chain.contains("hive sync failed"),
+ s"expected the procedure to wrap the failure as HoodieException, but
got: $chain")
+
assertTrue(chain.contains("org.apache.hudi.hive.MissingPartitionExtractor"),
+ s"expected the failure to originate from loading the supplied
extractor, but got: $chain")
+
+ // Each supplied optional argument must be applied to the
session/hadoop conf before the sync
+ // is attempted; asserting the written values catches dropping any of
the setConfString calls.
+ val sqlConf = spark.sessionState.conf
+ assertEquals("hive",
sqlConf.getConfString(HiveSyncConfig.HIVE_USER.key))
+ assertEquals("hive",
sqlConf.getConfString(HiveSyncConfig.HIVE_PASS.key))
+ assertEquals("false",
sqlConf.getConfString(HiveSyncConfig.HIVE_USE_JDBC.key))
+ assertEquals("hms",
sqlConf.getConfString(HiveSyncConfigHolder.HIVE_SYNC_MODE.key))
+ assertEquals("name",
sqlConf.getConfString(HoodieSyncConfig.META_SYNC_PARTITION_FIELDS.key))
+ assertEquals("org.apache.hudi.hive.MissingPartitionExtractor",
+
sqlConf.getConfString(HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key))
+ assertEquals("ALL",
sqlConf.getConfString(HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY.key))
+ assertEquals("false",
sqlConf.getConfString(HoodieSyncConfig.META_SYNC_INCREMENTAL.key))
+ assertEquals("thrift://localhost:9083",
+
spark.sparkContext.hadoopConfiguration.get(HiveConf.ConfVars.METASTOREURIS.varname))
+ } finally {
+
spark.sparkContext.hadoopConfiguration.unset(HiveConf.ConfVars.METASTOREURIS.varname)
+ sessionSyncConfKeys.foreach(spark.sessionState.conf.unsetConf)
+ }
+ }
+ }
+}
