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 483dd878b4ef feat(hive-sync): batch and parallelize HiveQL partition
operations (#18984)
483dd878b4ef is described below
commit 483dd878b4ef1b3df1c6135b5daae3fd0274c892
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Mon Jul 27 01:35:41 2026 -0700
feat(hive-sync): batch and parallelize HiveQL partition operations (#18984)
* feat(hive-sync): batch and parallelize HiveQL partition operations
Adds an opt-in pool that splits HiveQL partition DDL (add/update/touch/drop)
into batches of `hoodie.datasource.hive_sync.batch_num` and dispatches them
in parallel across a pool of single-thread workers. Each worker owns its
own Hive `Driver` + `SessionState` (both thread-bound in Hive 2.x), so the
fan-out is implemented as a fixed pool of dedicated single-thread executors
rather than a shared thread pool.
Table-level operations (create/alter table, last commit time, writer
version)
continue to use the single session `Driver`. Partition-phase SQL lists run
through the pool only when `hoodie.datasource.hive_sync.batching.enabled`
is set to true; default off, existing behavior unchanged.
Hive 2.x's `ALTER PARTITION SET LOCATION` ignores db.table qualifiers and
uses the connection's current database, so each worker is primed with the
correct USE statement before any partition ALTER is dispatched.
* fix(hive-sync): address review feedback on HiveQL partition batching
- Gate TOUCH batch-splitting on hoodie.datasource.hive_sync.batching.enabled
so the default (off) HiveQL path emits a single ALTER TABLE ... TOUCH
statement as before, instead of always splitting into batch_num chunks.
- Give each HiveDriverPool worker its own exclusively-owned SessionState
instead of sharing one SessionState object across all workers. Concurrent
Driver.run() calls against a shared session risked corrupting session-
scoped state (current db, scratch dirs, txn/lock manager), and closing
the shared session once per worker on teardown closed it multiple times
while racing worker Driver.close() calls. Workers now bootstrap one at a
time (each SessionState construction no longer races another).
- Close the driver pool in HiveQueryDDLExecutor's constructor failure path,
so a failed SessionState/Driver bootstrap doesn't leak the pool's worker
threads, Drivers, and sessions (the pool is constructed by the caller
before this constructor runs, and no one else can close it if we throw).
- Fix testHiveQLTouchPartitionsWithBatching to drive touchPartitionsToTable
directly; the previous resync-based version never reached the batched
TOUCH path because incremental sync short-circuits with no new commit.
- Make TestHiveDriverPool's cancel-on-first-error assertion race-free by
parking pending tasks on a latch and asserting isCancelled() || !isDone().
* fix(hive-sync): isolate per-worker HiveConf; document ADD parallel
dispatch
- HiveDriverPool.DefaultDriverFactory now builds a per-worker HiveConf copy
(new HiveConf(hiveConf)) instead of sharing one HiveConf instance across
all workers, and passes that copy to both SessionState and Driver. Hive's
QueryState/Driver mutate per-query keys (e.g. HIVEQUERYID) on the conf
during run(), so a shared HiveConf let concurrent Driver.run() calls
overwrite each other's query-scoped configuration even though each
worker already had its own SessionState object.
- Reworded HIVE_SYNC_BATCHING_ENABLED's doc to explicitly include ADD in
the parallel dispatch scope. addPartitionsToTable routes through
runSQLs, so ADD batches are dispatched across the pool same as TOUCH/
SET_LOCATION when the flag is on; only the batch size (not the fan-out)
was already unchanged before this flag existed.
* fix(hive-sync): abort dispatch on first error; scope TOUCH batching to
parallel paths
Two review fixes on the HiveQL partition batching path.
1. awaitAll did not actually cancel pending work on first error.
The futures returned by dispatchAll belong to N independent single-thread
executors, each draining its own queue. awaitAll blocked on Future.get() in
submission order, so a failure on a fast worker went unobserved while the
awaiting thread was parked on a slow worker's earlier future -- and the
failed
worker kept pulling and applying more partition DDL from its own queue. The
advertised "cancel pending futures on first error" behavior did not hold.
dispatchAll now returns a Dispatch handle carrying a shared abort flag. Each
task checks the flag on entry and bails with CancellationException without
touching its Driver; the first task to fail sets it. awaitAll blocks on a
latch
that trips on either all-settled or first-abort, then sweeps cancel(false)
before walking the futures. Cancelling from the awaiting thread is
inherently
late here, so the in-task check is what bounds how much extra DDL a failed
sync
can apply. mayInterruptIfRunning=false is preserved, so in-flight statements
still run to completion rather than leaving a Driver mid-statement.
2. hoodie.datasource.hive_sync.batching.enabled leaked into JDBC mode.
QueryBasedDDLExecutor is also the base class for JDBCExecutor, which does
not
override runSQLs. With the flag on in JDBC mode, TOUCH was split into
batch_num
statements that then executed serially -- changing statement count and
partial-application semantics for no benefit, and contradicting the
documented
"JDBC is unaffected" contract.
The config read in constructPartitionAlterStatements is replaced by a
getTouchBatchSize(int) hook. The base implementation returns the full
partition
count (one statement, the long-standing behavior); only HiveQueryDDLExecutor
overrides it, and only when a driver pool is actually present. Keying on
pool
presence rather than on the config means the split can never take effect on
a
path that would just execute the batches serially.
Tests:
- awaitAllStopsLaterWorkerWhenEarlierFutureIsSlow: pins the interleaving
the bug
needs (slow statement on worker 0, fast failure on worker 1) and asserts
the
statement queued behind the failure never reaches a Driver.
- awaitAllCancelsPendingFuturesOnFirstError: rewritten. The prior version
asserted isCancelled() || !isDone() to hedge around the race; the abort
flag
makes the outcome deterministic, so it now asserts on what actually
executed.
- New TestQueryBasedDDLExecutorTouchBatching: asserts a serial executor
emits a
single TOUCH statement with the flag on, that its SQL is byte-identical
with
the flag on and off, and that a parallel-dispatch executor still splits.
Both new pool tests were verified to fail on all surefire attempts against
the
prior logic and pass against the fix. Full hudi-hive-sync suite: 303/303.
* fix(hive-sync): close driver pool if sync client construction fails
HiveQueryDDLExecutor's own catch closes the pool, but it only covers throws
from
inside its try block. QueryBasedDDLExecutor's super(config) runs the
PartitionValueExtractor reflection first, so a bad
hoodie.datasource.hive_sync.partition_extractor_class throws before that
try is
entered -- the executor's catch never runs, HoodieHiveSyncClient's catch
just
rethrows, and the already-constructed pool leaks its worker threads and
Drivers.
Close the pool in HoodieHiveSyncClient's constructor catch, which covers
every
window between building the pool and handing ownership to the executor.
close()
is idempotent, so overlapping with the executor's own cleanup is harmless.
---
.../org/apache/hudi/hive/HiveSyncConfigHolder.java | 22 +
.../org/apache/hudi/hive/HoodieHiveSyncClient.java | 38 +-
.../apache/hudi/hive/ddl/HiveQueryDDLExecutor.java | 91 +++-
.../hudi/hive/ddl/QueryBasedDDLExecutor.java | 70 +++-
.../org/apache/hudi/hive/util/HiveDriverPool.java | 463 +++++++++++++++++++++
.../org/apache/hudi/hive/TestHiveSyncTool.java | 107 +++++
.../TestQueryBasedDDLExecutorTouchBatching.java | 192 +++++++++
.../apache/hudi/hive/util/TestHiveDriverPool.java | 316 ++++++++++++++
8 files changed, 1281 insertions(+), 18 deletions(-)
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
index 418676d59162..14afa81ea30f 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
@@ -122,6 +122,28 @@ public class HiveSyncConfigHolder {
.defaultValue(1000)
.markAdvanced()
.withDocumentation("The number of partitions one batch when synchronous
partitions to hive.");
+ public static final ConfigProperty<Boolean> HIVE_SYNC_BATCHING_ENABLED =
ConfigProperty
+ .key("hoodie.datasource.hive_sync.batching.enabled")
+ .defaultValue(false)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Only applies to HiveQL sync mode; has no effect in
HMS or JDBC mode. When true, "
+ + "ADD, TOUCH, and SET_LOCATION partition statements are dispatched
in parallel across a pool of "
+ + "Hive Driver workers, with ADD and TOUCH additionally split into
batches of "
+ + "`hoodie.datasource.hive_sync.batch_num` partitions per statement
(ADD was already batched "
+ + "before this flag existed; only its dispatch becomes parallel
here). SET_LOCATION remains one "
+ + "statement per partition, as Hive SQL has no multi-partition form.
DROP remains serial. "
+ + "Table-level statements (create/alter table, last commit time,
writer version) continue to run "
+ + "on the single session Driver. Default off; the default HiveQL
path is unchanged unless "
+ + "explicitly opted in.");
+ public static final ConfigProperty<Integer> HIVE_SYNC_BATCHING_THREADS =
ConfigProperty
+ .key("hoodie.datasource.hive_sync.batching.threads")
+ .defaultValue(4)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Pool size (number of Hive Driver workers) and
worker-thread count for parallel "
+ + "HiveQL partition dispatch when
`hoodie.datasource.hive_sync.batching.enabled` is true. "
+ + "Ignored otherwise.");
public static final ConfigProperty<String> HIVE_SYNC_MODE = ConfigProperty
.key("hoodie.datasource.hive_sync.mode")
.noDefaultValue()
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
index 9c5033c80bd4..accc60cb9066 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
@@ -37,6 +37,7 @@ import org.apache.hudi.hive.ddl.HiveQueryDDLExecutor;
import org.apache.hudi.hive.ddl.HiveSyncMode;
import org.apache.hudi.hive.ddl.JDBCBasedMetadataOperator;
import org.apache.hudi.hive.ddl.JDBCExecutor;
+import org.apache.hudi.hive.util.HiveDriverPool;
import org.apache.hudi.hive.util.IMetaStoreClientUtil;
import org.apache.hudi.hive.util.PartitionFilterGenerator;
import org.apache.hudi.sync.common.HoodieSyncClient;
@@ -66,6 +67,8 @@ import static
org.apache.hudi.hadoop.utils.HoodieHiveUtils.GLOBALLY_CONSISTENT_R
import static
org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.getInputFormatClassName;
import static
org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.getOutputFormatClassName;
import static
org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.getSerDeClassName;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_BATCHING_ENABLED;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_BATCHING_THREADS;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_USE_SPARK_CATALOG;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_USE_JDBC;
@@ -86,6 +89,10 @@ public class HoodieHiveSyncClient extends HoodieSyncClient {
private final Map<String, Table> initialTableByName = new HashMap<>();
DDLExecutor ddlExecutor;
private IMetaStoreClient client;
+ // Present only when HIVE_SYNC_BATCHING_ENABLED and sync mode is HIVEQL
(explicit
+ // or legacy default). Owned by HiveQueryDDLExecutor; this field is kept for
+ // reference only — close() is delegated through ddlExecutor.close().
+ private Option<HiveDriverPool> partitionDriverPool = Option.empty();
/**
* JDBC-based metadata operator, lazily initialized on first Thrift
@@ -124,7 +131,8 @@ public class HoodieHiveSyncClient extends HoodieSyncClient {
ddlExecutor = new HMSDDLExecutor(config, this.client);
break;
case HIVEQL:
- ddlExecutor = new HiveQueryDDLExecutor(config, this.client);
+ this.partitionDriverPool = maybeBuildHiveDriverPool(config);
+ ddlExecutor = new HiveQueryDDLExecutor(config, this.client,
this.partitionDriverPool);
break;
case JDBC:
JDBCExecutor jdbcExecutor = new JDBCExecutor(config);
@@ -142,14 +150,32 @@ public class HoodieHiveSyncClient extends
HoodieSyncClient {
jdbcMetadataOperator = new JDBCBasedMetadataOperator(
jdbcExecutor.getConnection(), databaseName);
} else {
- ddlExecutor = new HiveQueryDDLExecutor(config, this.client);
+ this.partitionDriverPool = maybeBuildHiveDriverPool(config);
+ ddlExecutor = new HiveQueryDDLExecutor(config, this.client,
this.partitionDriverPool);
}
}
} catch (Exception e) {
+ // The pool owns live daemon threads and Hive Drivers, and is built
before the
+ // executor that would otherwise own its lifecycle. Any throw between
those two
+ // points would leak it -- notably QueryBasedDDLExecutor's
super(config), which
+ // runs the PartitionValueExtractor reflection before
HiveQueryDDLExecutor's own
+ // try block is even entered. Closing here covers every such window; the
pool's
+ // close() is idempotent, so overlapping with the executor's cleanup is
harmless.
+ closePartitionDriverPoolQuietly();
throw new HoodieHiveSyncException("Failed to create
HiveMetaStoreClient", e);
}
}
+ private void closePartitionDriverPoolQuietly() {
+ partitionDriverPool.ifPresent(pool -> {
+ try {
+ pool.close();
+ } catch (Exception e) {
+ log.warn("Error closing HiveDriverPool during failed sync client
construction", e);
+ }
+ });
+ }
+
/**
* Returns true if Thrift API was detected as incompatible and JDBC
* fallback is available. When true, metadata operations should use
@@ -201,6 +227,14 @@ public class HoodieHiveSyncClient extends HoodieSyncClient
{
}
}
+ private Option<HiveDriverPool> maybeBuildHiveDriverPool(HiveSyncConfig
config) {
+ if (!config.getBooleanOrDefault(HIVE_SYNC_BATCHING_ENABLED)) {
+ return Option.empty();
+ }
+ int size = config.getIntOrDefault(HIVE_SYNC_BATCHING_THREADS);
+ return Option.of(new HiveDriverPool(config, size));
+ }
+
private Table getInitialTable(String table) {
return initialTableByName.computeIfAbsent(table, t -> {
try {
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 25434d29eb3f..c853313182ed 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
@@ -19,8 +19,10 @@
package org.apache.hudi.hive.ddl;
import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.hive.HiveSyncConfig;
import org.apache.hudi.hive.HoodieHiveSyncException;
+import org.apache.hudi.hive.util.HiveDriverPool;
import org.apache.hudi.hive.util.HivePartitionUtil;
import lombok.extern.slf4j.Slf4j;
@@ -41,6 +43,7 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_BATCH_SYNC_PARTITION_NUM;
import static org.apache.hudi.sync.common.util.TableUtils.tableId;
/**
@@ -52,10 +55,20 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
private final IMetaStoreClient metaStoreClient;
private SessionState sessionState;
private Driver hiveDriver;
+ // When present, partition-phase SQL lists fan out across this pool;
table-level SQL
+ // (createTable, schema evolution, single-statement runSQL callers) always
uses the
+ // session `hiveDriver` above. See HiveDriverPool javadoc.
+ private final Option<HiveDriverPool> driverPool;
public HiveQueryDDLExecutor(HiveSyncConfig config, IMetaStoreClient
metaStoreClient) {
+ this(config, metaStoreClient, Option.empty());
+ }
+
+ public HiveQueryDDLExecutor(HiveSyncConfig config, IMetaStoreClient
metaStoreClient,
+ Option<HiveDriverPool> driverPool) {
super(config);
this.metaStoreClient = metaStoreClient;
+ this.driverPool = driverPool;
try {
this.sessionState = new SessionState(config.getHiveConf(),
UserGroupInformation.getCurrentUser().getShortUserName());
@@ -73,6 +86,15 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
if (this.hiveDriver != null) {
this.hiveDriver.close();
}
+ // driverPool (if present) was already constructed by the caller before
this
+ // ctor ran; since we're about to throw, no one else will call close()
on it.
+ driverPool.ifPresent(pool -> {
+ try {
+ pool.close();
+ } catch (Exception poolCloseException) {
+ log.error("Error while closing HiveDriverPool", poolCloseException);
+ }
+ });
throw new HoodieHiveSyncException("Failed to create HiveQueryDDL
object", e);
}
}
@@ -82,19 +104,74 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
updateHiveSQLs(Collections.singletonList(sql));
}
+ /**
+ * Partition-phase SQL fan-out. When the driver pool is present, any leading
+ * {@code USE database} statements are run on every worker (Hive 2.x's
+ * ALTER PARTITION SET LOCATION ignores db.table qualifiers and uses the
+ * connection's current database, so each worker needs to USE the right db
+ * before any partition ALTER). The remaining statements are then dispatched
+ * round-robin across the pool. Falls through to the sequential path on the
+ * session Driver when no pool is configured.
+ */
+ @Override
+ protected void runSQLs(List<String> sqls) {
+ if (sqls.isEmpty()) {
+ return;
+ }
+ if (!driverPool.isPresent()) {
+ updateHiveSQLs(sqls);
+ return;
+ }
+ HiveDriverPool pool = driverPool.get();
+ int useStatementCount = 0;
+ while (useStatementCount < sqls.size() &&
isUseStatement(sqls.get(useStatementCount))) {
+ useStatementCount++;
+ }
+ if (useStatementCount > 0) {
+ List<String> setupStatements = sqls.subList(0, useStatementCount);
+ pool.runOnEachWorker(setupStatements);
+ }
+ List<String> partitionStatements = sqls.subList(useStatementCount,
sqls.size());
+ if (partitionStatements.isEmpty()) {
+ return;
+ }
+ pool.awaitAll(pool.dispatchAll(partitionStatements));
+ }
+
+ /**
+ * Splits TOUCH into batches of {@code HIVE_BATCH_SYNC_PARTITION_NUM} only
when a
+ * driver pool is actually present — i.e. only when {@link #runSQLs(List)}
will
+ * dispatch those batches in parallel. Keyed on pool presence rather than on
the
+ * {@code batching.enabled} config so the split can never take effect on a
path
+ * that would just execute the batches serially (the base class, and
therefore
+ * {@code JDBCExecutor}, always emits one statement).
+ */
+ @Override
+ protected int getTouchBatchSize(int partitionCount) {
+ return driverPool.isPresent()
+ ? config.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM) :
partitionCount;
+ }
+
+ // Strict 4-char prefix match on "USE ". Internal callers
(constructPartitionAlterStatements)
+ // always emit the USE statement without leading whitespace; do not call
with externally
+ // supplied SQL that might be padded.
+ private static boolean isUseStatement(String sql) {
+ return sql != null && sql.regionMatches(true, 0, "USE ", 0, 4);
+ }
+
private List<CommandProcessorResponse> updateHiveSQLs(List<String> sqls) {
List<CommandProcessorResponse> responses = new ArrayList<>();
+ HoodieTimer timer = HoodieTimer.start();
try {
for (String sql : sqls) {
if (hiveDriver != null) {
- HoodieTimer timer = HoodieTimer.start();
responses.add(hiveDriver.run(sql));
- log.info("Time taken to execute [{}]: {} ms", sql, timer.endTimer());
}
}
} catch (Exception e) {
throw new HoodieHiveSyncException("Failed in executing SQL", e);
}
+ log.info("Executed {} SQL statements sequentially in {} ms", sqls.size(),
timer.endTimer());
return responses;
}
@@ -149,6 +226,16 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
@Override
public void close() {
+ // Close the pool first so the worker threads stop dispatching against
their
+ // Drivers before we tear down anything else. The pool's close() runs
+ // Driver/SessionState cleanup on each worker's own thread.
+ driverPool.ifPresent(pool -> {
+ try {
+ pool.close();
+ } catch (Exception e) {
+ log.warn("Error closing HiveDriverPool", e);
+ }
+ });
if (metaStoreClient != null) {
Hive.closeCurrent();
}
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java
index 78b8e684e49a..3bcbbe1841df 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java
@@ -20,6 +20,7 @@ package org.apache.hudi.hive.ddl;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.PartitionPathEncodeUtils;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.Pair;
@@ -75,6 +76,35 @@ public abstract class QueryBasedDDLExecutor implements
DDLExecutor {
*/
public abstract void runSQL(String sql);
+ /**
+ * Runs a list of SQL statements. The default implementation executes them
+ * sequentially via {@link #runSQL(String)}. Subclasses that can parallelize
+ * (e.g. {@link HiveQueryDDLExecutor} with a driver pool) override this hook
+ * to fan the list out across workers. The contract requires that the list
+ * has no positional dependencies — callers must fully qualify table names
+ * with {@code `db`.`tbl`} so any statement can run on any worker.
+ */
+ protected void runSQLs(List<String> sqls) {
+ for (String sql : sqls) {
+ runSQL(sql);
+ }
+ }
+
+ /**
+ * Number of partitions to pack into a single {@code ALTER TABLE ... TOUCH}
statement.
+ *
+ * <p>The base implementation returns {@code partitionCount}, i.e. one
statement
+ * covering every partition — the long-standing behavior, and the only
correct choice
+ * when {@link #runSQLs(List)} executes the list serially. Splitting a TOUCH
into
+ * several statements changes failure semantics (a mid-list failure leaves
some
+ * partitions touched and some not), so it is only worth doing when the
resulting
+ * statements are actually dispatched in parallel. Subclasses that
parallelize
+ * override this; see {@link HiveQueryDDLExecutor}.
+ */
+ protected int getTouchBatchSize(int partitionCount) {
+ return partitionCount;
+ }
+
@Override
public void createDatabase(String databaseName) {
runSQL("create database if not exists " + databaseName);
@@ -120,7 +150,7 @@ public abstract class QueryBasedDDLExecutor implements
DDLExecutor {
}
log.info("Adding partitions {} to table {}", partitionsToAdd.size(),
tableName);
List<String> sqls = constructAddPartitions(tableName, partitionsToAdd);
- sqls.stream().forEach(sql -> runSQL(sql));
+ runSQLs(sqls);
}
@Override
@@ -131,9 +161,7 @@ public abstract class QueryBasedDDLExecutor implements
DDLExecutor {
}
log.info("Changing partitions {} on {}", changedPartitions.size(),
tableName);
List<String> sqls = constructPartitionAlterStatements(tableName,
changedPartitions, PartitionAlterType.SET_LOCATION);
- for (String sql : sqls) {
- runSQL(sql);
- }
+ runSQLs(sqls);
}
@Override
@@ -216,29 +244,43 @@ public abstract class QueryBasedDDLExecutor implements
DDLExecutor {
}
log.info("Touching partitions {} on {}", touchPartitions.size(),
tableName);
List<String> sqls = constructPartitionAlterStatements(tableName,
touchPartitions, PartitionAlterType.TOUCH);
- for (String sql : sqls) {
- runSQL(sql);
- }
+ runSQLs(sqls);
}
/**
* Builds SQL statements to either touch partitions or set their location.
- * TOUCH: one ALTER TABLE ... TOUCH PARTITION (p1) PARTITION (p2) ...
- * SET_LOCATION: one ALTER TABLE ... PARTITION (p) SET LOCATION '...' per
partition.
+ *
+ * <p>The first element of the returned list is always a {@code USE database}
+ * statement. Hive 2.x's ALTER PARTITION ... SET LOCATION does not respect
the
+ * {@code db.table} qualifier (silently routes to the connection's current
+ * database), so the {@code USE} is load-bearing. Parallel execution paths
must
+ * run this statement on every worker before fanning out the rest.
+ *
+ * <p>TOUCH: one {@code ALTER TABLE ... TOUCH PARTITION (p1) ...} statement
per
+ * batch of {@link #getTouchBatchSize(int)} partitions. The base
implementation
+ * returns the full partition count, i.e. a single statement covering
everything,
+ * matching pre-batching behavior. Only subclasses that actually dispatch the
+ * resulting statements in parallel override it to split.
+ *
+ * <p>SET_LOCATION: one {@code ALTER TABLE ... PARTITION (p) SET LOCATION
'...'}
+ * per partition (Hive SQL does not support multi-partition SET LOCATION in
one
+ * statement).
*/
private List<String> constructPartitionAlterStatements(String tableName,
List<String> partitions, PartitionAlterType alterType) {
List<String> result = new ArrayList<>();
- // Hive 2.x doesn't like db.table name for operations, hence we need to
change to using the database first
String useDatabase = "USE " + HIVE_ESCAPE_CHARACTER + databaseName +
HIVE_ESCAPE_CHARACTER;
result.add(useDatabase);
String alterTablePrefix = "ALTER TABLE " + HIVE_ESCAPE_CHARACTER +
tableName + HIVE_ESCAPE_CHARACTER;
+ int batchSyncPartitionNum = getTouchBatchSize(partitions.size());
switch (alterType) {
case TOUCH:
- String alterTable = alterTablePrefix + " TOUCH";
- for (String partition : partitions) {
- alterTable += " PARTITION (" + getPartitionClause(partition) + ")";
+ for (List<String> batch : CollectionUtils.batches(partitions,
batchSyncPartitionNum)) {
+ StringBuilder alterTable = new
StringBuilder(alterTablePrefix).append(" TOUCH");
+ for (String partition : batch) {
+ alterTable.append(" PARTITION
(").append(getPartitionClause(partition)).append(")");
+ }
+ result.add(alterTable.toString());
}
- result.add(alterTable);
break;
case SET_LOCATION:
for (String partition : partitions) {
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
new file mode 100644
index 000000000000..5950bf442313
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
@@ -0,0 +1,463 @@
+/*
+ * 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.common.util.VisibleForTesting;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hive.HiveSyncConfig;
+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.session.SessionState;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
+
+/**
+ * Pool of Hive {@link Driver} + {@link SessionState} pairs for parallel
HiveQL DDL.
+ *
+ * <p>Hive's {@code SessionState.start(state)} binds state to the calling
thread's
+ * thread-local, and {@code Driver} reads from that thread-local during {@code
run()}.
+ * A Driver constructed on one thread cannot be safely used from another. This
pool
+ * solves that by giving each slot its own dedicated worker thread (a
single-thread
+ * executor) — the Driver and SessionState are built on that thread by a
bootstrap
+ * task, and all subsequent SQL for that slot runs on the same thread.
+ *
+ * <p><b>Usage contract:</b> use this pool only for partition-row DDL
statements that
+ * are independent of each other and freely shuffleable across workers.
Table-level
+ * statements (createTable, schema evolution, USE database) must continue to
run on
+ * the session {@code Driver} held by {@code HiveQueryDDLExecutor} on the sync
driver
+ * thread. The pool is gated behind {@code
hoodie.datasource.hive_sync.batching.enabled}
+ * and is constructed only for HiveQL sync mode.
+ */
+public class HiveDriverPool implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(HiveDriverPool.class);
+
+ // Per-worker Driver construction has to be fast in practice (a few hundred
ms
+ // for the SessionState + Driver init). A 60s ceiling per worker leaves
plenty of
+ // headroom for a slow JVM warm-up but bounds the failure mode if the
metastore
+ // is unreachable or Hive hangs during init.
+ private static final long BOOTSTRAP_TIMEOUT_SECONDS = 60;
+
+ private final List<Worker> workers;
+ private final int size;
+ private volatile boolean closed;
+
+ public HiveDriverPool(HiveSyncConfig config, int size) {
+ this(config, size, new DefaultDriverFactory(config));
+ }
+
+ // Package-private for tests: accepts a DriverFactory so unit tests can
inject
+ // mock Driver instances without standing up a real Hive instance.
+ HiveDriverPool(HiveSyncConfig config, int size, DriverFactory factory) {
+ if (size < 1) {
+ throw new IllegalArgumentException("Pool size must be >= 1, got " +
size);
+ }
+ this.size = size;
+ this.workers = new ArrayList<>(size);
+ String databaseName = config.getStringOrDefault(META_SYNC_DATABASE_NAME);
+ PoolThreadFactory threadFactory = new PoolThreadFactory();
+ try {
+ // Bootstrap workers one at a time (not concurrently): each worker
builds its
+ // own exclusively-owned SessionState, and constructing several
SessionStates
+ // in parallel risks racing on shared scratch-dir creation. This only
affects
+ // one-time pool startup cost, not per-statement dispatch latency.
+ for (int i = 0; i < size; i++) {
+ Worker worker = new Worker(threadFactory);
+ workers.add(worker);
+ worker.executor.submit(() -> {
+ worker.driver = factory.newDriver(databaseName);
+ worker.sessionState = SessionState.get();
+ return null;
+ }).get(BOOTSTRAP_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
+ } catch (Exception e) {
+ tearDown();
+ throw new HoodieException("Failed to construct HiveDriverPool of size "
+ size, e);
+ }
+ LOG.info("Initialized HiveDriverPool with {} workers", size);
+ }
+
+ /**
+ * Runs each given SQL on <i>every</i> worker, in order. Used for setup
statements
+ * (e.g. {@code USE database}) that must establish per-thread session context
+ * before any partition statement runs. Blocks until all workers have
completed
+ * the setup. Throws on first error.
+ */
+ public void runOnEachWorker(List<String> setupSqls) {
+ if (closed) {
+ throw new IllegalStateException("Cannot dispatch to a closed
HiveDriverPool");
+ }
+ if (setupSqls.isEmpty()) {
+ return;
+ }
+ Dispatch dispatch = new Dispatch(workers.size());
+ for (Worker worker : workers) {
+ dispatch.add(worker.executor.submit(() -> {
+ if (dispatch.aborted()) {
+ throw new CancellationException("Skipped after an earlier setup
statement failed");
+ }
+ try {
+ for (String sql : setupSqls) {
+ worker.driver.run(sql);
+ }
+ } catch (Throwable t) {
+ dispatch.abort();
+ throw t;
+ } finally {
+ dispatch.taskSettled();
+ }
+ return null;
+ }));
+ }
+ dispatch.sealed();
+ awaitAll(dispatch);
+ }
+
+ /**
+ * Dispatches each SQL string to a worker (round-robin) and returns a handle
to the
+ * in-flight batch — this method does not block. The caller is responsible
for
+ * awaiting completion via {@link #awaitAll(Dispatch)} and collecting
errors. SQL text
+ * is intentionally not logged per-statement here: batched TOUCH/ADD
statements can
+ * be many kilobytes, and N parallel workers would multiply the log volume.
See
+ * {@link #awaitAll(Dispatch)} for the per-call summary log.
+ *
+ * <p>Statements are spread round-robin across workers, so worker <i>w</i>
owns
+ * indices {@code w, w + N, w + 2N, ...}. Each worker drains its own queue
+ * independently, which is why abort has to be observed by the tasks
themselves
+ * rather than by the awaiting thread — see {@link Dispatch}.
+ */
+ public Dispatch dispatchAll(List<String> sqls) {
+ if (closed) {
+ throw new IllegalStateException("Cannot dispatch to a closed
HiveDriverPool");
+ }
+ Dispatch dispatch = new Dispatch(sqls.size());
+ for (int i = 0; i < sqls.size(); i++) {
+ String sql = sqls.get(i);
+ Worker worker = workers.get(i % workers.size());
+ dispatch.add(worker.executor.submit(() -> {
+ // Abort check inside the task: a worker can dequeue this statement
while the
+ // awaiting thread is still parked on some other worker's slower
statement, so
+ // Future.cancel() alone cannot stop it in time. Checking here means no
+ // statement starts after a sibling has already failed.
+ if (dispatch.aborted()) {
+ throw new CancellationException("Skipped after an earlier statement
failed");
+ }
+ try {
+ worker.driver.run(sql);
+ } catch (Throwable t) {
+ dispatch.abort();
+ throw t;
+ } finally {
+ dispatch.taskSettled();
+ }
+ return null;
+ }));
+ }
+ dispatch.sealed();
+ return dispatch;
+ }
+
+ /**
+ * Awaits the dispatched batch and throws the first error encountered.
Errors are
+ * observed in <i>completion</i> order, not submission order: the awaiting
thread
+ * blocks until every task has settled (or the batch has aborted), so a
failure on a
+ * fast worker cancels the queues of all other workers even while a slow
worker is
+ * still mid-statement. Errors that finished before cancellation are logged
at WARN.
+ * Callers do not need per-statement results (Hive's Driver.run side-effects
the
+ * metastore), so this method is void.
+ */
+ public void awaitAll(Dispatch dispatch) {
+ long start = System.currentTimeMillis();
+ // Block until either every task settled or one of them aborted the batch.
Only
+ // then walk the futures — by that point no un-started task can still
begin, so
+ // the walk order no longer affects how much extra DDL gets applied.
+ dispatch.awaitSettledOrAborted();
+ int cancelled = dispatch.cancelPending();
+
+ Exception firstError = null;
+ int completed = 0;
+ for (Future<?> f : dispatch.futures()) {
+ try {
+ f.get();
+ completed++;
+ } catch (CancellationException ce) {
+ // Either we cancelled it before it started, or the task itself
observed the
+ // abort flag and bailed. Not a new failure; just note it for the
summary.
+ cancelled++;
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ if (firstError == null) {
+ firstError = ie;
+ }
+ } catch (ExecutionException ee) {
+ Exception cause = unwrap(ee);
+ if (cause instanceof CancellationException) {
+ cancelled++;
+ } else if (firstError == null) {
+ firstError = cause;
+ } else {
+ LOG.warn("Additional SQL batch failed (suppressed in favor of first
error)", cause);
+ }
+ }
+ }
+ if (firstError != null) {
+ throw new HoodieHiveSyncException("Failed in executing SQL", firstError);
+ }
+ LOG.info("Completed {} SQL statements ({} cancelled) in {} ms across {}
workers",
+ completed, cancelled, System.currentTimeMillis() - start, size);
+ }
+
+ /**
+ * Handle to one {@link #dispatchAll(List)} batch: the submitted futures
plus the
+ * shared abort flag the tasks consult before running.
+ *
+ * <p>The flag exists because the futures belong to N independent
single-thread
+ * executors. Cancelling from the awaiting thread is inherently late — a
worker can
+ * pull its next statement off its own queue at any moment — so each task
also
+ * re-checks {@link #aborted()} on entry. That is what actually bounds how
much extra
+ * partition DDL a failed sync can apply.
+ */
+ public static final class Dispatch {
+ private final List<Future<?>> futures;
+ private final int total;
+ private final AtomicInteger settled = new AtomicInteger(0);
+ private final AtomicBoolean aborted = new AtomicBoolean(false);
+ private final CountDownLatch done = new CountDownLatch(1);
+ private volatile boolean sealed;
+
+ private Dispatch(int total) {
+ this.total = total;
+ this.futures = new ArrayList<>(total);
+ }
+
+ private void add(Future<?> future) {
+ futures.add(future);
+ }
+
+ // Called once submission finishes. A task that settles before the last
submit
+ // would otherwise see settled < total and never trip the latch, so
re-check here.
+ private void sealed() {
+ sealed = true;
+ signalIfComplete();
+ }
+
+ private boolean aborted() {
+ return aborted.get();
+ }
+
+ private void abort() {
+ aborted.set(true);
+ done.countDown();
+ }
+
+ private void taskSettled() {
+ settled.incrementAndGet();
+ signalIfComplete();
+ }
+
+ private void signalIfComplete() {
+ if (sealed && settled.get() >= total) {
+ done.countDown();
+ }
+ }
+
+ private void awaitSettledOrAborted() {
+ if (total == 0) {
+ return;
+ }
+ try {
+ done.await();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ aborted.set(true);
+ }
+ }
+
+ // mayInterruptIfRunning=false: the worker thread is bound to a Hive
Driver whose
+ // state we don't want to corrupt mid-statement. Cancel only tasks that
haven't
+ // started; in-flight statements run to completion.
+ private int cancelPending() {
+ int cancelled = 0;
+ for (Future<?> f : futures) {
+ if (f.cancel(false)) {
+ cancelled++;
+ }
+ }
+ return cancelled;
+ }
+
+ private List<Future<?>> futures() {
+ return futures;
+ }
+
+ @VisibleForTesting
+ public int size() {
+ return futures.size();
+ }
+
+ @VisibleForTesting
+ public Future<?> futureAt(int index) {
+ return futures.get(index);
+ }
+ }
+
+ private static Exception unwrap(ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ return (cause instanceof Exception) ? (Exception) cause : ee;
+ }
+
+ public int size() {
+ return size;
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ tearDown();
+ }
+
+ private void tearDown() {
+ // Close each worker's own Driver and SessionState on its own thread, then
shut
+ // the executor down. Each worker owns an exclusive SessionState (see
+ // DefaultDriverFactory), so there is no cross-worker close ordering to
worry
+ // about here — closing worker i never affects worker j.
+ for (Worker worker : workers) {
+ try {
+ worker.executor.submit(() -> {
+ if (worker.driver != null) {
+ try {
+ worker.driver.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing pooled Driver", e);
+ }
+ }
+ if (worker.sessionState != null) {
+ try {
+ worker.sessionState.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing pooled SessionState", e);
+ }
+ }
+ return null;
+ }).get(30, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ LOG.warn("Error during pool worker shutdown", e);
+ }
+ worker.executor.shutdown();
+ try {
+ if (!worker.executor.awaitTermination(10, TimeUnit.SECONDS)) {
+ worker.executor.shutdownNow();
+ }
+ } catch (InterruptedException ie) {
+ worker.executor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+ workers.clear();
+ }
+
+ /**
+ * Per-slot state: a single-thread executor and the Driver + SessionState
bound to
+ * its thread. Both are volatile because they are written by the bootstrap
task and
+ * read by subsequent dispatch/teardown tasks on the same executor.
+ */
+ private static final class Worker {
+ final ExecutorService executor;
+ volatile Driver driver;
+ volatile SessionState sessionState;
+
+ Worker(ThreadFactory threadFactory) {
+ this.executor = Executors.newSingleThreadExecutor(threadFactory);
+ }
+ }
+
+ @FunctionalInterface
+ interface DriverFactory {
+ Driver newDriver(String databaseName) throws Exception;
+ }
+
+ /**
+ * Builds a real Hive {@link Driver} on the calling thread, backed by a
+ * {@link SessionState} that is exclusively owned by that thread (not shared
with
+ * any other worker). Hive's session-scoped state (current database, scratch
+ * directories, and the transaction/lock manager under {@code DbTxnManager})
is
+ * mutated by {@code Driver.run()} and is not safe for concurrent use from
multiple
+ * threads, so each worker must have its own instance. Bootstrap of all
workers is
+ * done sequentially by the pool constructor specifically so these
constructions
+ * don't race each other (e.g. on scratch-dir creation).
+ *
+ * <p>Each worker also gets its own {@link HiveConf} copy. {@code
SessionState}
+ * retains whatever {@code HiveConf} it's given, and {@code
QueryState}/{@code Driver}
+ * mutate per-query keys on that conf during {@code run()} (e.g. {@code
HIVEQUERYID}).
+ * Sharing one {@code HiveConf} instance across workers would let concurrent
+ * {@code Driver.run()} calls overwrite each other's query-scoped
configuration even
+ * though each worker has its own {@code SessionState} object.
+ */
+ private static final class DefaultDriverFactory implements DriverFactory {
+ private final HiveConf hiveConf;
+
+ DefaultDriverFactory(HiveSyncConfig config) {
+ this.hiveConf = config.getHiveConf();
+ }
+
+ @Override
+ public Driver newDriver(String databaseName) throws Exception {
+ HiveConf workerConf = new HiveConf(hiveConf);
+ SessionState sessionState = new SessionState(workerConf,
+ UserGroupInformation.getCurrentUser().getShortUserName());
+ sessionState.setCurrentDatabase(databaseName);
+ SessionState.start(sessionState);
+ return new Driver(workerConf);
+ }
+ }
+
+ private static final class PoolThreadFactory implements ThreadFactory {
+ private static final AtomicInteger POOL_ID = new AtomicInteger(0);
+ private final AtomicInteger threadId = new AtomicInteger(0);
+ private final String namePrefix = "hudi-hive-driver-pool-" +
POOL_ID.incrementAndGet() + "-";
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r, namePrefix + threadId.incrementAndGet());
+ t.setDaemon(true);
+ return t;
+ }
+ }
+}
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 d524d1c44064..4f109c9cdd64 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
@@ -96,9 +96,12 @@ import static
org.apache.hudi.hadoop.fs.HadoopFSUtils.getRelativePartitionPath;
import static
org.apache.hudi.hive.HiveSyncConfig.HIVE_SYNC_FILTER_PUSHDOWN_ENABLED;
import static org.apache.hudi.hive.HiveSyncConfig.RECREATE_HIVE_TABLE_ON_ERROR;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_AUTO_CREATE_DATABASE;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_BATCH_SYNC_PARTITION_NUM;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_CREATE_MANAGED_TABLE;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_IGNORE_EXCEPTIONS;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_AS_DATA_SOURCE_TABLE;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_BATCHING_ENABLED;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_BATCHING_THREADS;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_COMMENT;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY;
@@ -329,6 +332,110 @@ public class TestHiveSyncTool {
"Table partitions should match the number of partitions we wrote");
}
+ /**
+ * Exercises HiveQL sync with parallel partition batching enabled. Routes
through
+ * the HiveDriverPool — each worker thread owns a Driver+SessionState pair,
and
+ * the SQL list (qualified with `db`.`tbl`) is fanned out across them.
+ */
+ @Test
+ public void testHiveQLSyncWithBatchingEnabled() throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(),
HiveSyncMode.HIVEQL.name());
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_ENABLED.key(), "true");
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_THREADS.key(), "3");
+ hiveSyncProps.setProperty(HIVE_BATCH_SYNC_PARTITION_NUM.key(), "3");
+
+ int partitionCount = 10;
+ HiveTestUtil.createCOWTable("100", partitionCount, true);
+
+ reInitHiveSyncClient();
+ assertFalse(hiveClient.tableExists(HiveTestUtil.TABLE_NAME),
+ "Table should not exist before initial sync");
+ reSyncHiveTable();
+ assertEquals(partitionCount,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size(),
+ "All partitions should be added under parallel HiveQL batching");
+
+ // Add more partitions, then sync again to exercise the parallel update
path.
+ HiveTestUtil.addCOWPartition("2050/01/01", true, true, "101");
+ HiveTestUtil.addCOWPartition("2050/01/02", true, true, "102");
+ HiveTestUtil.addCOWPartition("2050/01/03", true, true, "103");
+ HiveTestUtil.addCOWPartition("2050/01/04", true, true, "104");
+ reInitHiveSyncClient();
+ reSyncHiveTable();
+ assertEquals(partitionCount + 4,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size(),
+ "Incremental add via parallel HiveQL batching should sync the new
partitions");
+ }
+
+ /**
+ * Exercises the SET_LOCATION path in HiveQL mode with batching on.
SET_LOCATION
+ * emits one ALTER PARTITION ... SET LOCATION statement per partition (Hive
SQL
+ * has no multi-partition SET LOCATION), so this is the fan-out path most
likely
+ * to exercise concurrent ALTER PARTITION calls against the same table.
+ */
+ @Test
+ public void testHiveQLSetLocationWithBatching() throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(),
HiveSyncMode.HIVEQL.name());
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_ENABLED.key(), "true");
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_THREADS.key(), "3");
+ hiveSyncProps.setProperty(HIVE_BATCH_SYNC_PARTITION_NUM.key(), "2");
+
+ int partitionCount = 6;
+ HiveTestUtil.createCOWTable("100", partitionCount, true);
+ reInitHiveSyncClient();
+ reSyncHiveTable();
+ assertEquals(partitionCount,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size());
+
+ // Drive the SET_LOCATION path by directly calling updatePartitionsToTable
with
+ // existing partition paths. Each partition produces its own ALTER ... SET
LOCATION
+ // statement, fanned out across the 3 workers in the pool.
+ List<String> existingPartitions =
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).stream()
+ .map(p -> getRelativePartitionPath(new Path(basePath), new
Path(p.getStorageLocation())))
+ .collect(Collectors.toList());
+ hiveClient.updatePartitionsToTable(HiveTestUtil.TABLE_NAME,
existingPartitions);
+
+ List<Partition> after =
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
+ assertEquals(partitionCount, after.size(),
+ "Parallel SET_LOCATION must not change the partition set");
+ Set<String> relativePaths = after.stream()
+ .map(p -> getRelativePartitionPath(new Path(basePath), new
Path(p.getStorageLocation())))
+ .collect(Collectors.toSet());
+ assertEquals(partitionCount, relativePaths.size(),
+ "Each partition should resolve to a unique relative path after
parallel SET_LOCATION");
+ assertTrue(relativePaths.containsAll(existingPartitions),
+ "All original partition paths should still be present after parallel
SET_LOCATION");
+ }
+
+ /**
+ * Exercises the TOUCH path in HiveQL mode with batching on. Verifies that
+ * splitting one giant ALTER TABLE TOUCH PARTITION(...)... into multiple
smaller
+ * statements does not break partition visibility downstream.
+ */
+ @Test
+ public void testHiveQLTouchPartitionsWithBatching() throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(),
HiveSyncMode.HIVEQL.name());
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_ENABLED.key(), "true");
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_THREADS.key(), "2");
+ hiveSyncProps.setProperty(HIVE_BATCH_SYNC_PARTITION_NUM.key(), "2");
+ hiveSyncProps.setProperty(META_SYNC_TOUCH_PARTITIONS_ENABLED.key(),
"true");
+
+ int partitionCount = 6;
+ HiveTestUtil.createCOWTable("100", partitionCount, true);
+ reInitHiveSyncClient();
+ reSyncHiveTable();
+ assertEquals(partitionCount,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size());
+
+ // Drive the TOUCH path directly by calling touchPartitionsToTable with
existing
+ // partition paths. Partitions are batched into groups of
HIVE_BATCH_SYNC_PARTITION_NUM
+ // and each batch's ALTER ... TOUCH statement is fanned out across the 2
workers in
+ // the pool.
+ List<String> existingPartitions =
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).stream()
+ .map(p -> getRelativePartitionPath(new Path(basePath), new
Path(p.getStorageLocation())))
+ .collect(Collectors.toList());
+ hiveClient.touchPartitionsToTable(HiveTestUtil.TABLE_NAME,
existingPartitions);
+
+ assertEquals(partitionCount,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size(),
+ "TOUCH batching must not change the partition set");
+ }
+
@ParameterizedTest
@MethodSource({"syncModeAndSchemaFromCommitMetadata"})
public void testBasicSync(boolean useSchemaFromCommitMetadata, String
syncMode, String enablePushDown) throws Exception {
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestQueryBasedDDLExecutorTouchBatching.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestQueryBasedDDLExecutorTouchBatching.java
new file mode 100644
index 000000000000..73d23859dbfa
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestQueryBasedDDLExecutorTouchBatching.java
@@ -0,0 +1,192 @@
+/*
+ * 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.ddl;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.hive.HiveSyncConfig;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_BATCH_SYNC_PARTITION_NUM;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_BATCHING_ENABLED;
+import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_BASE_PATH;
+import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
+import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_PARTITION_FIELDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Verifies which execution paths TOUCH batching applies to.
+ *
+ * <p>{@code hoodie.datasource.hive_sync.batching.enabled} only makes sense
where the
+ * resulting statements are dispatched in parallel. {@link
QueryBasedDDLExecutor} is
+ * also the base class for {@link JDBCExecutor}, which executes the list
serially — so
+ * splitting there would change statement count and partial-application
semantics for
+ * no benefit. The split is therefore driven by {@link
QueryBasedDDLExecutor#getTouchBatchSize(int)},
+ * which only {@link HiveQueryDDLExecutor} overrides (and only when a driver
pool is
+ * actually present).
+ */
+class TestQueryBasedDDLExecutorTouchBatching {
+
+ private static final String TABLE_NAME = "tbl";
+ private static final int PARTITION_COUNT = 5;
+ private static final int BATCH_SIZE = 2;
+
+ /**
+ * Captures the SQL handed to the executor instead of running it. Uses the
base
+ * class's serial {@code runSQLs}, exactly as {@link JDBCExecutor} does.
+ *
+ * <p>{@code parallelBatchSize} stands in for a subclass that dispatches in
parallel:
+ * when set, {@link #getTouchBatchSize(int)} returns it, mimicking
+ * {@link HiveQueryDDLExecutor} with a driver pool present. When unset, the
base-class
+ * default applies — the JDBC-mode shape.
+ */
+ private static final class RecordingExecutor extends QueryBasedDDLExecutor {
+ private final List<String> executed = new ArrayList<>();
+ private final Integer parallelBatchSize;
+
+ RecordingExecutor(HiveSyncConfig config) {
+ this(config, null);
+ }
+
+ RecordingExecutor(HiveSyncConfig config, Integer parallelBatchSize) {
+ super(config);
+ this.parallelBatchSize = parallelBatchSize;
+ }
+
+ @Override
+ protected int getTouchBatchSize(int partitionCount) {
+ return parallelBatchSize != null ? parallelBatchSize :
super.getTouchBatchSize(partitionCount);
+ }
+
+ @Override
+ public void runSQL(String sql) {
+ executed.add(sql);
+ }
+
+ @Override
+ public Map<String, String> getTableSchema(String tableName) {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public void dropPartitionsToTable(String tableName, List<String>
partitionsToDrop) {
+ // not exercised here
+ }
+
+ @Override
+ public void close() {
+ // no resources held
+ }
+ }
+
+ private static HiveSyncConfig configWithBatching(boolean batchingEnabled) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(META_SYNC_DATABASE_NAME.key(), "db");
+ props.setProperty(META_SYNC_BASE_PATH.key(), "file:///tmp/base");
+ props.setProperty(META_SYNC_PARTITION_FIELDS.key(), "dt");
+ props.setProperty(HIVE_BATCH_SYNC_PARTITION_NUM.key(),
String.valueOf(BATCH_SIZE));
+ props.setProperty(HIVE_SYNC_BATCHING_ENABLED.key(),
String.valueOf(batchingEnabled));
+ return new HiveSyncConfig(props);
+ }
+
+ private static List<String> partitions() {
+ return IntStream.range(0, PARTITION_COUNT)
+ .mapToObj(i -> "2026-01-0" + (i + 1))
+ .collect(Collectors.toList());
+ }
+
+ private static List<String> touchStatements(RecordingExecutor executor) {
+ // constructPartitionAlterStatements always emits a leading `USE db`; the
TOUCH
+ // statements are everything after it.
+ return executor.executed.stream()
+ .filter(sql -> sql.contains(" TOUCH "))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Given a serial executor (the JDBC-mode shape) with batching enabled, when
TOUCH is
+ * issued for more partitions than the batch size, then a single TOUCH
statement is
+ * still emitted — the flag must not reach non-parallel execution paths.
+ */
+ @Test
+ void serialExecutorEmitsSingleTouchStatementEvenWithBatchingEnabled() {
+ RecordingExecutor executor = new
RecordingExecutor(configWithBatching(true));
+
+ executor.touchPartitionsToTable(TABLE_NAME, partitions());
+
+ List<String> touches = touchStatements(executor);
+ assertEquals(1, touches.size(),
+ "Serial executors (e.g. JDBC mode) must emit one TOUCH statement
regardless of "
+ + "hoodie.datasource.hive_sync.batching.enabled");
+ assertEquals(PARTITION_COUNT, countPartitionClauses(touches.get(0)),
+ "The single statement must still cover every partition");
+ }
+
+ /**
+ * Given the same executor with batching disabled, when TOUCH is issued,
then the SQL
+ * is byte-identical to the enabled case — pinning that the flag is a no-op
here.
+ */
+ @Test
+ void serialExecutorTouchSqlIsIdenticalWithAndWithoutBatchingFlag() {
+ RecordingExecutor withFlag = new
RecordingExecutor(configWithBatching(true));
+ RecordingExecutor withoutFlag = new
RecordingExecutor(configWithBatching(false));
+
+ withFlag.touchPartitionsToTable(TABLE_NAME, partitions());
+ withoutFlag.touchPartitionsToTable(TABLE_NAME, partitions());
+
+ assertEquals(withoutFlag.executed, withFlag.executed,
+ "Enabling the batching flag must not change JDBC-mode TOUCH SQL
shape");
+ }
+
+ /**
+ * Given an executor that reports a parallel-dispatch batch size (the
HiveQL-with-pool
+ * shape), when TOUCH is issued, then partitions are split across multiple
statements.
+ * This pins that the base-class default is the only thing suppressing the
split.
+ */
+ @Test
+ void parallelExecutorSplitsTouchIntoBatches() {
+ RecordingExecutor executor = new
RecordingExecutor(configWithBatching(true), BATCH_SIZE);
+
+ executor.touchPartitionsToTable(TABLE_NAME, partitions());
+
+ List<String> touches = touchStatements(executor);
+ // 5 partitions at 2 per batch -> 3 statements (2, 2, 1).
+ assertEquals(3, touches.size());
+ assertEquals(PARTITION_COUNT,
+
touches.stream().mapToInt(TestQueryBasedDDLExecutorTouchBatching::countPartitionClauses).sum(),
+ "Batching must not drop or duplicate partitions");
+ }
+
+ private static int countPartitionClauses(String sql) {
+ int count = 0;
+ int idx = sql.indexOf("PARTITION (");
+ while (idx >= 0) {
+ count++;
+ idx = sql.indexOf("PARTITION (", idx + 1);
+ }
+ return count;
+ }
+}
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
new file mode 100644
index 000000000000..fd58b7023aa1
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
@@ -0,0 +1,316 @@
+/*
+ * 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.exception.HoodieException;
+import org.apache.hudi.hive.HiveSyncConfig;
+import org.apache.hudi.hive.HoodieHiveSyncException;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.Driver;
+import org.junit.jupiter.api.Test;
+import org.mockito.invocation.InvocationOnMock;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Unit tests for {@link HiveDriverPool} that exercise bootstrap, dispatch,
error
+ * propagation, and close semantics without standing up a real Hive instance.
+ */
+class TestHiveDriverPool {
+
+ private static HiveSyncConfig configWithEmptyHiveConf() {
+ HiveSyncConfig config = mock(HiveSyncConfig.class);
+ doAnswer(inv -> new HiveConf()).when(config).getHiveConf();
+ doAnswer(inv -> "default").when(config).getStringOrDefault(
+ org.mockito.ArgumentMatchers.any());
+ return config;
+ }
+
+ @Test
+ void bootstrapBuildsOneDriverPerSlot() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ AtomicInteger built = new AtomicInteger();
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ built.incrementAndGet();
+ return mock(Driver.class);
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 3, factory)) {
+ assertEquals(3, pool.size());
+ assertEquals(3, built.get(), "One Driver per slot should be constructed
eagerly");
+ }
+ }
+
+ @Test
+ void bootstrapFailurePropagatesAndTearsDown() {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ AtomicInteger calls = new AtomicInteger();
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ int n = calls.incrementAndGet();
+ if (n == 2) {
+ throw new RuntimeException("simulated driver build failure");
+ }
+ return mock(Driver.class);
+ };
+ HoodieException ex = assertThrows(HoodieException.class,
+ () -> new HiveDriverPool(config, 3, factory));
+ assertTrue(ex.getMessage().contains("Failed to construct HiveDriverPool"));
+ }
+
+ @Test
+ void runAllDispatchesEachSqlAcrossWorkers() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ // Each worker counts how many SQLs it received and remembers the thread.
+ ConcurrentHashMap<Driver, Set<String>> seenThreadsByDriver = new
ConcurrentHashMap<>();
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ seenThreadsByDriver.put(d, ConcurrentHashMap.newKeySet());
+ doAnswer((InvocationOnMock inv) -> {
+ seenThreadsByDriver.get(d).add(Thread.currentThread().getName());
+ return null;
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
+ List<String> sqls = Arrays.asList("SELECT 1", "SELECT 2", "SELECT 3",
"SELECT 4");
+ HiveDriverPool.Dispatch futures = pool.dispatchAll(sqls);
+ pool.awaitAll(futures);
+ assertEquals(2, seenThreadsByDriver.size(), "Expected exactly 2 worker
Drivers");
+ int totalCalls =
seenThreadsByDriver.values().stream().mapToInt(Set::size).sum();
+ assertTrue(totalCalls >= 1, "At least one worker should have logged a
thread");
+ // Each Driver should have been invoked exactly twice (round-robin with
4 sqls, 2 workers).
+ for (Driver d : seenThreadsByDriver.keySet()) {
+ verify(d, times(2)).run(anyString());
+ }
+ }
+ }
+
+ @Test
+ void awaitAllThrowsFirstError() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ doAnswer(inv -> {
+ String sql = inv.getArgument(0);
+ if (sql.equals("FAIL")) {
+ throw new RuntimeException("boom: " + sql);
+ }
+ return null;
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
+ HiveDriverPool.Dispatch futures = pool.dispatchAll(Arrays.asList("OK",
"FAIL", "OK"));
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> pool.awaitAll(futures));
+ assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ }
+ }
+
+ @Test
+ void concurrentDispatchBoundedByPoolSize() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ AtomicInteger inFlight = new AtomicInteger();
+ AtomicInteger maxInFlight = new AtomicInteger();
+ CountDownLatch hold = new CountDownLatch(1);
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ doAnswer(inv -> {
+ int now = inFlight.incrementAndGet();
+ maxInFlight.updateAndGet(prev -> Math.max(prev, now));
+ hold.await(2, TimeUnit.SECONDS);
+ inFlight.decrementAndGet();
+ return null;
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
+ // 5 SQLs against pool of size 2 → max in-flight should be 2.
+ HiveDriverPool.Dispatch futures = pool.dispatchAll(Arrays.asList("a",
"b", "c", "d", "e"));
+ // Release after a short wait so all SQLs progress.
+ Thread.sleep(150);
+ hold.countDown();
+ pool.awaitAll(futures);
+ assertTrue(maxInFlight.get() <= 2,
+ "Max concurrent dispatches must not exceed pool size, observed " +
maxInFlight.get());
+ assertTrue(maxInFlight.get() >= 1, "Sanity: at least one dispatch ran");
+ }
+ }
+
+ @Test
+ void closeIsIdempotentAndPreventsFurtherDispatch() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ HiveDriverPool.DriverFactory factory = (db) -> mock(Driver.class);
+ HiveDriverPool pool = new HiveDriverPool(config, 2, factory);
+ pool.close();
+ pool.close();
+ assertThrows(IllegalStateException.class,
+ () -> pool.dispatchAll(Arrays.asList("anything")));
+ }
+
+ @Test
+ void invalidSizeRejected() {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ HiveDriverPool.DriverFactory factory = (db) -> mock(Driver.class);
+ assertThrows(IllegalArgumentException.class,
+ () -> new HiveDriverPool(config, 0, factory));
+ }
+
+ /**
+ * runOnEachWorker must execute the setup SQL on every worker (each on its
bound
+ * thread) before {@code dispatchAll()} fans the partition statements out.
Without this,
+ * Hive 2.x's SET LOCATION would silently route to the wrong database on the
workers
+ * that never saw the leading USE statement.
+ */
+ @Test
+ void runOnEachWorkerRunsSetupOnEveryWorker() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ ConcurrentHashMap<Driver, List<String>> sqlsByDriver = new
ConcurrentHashMap<>();
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ sqlsByDriver.put(d, java.util.Collections.synchronizedList(new
java.util.ArrayList<>()));
+ doAnswer((InvocationOnMock inv) -> {
+ sqlsByDriver.get(d).add(inv.getArgument(0));
+ return null;
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 3, factory)) {
+ pool.runOnEachWorker(Arrays.asList("USE `db1`"));
+ HiveDriverPool.Dispatch futures = pool.dispatchAll(Arrays.asList("ALTER
1", "ALTER 2", "ALTER 3"));
+ pool.awaitAll(futures);
+
+ assertEquals(3, sqlsByDriver.size(), "Expected one Driver per worker");
+ for (Map.Entry<Driver, List<String>> e : sqlsByDriver.entrySet()) {
+ List<String> seen = e.getValue();
+ assertTrue(!seen.isEmpty() && seen.get(0).equals("USE `db1`"),
+ "Each worker must see USE first; saw " + seen);
+ }
+ }
+ }
+
+ /**
+ * Given a single-worker pool where the first statement fails, when awaitAll
runs,
+ * then it throws the original cause and neither queued statement is ever
executed.
+ *
+ * <p>Deterministic: statements queued behind the failure observe the
batch's abort
+ * flag on entry and bail out without touching the Driver, so it does not
matter
+ * whether the worker dequeues them before or after awaitAll's cancel()
sweep.
+ */
+ @Test
+ void awaitAllCancelsPendingFuturesOnFirstError() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ List<String> executed = Collections.synchronizedList(new ArrayList<>());
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ doAnswer(inv -> {
+ String sql = inv.getArgument(0);
+ executed.add(sql);
+ if (sql.equals("FAIL")) {
+ throw new RuntimeException("boom");
+ }
+ return null;
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 1, factory)) {
+ HiveDriverPool.Dispatch dispatch =
pool.dispatchAll(Arrays.asList("FAIL", "PENDING_A", "PENDING_B"));
+
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> pool.awaitAll(dispatch));
+
+ assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ assertEquals(Collections.singletonList("FAIL"), executed,
+ "Statements queued behind the failure must never reach the Driver");
+ }
+ }
+
+ /**
+ * Regression for the in-order-await bug: a slow statement on worker 0 must
not let
+ * worker 1 keep applying partition DDL after worker 1 has already failed.
+ *
+ * <p>Given two workers, statements are dispatched round-robin — worker 0
gets
+ * {@code SLOW} and worker 1 gets {@code FAIL} then {@code AFTER_FAIL}. When
awaitAll
+ * blocks in submission order, it parks on SLOW's future while worker 1
races ahead
+ * and runs AFTER_FAIL. Then AFTER_FAIL must never execute: the abort flag
is set by
+ * FAIL before worker 1 can dequeue its next statement.
+ *
+ * <p>SLOW is released only after the batch has aborted, which pins the
interleaving
+ * the bug needs — without that, SLOW could finish first and mask the race.
+ */
+ @Test
+ void awaitAllStopsLaterWorkerWhenEarlierFutureIsSlow() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ List<String> executed = Collections.synchronizedList(new ArrayList<>());
+ CountDownLatch failed = new CountDownLatch(1);
+ CountDownLatch releaseSlow = new CountDownLatch(1);
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ doAnswer(inv -> {
+ String sql = inv.getArgument(0);
+ executed.add(sql);
+ if (sql.equals("FAIL")) {
+ failed.countDown();
+ throw new RuntimeException("boom");
+ }
+ if (sql.equals("SLOW")) {
+ // Hold worker 0 until worker 1 has failed, so awaitAll is
definitely still
+ // parked on future 0 at the moment worker 1 would pick up
AFTER_FAIL.
+ releaseSlow.await(5, TimeUnit.SECONDS);
+ }
+ return null;
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
+ // Round-robin over 2 workers: index 0 -> worker 0, indices 1 and 2 ->
worker 1.
+ HiveDriverPool.Dispatch dispatch =
+ pool.dispatchAll(Arrays.asList("SLOW", "FAIL", "AFTER_FAIL"));
+ assertTrue(failed.await(5, TimeUnit.SECONDS), "FAIL must have run");
+ releaseSlow.countDown();
+
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> pool.awaitAll(dispatch));
+
+ assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ assertFalse(executed.contains("AFTER_FAIL"),
+ "Statement queued behind a failure on the same worker must not be
applied, "
+ + "even while an earlier future on another worker is still
running");
+ }
+ }
+}
