danny0405 commented on code in PR #18984: URL: https://github.com/apache/hudi/pull/18984#discussion_r3649367144
########## hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java: ########## @@ -0,0 +1,342 @@ +/* + * 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.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.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.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; + } + List<Future<?>> futures = new ArrayList<>(workers.size()); + for (Worker worker : workers) { + futures.add(worker.executor.submit(() -> { + for (String sql : setupSqls) { + worker.driver.run(sql); + } + return null; + })); + } + awaitAll(futures); + } + + /** + * Dispatches each SQL string to a worker (round-robin) and returns the list of + * in-flight futures — this method does not block. The caller is responsible for + * awaiting completion via {@link #awaitAll(List)} 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(List)} for the per-call summary log. + */ + public List<Future<?>> dispatchAll(List<String> sqls) { + if (closed) { + throw new IllegalStateException("Cannot dispatch to a closed HiveDriverPool"); + } + List<Future<?>> futures = new ArrayList<>(sqls.size()); + for (int i = 0; i < sqls.size(); i++) { + String sql = sqls.get(i); + Worker worker = workers.get(i % workers.size()); + futures.add(worker.executor.submit(() -> { + worker.driver.run(sql); + return null; + })); + } + return futures; + } + + /** + * Awaits all futures and throws the first exception encountered. On first failure, + * cancels the remaining (not yet started) futures so workers don't keep running + * pointless work after a fatal error. Any 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(List<Future<?>> futures) { + long start = System.currentTimeMillis(); + Exception firstError = null; + int completed = 0; + int cancelled = 0; + for (int i = 0; i < futures.size(); i++) { + Future<?> f = futures.get(i); + try { + f.get(); Review Comment: awaitAll observes completions in submission order because this loop blocks on each Future.get(). If an earlier future is slow while a later worker fails, that worker can immediately dequeue and execute more partition DDL before this loop reaches the failed future and calls cancelRemaining. I reproduced this deterministically with two workers: the later worker ran FAIL and then a queued AFTER_FAIL statement while future 0 was still blocked. This means the advertised "cancel pending futures on first error" behavior does not hold, and a failed sync can apply avoidable extra partition changes. Could we consume results in completion order and cancel all outstanding submissions as soon as any completed task fails, with a regression test covering an earlier slow future and a later fast failure? ########## hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/QueryBasedDDLExecutor.java: ########## @@ -210,29 +224,43 @@ 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: when {@code HIVE_SYNC_BATCHING_ENABLED} is set, one + * {@code ALTER TABLE ... TOUCH PARTITION (p1) ...} per batch of + * {@code HIVE_BATCH_SYNC_PARTITION_NUM} partitions. Otherwise a single statement + * covering all partitions, matching pre-batching behavior. + * + * <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.getBooleanOrDefault(HIVE_SYNC_BATCHING_ENABLED) Review Comment: QueryBasedDDLExecutor is also the base class for JDBCExecutor, so this flag is not actually HiveQL-only here. With hoodie.datasource.hive_sync.batching.enabled=true in JDBC mode, TOUCH is still split into batch_num statements and the default runSQLs executes those batches serially. That contradicts the config and impact documentation saying JDBC is unaffected, and it changes JDBC statement count and partial-application semantics. Could we scope this decision to HiveQueryDDLExecutor or actual pool presence, for example via a HiveQL-only hook, and add a regression test showing that enabling this flag does not change JDBC TOUCH SQL shape? -- 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]
