This is an automated email from the ASF dual-hosted git repository.
danny0405 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 d6c871bc4d71 fix(hive-sync): fail HiveQL-mode sync when Hive rejects a
statement (#20145)
d6c871bc4d71 is described below
commit d6c871bc4d710d1ead43970adcdd7b952e28c490
Author: Shihuan Liu <[email protected]>
AuthorDate: Wed Sep 30 03:45:47 2026 -0700
fix(hive-sync): fail HiveQL-mode sync when Hive rejects a statement (#20145)
* fix(hive-sync): fail HiveQL-mode sync when Hive rejects a statement
Hive's Driver.run() reports compile and execution failures through a
non-zero CommandProcessorResponse rather than by throwing. HiveQL-mode
sync discarded the response on both the sequential path and the
HiveDriverPool path, so a rejected ALTER TABLE ... ADD PARTITION or
REPLACE COLUMNS was logged by Hive and otherwise ignored: the sync
reported success, advanced last_commit_time_sync past the failed DDL,
and recreate_table_on_error never triggered.
Check the response code for every statement and throw
HoodieHiveSyncException with Hive's error message, SQLState and the
(abbreviated) statement, matching how the JDBC and HMS modes already
fail on SQLException and TException.
* Let the sequential path wrap the rejection like the pool path does
* Rename HiveDriverUtil.runOrThrow to HiveStatementExecutor.executeOrThrow
* Drop the unreachable null check on the session Driver
* Use ExceptionUtil.validateErrorMsg in the HiveQL rejection test
---
.../apache/hudi/hive/ddl/HiveQueryDDLExecutor.java | 10 ++--
.../org/apache/hudi/hive/util/HiveDriverPool.java | 4 +-
.../hudi/hive/util/HiveStatementExecutor.java | 63 ++++++++++++++++++++++
.../org/apache/hudi/hive/TestHiveSyncTool.java | 29 ++++++++++
.../hive/ddl/TestHiveQueryDDLExecutorSession.java | 35 +++++++++++-
.../apache/hudi/hive/util/TestHiveDriverPool.java | 55 ++++++++++++++++---
6 files changed, 180 insertions(+), 16 deletions(-)
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
index 32ebee0ead00..ec674db4daba 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
@@ -26,6 +26,7 @@ import org.apache.hudi.hive.HoodieHiveSyncException;
import org.apache.hudi.hive.util.HiveDriverPool;
import org.apache.hudi.hive.util.HiveMetaStoreClientPool;
import org.apache.hudi.hive.util.HivePartitionUtil;
+import org.apache.hudi.hive.util.HiveStatementExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -34,7 +35,6 @@ import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.Driver;
import org.apache.hadoop.hive.ql.metadata.Hive;
-import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.security.UserGroupInformation;
@@ -181,17 +181,14 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
* bound for the duration, and the thread handed back as found: it may
belong to another executor
* or to an application that embeds this sync and holds a session of its own.
*/
- private List<CommandProcessorResponse> updateHiveSQLs(List<String> sqls) {
- List<CommandProcessorResponse> responses = new ArrayList<>();
+ private void updateHiveSQLs(List<String> sqls) {
HoodieTimer timer = HoodieTimer.start();
SessionState previousSession = SessionState.get();
ClassLoader previousLoader =
Thread.currentThread().getContextClassLoader();
try {
SessionState.setCurrentSessionState(sessionState);
for (String sql : sqls) {
- if (hiveDriver != null) {
- responses.add(hiveDriver.run(sql));
- }
+ HiveStatementExecutor.executeOrThrow(hiveDriver, sql);
}
} catch (Exception e) {
throw new HoodieHiveSyncException("Failed in executing SQL", e);
@@ -199,7 +196,6 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
restoreThread(previousSession, previousLoader);
}
log.info("Executed {} SQL statements sequentially in {} ms", sqls.size(),
timer.endTimer());
- return responses;
}
//TODO Duplicating it here from HMSDLExecutor as HiveQueryQL has no way of
doing it on its own currently. Need to refactor it
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
index e73eee19cf5e..303af59e699c 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
@@ -122,7 +122,7 @@ public class HiveDriverPool implements AutoCloseable {
for (Worker worker : workers) {
dispatch.add(worker.executor.submit(dispatch.guard(() -> {
for (String sql : setupSqls) {
- worker.driver.run(sql);
+ HiveStatementExecutor.executeOrThrow(worker.driver, sql);
}
return null;
}, "Skipped after an earlier setup statement failed")));
@@ -153,7 +153,7 @@ public class HiveDriverPool implements AutoCloseable {
String sql = sqls.get(i);
Worker worker = workers.get(i % workers.size());
dispatch.add(worker.executor.submit(dispatch.guard(() -> {
- worker.driver.run(sql);
+ HiveStatementExecutor.executeOrThrow(worker.driver, sql);
return null;
}, "Skipped after an earlier statement failed")));
}
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveStatementExecutor.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveStatementExecutor.java
new file mode 100644
index 000000000000..b852a00ce3ef
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveStatementExecutor.java
@@ -0,0 +1,63 @@
+/*
+ * 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.hudi.hive.util;
+
+import org.apache.hudi.hive.HoodieHiveSyncException;
+
+import org.apache.hadoop.hive.ql.CommandNeedRetryException;
+import org.apache.hadoop.hive.ql.Driver;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
+
+/**
+ * Runs HiveQL through a Hive {@link Driver}.
+ */
+public final class HiveStatementExecutor {
+
+ /**
+ * Batched partition statements can run to many kilobytes; the error only
needs enough of the
+ * statement to identify it.
+ */
+ static final int MAX_STATEMENT_LENGTH_IN_ERROR = 1000;
+
+ private HiveStatementExecutor() {
+ }
+
+ /**
+ * Runs the statement and throws if Hive rejects it. {@link
Driver#run(String)} reports compile
+ * and execution failures (a partition spec that names a non-partition
column, a DDL task the
+ * metastore refuses) through a non-zero response code rather than by
throwing, so a caller that
+ * only catches exceptions goes on as though the statement had been applied.
+ */
+ public static void executeOrThrow(Driver driver, String sql) throws
CommandNeedRetryException {
+ CommandProcessorResponse response = driver.run(sql);
+ if (response.getResponseCode() != 0) {
+ throw new HoodieHiveSyncException(String.format(
+ "Hive rejected the statement with response code %d (SQLState %s):
%s%nStatement: %s",
+ response.getResponseCode(), response.getSQLState(),
response.getErrorMessage(), abbreviate(sql)),
+ response.getException());
+ }
+ }
+
+ private static String abbreviate(String sql) {
+ if (sql.length() <= MAX_STATEMENT_LENGTH_IN_ERROR) {
+ return sql;
+ }
+ return sql.substring(0, MAX_STATEMENT_LENGTH_IN_ERROR) + "... (" +
sql.length() + " characters)";
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index 15029094ef75..0bb527d0b013 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -41,6 +41,7 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.ExceptionUtil;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.hadoop.HoodieParquetInputFormat;
import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
@@ -403,6 +404,34 @@ public class TestHiveSyncTool {
"Incremental add via parallel HiveQL batching should sync the new
partitions");
}
+ /**
+ * Hive's Driver reports a rejected statement through its response code
rather than by throwing.
+ * A table registered with a different partition column than the one the
sync is configured with
+ * rejects every ADD PARTITION, and the sync must fail instead of recording
the commit as synced.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testHiveQLSyncFailsWhenHiveRejectsAStatement(boolean batchingEnabled)
throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(),
HiveSyncMode.HIVEQL.name());
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_ENABLED.key(),
String.valueOf(batchingEnabled));
+ hiveSyncProps.setProperty(META_SYNC_PARTITION_FIELDS.key(),
HoodieRecord.PARTITION_PATH_METADATA_FIELD);
+ HiveTestUtil.createCOWTable("100", 1, true);
+ reInitHiveSyncClient();
+ reSyncHiveTable();
+ assertEquals("100",
hiveClient.getLastCommitTimeSynced(HiveTestUtil.TABLE_NAME).get());
+
+ hiveSyncProps.setProperty(META_SYNC_PARTITION_FIELDS.key(), "datestr");
+ HiveTestUtil.addCOWPartition("2050/01/01", true, true, "101");
+ reInitHiveSyncClient();
+
+ HoodieException ex = assertThrows(HoodieException.class,
this::reSyncHiveTable);
+ assertTrue(ExceptionUtil.validateErrorMsg(ex, "Hive rejected the
statement"),
+ "The sync must fail with Hive's rejection of the ADD PARTITION");
+ assertTrue(ExceptionUtil.validateErrorMsg(ex, "datestr"));
+ assertEquals("100",
hiveClient.getLastCommitTimeSynced(HiveTestUtil.TABLE_NAME).get(),
+ "A commit whose partitions Hive rejected must not be recorded as
synced");
+ }
+
/**
* Exercises the DROP path in HiveQL mode with batching on. DROP routes
through
* IMetaStoreClient.dropPartition (Thrift, not Hive Driver), so when
batching is
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestHiveQueryDDLExecutorSession.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestHiveQueryDDLExecutorSession.java
index 74687d081ea0..2c5947e10014 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestHiveQueryDDLExecutorSession.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestHiveQueryDDLExecutorSession.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.ql.Driver;
import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -158,7 +159,7 @@ class TestHiveQueryDDLExecutorSession {
when(driver.run(anyString())).thenAnswer(invocation -> {
sessionsSeenByDriver.add(SessionState.get());
loadersSeenByDriver.add(Thread.currentThread().getContextClassLoader());
- return null;
+ return new CommandProcessorResponse(0);
});
HiveQueryDDLExecutor executor = executorWith(driver, sessionState);
SessionState.detachSession();
@@ -186,6 +187,7 @@ class TestHiveQueryDDLExecutorSession {
when(otherSession.getConf()).thenReturn(new HiveConf());
SessionState sessionState = mock(SessionState.class);
Driver driver = mock(Driver.class);
+ when(driver.run(anyString())).thenReturn(new CommandProcessorResponse(0));
HiveQueryDDLExecutor executor = executorWith(driver, sessionState);
SessionState.setCurrentSessionState(otherSession);
ClassLoader callerLoader = Thread.currentThread().getContextClassLoader();
@@ -198,6 +200,37 @@ class TestHiveQueryDDLExecutorSession {
verify(driver, times(1)).run("SHOW TABLES");
}
+ /**
+ * Driver.run() reports a statement Hive rejects through its response rather
than by throwing.
+ * The sync has to stop there: the statements after it assume it was
applied, and a sync that
+ * carries on records the commit as synced to a table that never received it.
+ */
+ @Test
+ void statementHiveRejectsFailsTheSyncAndHandsTheThreadBack() throws
Exception {
+ SessionState otherSession = mock(SessionState.class);
+ when(otherSession.getConf()).thenReturn(new HiveConf());
+ SessionState sessionState = mock(SessionState.class);
+ Driver driver = mock(Driver.class);
+ String rejected = "ALTER TABLE `tbl` ADD IF NOT EXISTS PARTITION
(`datestr`='2026-09-13')";
+ when(driver.run(rejected)).thenReturn(new CommandProcessorResponse(10,
+ "FAILED: ValidationFailureSemanticException Partition spec
{datestr=2026-09-13} contains non-partition columns",
+ "42000"));
+ HiveQueryDDLExecutor executor = executorWith(driver, sessionState);
+ SessionState.setCurrentSessionState(otherSession);
+ ClassLoader callerLoader = Thread.currentThread().getContextClassLoader();
+
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> executor.runSQLs(Arrays.asList(rejected, "ALTER TABLE `tbl` SET
TBLPROPERTIES ('k'='v')")));
+
+ String message = ex.getCause().getMessage();
+ assertTrue(message.contains("contains non-partition columns"), message);
+ assertTrue(message.contains(rejected), "The error must name the statement;
was " + message);
+ verify(driver, times(1)).run(anyString());
+ assertSame(otherSession, SessionState.get(), "The session held before the
statements must be put back");
+ assertSame(callerLoader, Thread.currentThread().getContextClassLoader(),
+ "The class loader held before the statements must be put back");
+ }
+
/**
* Driver.close() clears the current session's lineage state and
Driver.destroy() takes its
* transaction manager to release locks, so both have to see the session
that owns the Driver.
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
index efb99c5d5bbe..6244f7005400 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
@@ -24,6 +24,7 @@ import org.apache.hudi.hive.HoodieHiveSyncException;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.Driver;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.junit.jupiter.api.Test;
import org.mockito.invocation.InvocationOnMock;
@@ -56,6 +57,8 @@ import static org.mockito.Mockito.when;
*/
class TestHiveDriverPool {
+ private static final CommandProcessorResponse SUCCESS = new
CommandProcessorResponse(0);
+
private static HiveSyncConfig configWithEmptyHiveConf() {
HiveSyncConfig config = mock(HiveSyncConfig.class);
doAnswer(inv -> new HiveConf()).when(config).getHiveConf();
@@ -104,7 +107,7 @@ class TestHiveDriverPool {
seenThreadsByDriver.put(d, ConcurrentHashMap.newKeySet());
doAnswer((InvocationOnMock inv) -> {
seenThreadsByDriver.get(d).add(Thread.currentThread().getName());
- return null;
+ return SUCCESS;
}).when(d).run(anyString());
return d;
};
@@ -132,7 +135,7 @@ class TestHiveDriverPool {
if (sql.equals("FAIL")) {
throw new RuntimeException("boom: " + sql);
}
- return null;
+ return SUCCESS;
}).when(d).run(anyString());
return d;
};
@@ -145,6 +148,46 @@ class TestHiveDriverPool {
}
}
+ /**
+ * Driver.run() reports a statement Hive rejects through its response rather
than by throwing,
+ * so a batch that only watched for exceptions would report the sync as
applied.
+ */
+ @Test
+ void awaitAllThrowsWhenHiveRejectsAStatement() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ doAnswer(inv -> "REJECTED".equals(inv.getArgument(0))
+ ? new CommandProcessorResponse(10, "FAILED: SemanticException table
is not partitioned", "42000")
+ : SUCCESS).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
+ ParallelDispatch dispatch = pool.dispatchAll(Arrays.asList("OK",
"REJECTED", "OK"));
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> pool.awaitAll(dispatch));
+ String message = ex.getCause().getMessage();
+ assertTrue(message.contains("SemanticException table is not
partitioned"), message);
+ assertTrue(message.contains("REJECTED"), "The error must name the
statement; was " + message);
+ }
+ }
+
+ @Test
+ void runOnEachWorkerThrowsWhenHiveRejectsASetupStatement() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ when(d.run(anyString())).thenReturn(
+ new CommandProcessorResponse(10, "FAILED: SemanticException Database
does not exist: db1", "42000"));
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> pool.runOnEachWorker(Collections.singletonList("USE `db1`")));
+ assertTrue(ex.getCause().getMessage().contains("Database does not exist:
db1"));
+ }
+ }
+
@Test
void concurrentDispatchBoundedByPoolSize() throws Exception {
HiveSyncConfig config = configWithEmptyHiveConf();
@@ -158,7 +201,7 @@ class TestHiveDriverPool {
maxInFlight.updateAndGet(prev -> Math.max(prev, now));
hold.await(2, TimeUnit.SECONDS);
inFlight.decrementAndGet();
- return null;
+ return SUCCESS;
}).when(d).run(anyString());
return d;
};
@@ -258,7 +301,7 @@ class TestHiveDriverPool {
sqlsByDriver.put(d, java.util.Collections.synchronizedList(new
java.util.ArrayList<>()));
doAnswer((InvocationOnMock inv) -> {
sqlsByDriver.get(d).add(inv.getArgument(0));
- return null;
+ return SUCCESS;
}).when(d).run(anyString());
return d;
};
@@ -296,7 +339,7 @@ class TestHiveDriverPool {
if (sql.equals("FAIL")) {
throw new RuntimeException("boom");
}
- return null;
+ return SUCCESS;
}).when(d).run(anyString());
return d;
};
@@ -346,7 +389,7 @@ class TestHiveDriverPool {
// parked on future 0 at the moment worker 1 would pick up
AFTER_FAIL.
releaseSlow.await(5, TimeUnit.SECONDS);
}
- return null;
+ return SUCCESS;
}).when(d).run(anyString());
return d;
};
