danny0405 commented on code in PR #18984:
URL: https://github.com/apache/hudi/pull/18984#discussion_r3575446303
##########
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:
The new opt-in flag is only consulted when constructing the Driver pool, but
this path always splits TOUCH into `batch_num` chunks. With
`hoodie.datasource.hive_sync.batching.enabled` left at its default `false`, a
large TOUCH now becomes multiple ALTER statements instead of the previous
single statement, changing failure/partial-application semantics despite the
documented "existing behavior is unchanged" contract. Please gate this split on
`HIVE_SYNC_BATCHING_ENABLED` (use one batch when false), or explicitly make and
document TOUCH batching as unconditional.
##########
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:
This second sync has no new commit. With incremental sync enabled by
default, instant `100` is already recorded by the first sync, so
`getWrittenPartitionsSince` returns an empty list and
`HiveSyncTool.syncPartitions` returns before calling `touchPartitionsToTable`.
As a result, this test never reaches the batched TOUCH path it claims to cover.
Please directly call `hiveClient.touchPartitionsToTable` with the existing
partition paths (as the SET_LOCATION test does), or add a new commit that
rewrites existing partitions and verify the TOUCH behavior.
##########
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:
This assertion is racy and fails consistently in the focused test run (the
initial run plus all three Surefire retries). Once `FAIL` returns, the
single-thread executor can immediately run and complete `PENDING_A` before
`awaitAll` observes the failed future and cancels the queue, in which case
`isCancelled()` is correctly false. Please block the pending driver calls on a
latch until after the cancellation assertions (releasing it in `finally`), or
otherwise make the queued-task state deterministic.
--
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]
