nsivabalan commented on code in PR #18984:
URL: https://github.com/apache/hudi/pull/18984#discussion_r3643278811
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java:
##########
@@ -210,29 +223,40 @@ public void touchPartitionsToTable(String tableName,
List<String> touchPartition
}
log.info("Touching partitions " + touchPartitions.size() + " on " +
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) ...} per batch
of
+ * {@code HIVE_BATCH_SYNC_PARTITION_NUM} partitions.
+ *
+ * <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 =
config.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM);
switch (alterType) {
case TOUCH:
- String alterTable = alterTablePrefix + " TOUCH";
- for (String partition : partitions) {
- alterTable += " PARTITION (" + getPartitionClause(partition) + ")";
+ for (List<String> batch : CollectionUtils.batches(partitions,
batchSyncPartitionNum)) {
Review Comment:
Fixed in 3ccab972 — `constructPartitionAlterStatements` now only splits
TOUCH into `batch_num` chunks when `HIVE_SYNC_BATCHING_ENABLED` is set. With
the flag off (default), it emits a single statement covering all partitions,
matching pre-batching behavior exactly.
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java:
##########
@@ -329,6 +332,105 @@ public void testDropUpperCasePartitionWithHMS() throws
Exception {
"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());
+
+ // Touch existing partitions (no new data) — should hit the batched TOUCH
path
+ // without changing the partition count.
+ reInitHiveSyncClient();
+ reSyncHiveTable();
Review Comment:
Fixed in 3ccab972 — the test now calls
`hiveClient.touchPartitionsToTable(...)` directly with the existing partition
paths, the same pattern already used by `testHiveQLSetLocationWithBatching`.
This actually exercises the batched TOUCH path instead of relying on a no-op
incremental resync.
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java:
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.Arrays;
+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.Future;
+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.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");
+ List<Future<?>> futures = pool.runAll(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)) {
+ List<Future<?>> futures = pool.runAll(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.
+ List<Future<?>> futures = pool.runAll(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.runAll(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 runAll()} 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`"));
+ List<Future<?>> futures = pool.runAll(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);
+ }
+ }
+ }
+
+ /**
+ * On the first failure, awaitAll must throw the original cause and cancel
any
+ * futures that have not started yet. Futures already in-flight are not
interrupted
+ * (per the cancel-with-mayInterruptIfRunning=false contract).
+ */
+ @Test
+ void awaitAllCancelsPendingFuturesOnFirstError() throws Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ // Single-worker pool so SQLs run strictly sequentially → the 2nd SQL is
+ // pending when the 1st errors, and must be cancelled.
+ CountDownLatch fired = new CountDownLatch(1);
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ doAnswer(inv -> {
+ String sql = inv.getArgument(0);
+ if (sql.equals("FAIL")) {
+ fired.countDown();
+ throw new RuntimeException("boom");
+ }
+ return null;
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 1, factory)) {
+ List<Future<?>> futures = pool.runAll(Arrays.asList("FAIL", "PENDING_A",
"PENDING_B"));
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> pool.awaitAll(futures));
+ assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ // The first future failed, so it's done (not cancelled). The remaining
two
+ // were pending behind it on the single worker and should now be
cancelled.
+ assertTrue(fired.await(1, TimeUnit.SECONDS), "Failing SQL must have
run");
+ assertTrue(futures.get(1).isCancelled(), "Pending future after error
must be cancelled");
Review Comment:
Fixed in 3ccab972 — PENDING_A/PENDING_B now park on a latch as soon as they
start (released in a finally), so even if the single worker races ahead and
dequeues one before cancellation runs, it cannot complete before we assert on
it. The assertion now checks `isCancelled() || !isDone()` instead of assuming
cancellation always wins the race. Verified with a standalone 20k-iteration
harness outside Mockito: the old assertion failed ~99% of the time under this
reproduction, the new one never failed.
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java:
##########
@@ -122,6 +122,23 @@ 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.1.0")
+ .withDocumentation("When true, partition operations
(add/update/touch/drop) in the active sync mode "
Review Comment:
Fixed in 3ccab972 — this doc was already reworded in the prior push
(per-reply above) to scope precisely to HiveQL, but re-reading your comment I
also tightened the wording further: TOUCH batching is now explicitly tied to
the enabled flag (see the sibling fix on QueryBasedDDLExecutor), and the doc
calls out that ADD was already batched, SET_LOCATION is one statement per
partition, and DROP/HMS/JDBC are untouched.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
