feat(hive-sync): parallelize DROP partitions in HiveQL sync mode - #19033
feat(hive-sync): parallelize DROP partitions in HiveQL sync mode#19033nsivabalan wants to merge 4 commits into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for the contribution! This PR extends the parallel batching infrastructure to HiveQL DROP partitions by wiring the existing IMetaStoreClientPool into HiveQueryDDLExecutor. The structure is cleanly gated behind the existing opt-in flag, and the test coverage for both the new pool plumbing and the end-to-end DROP path is solid. A couple of correctness concerns around resource cleanup and SessionState sharing are worth a closer look in the inline comments. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of nits on a field that exists purely for documentation and a duplicated futures-error loop.
| // SessionState is shared across workers; build it once on the first call (with | ||
| // currentDatabase already set) and attach it to each worker's thread-local on | ||
| // subsequent calls. The database is a pool-wide property and never changes | ||
| // across workers, so setting it once at construction time is sufficient. |
There was a problem hiding this comment.
🤖 The shared SessionState pattern here makes me a bit nervous — every worker thread calls SessionState.start(sharedSessionState) and then runs concurrent Driver.run() against the same instance. Hive's SessionState has plenty of mutable fields the Driver writes during query execution (e.g. currentDatabase via USE, hiveOperation, cmd, lineageState, tablesAlreadyMarkedAsRead). Are we confident none of these get written concurrently by parallel ALTER PARTITION / TOUCH statements? @yihua could you weigh in on whether sharing SessionState across worker threads is safe in practice, or whether each worker should construct its own (paying the SessionState init cost in exchange for isolation)?
- AI-generated; verify before applying. React 👍/👎 to flag quality.
There was a problem hiding this comment.
🤖 Sharing one SessionState across worker threads is risky here. Hive stores the active SessionState in a ThreadLocal (SessionState.start() sets tss.set(...)), so this pattern installs the same mutable instance into N threads' ThreadLocals. The Hive Driver isn't documented as thread-safe, and Driver execution mutates session-scoped state — currentDatabase (USE), conf overrides, and the compile/execute lineage/read-entity sets — so concurrent ALTER PARTITION/TOUCH runs can interleave on those fields. Worth confirming whether any of the batched statements can issue a USE or set conf; if so, per-worker SessionState (or serializing the session-mutating bits) is the safer default even at the init cost.
There was a problem hiding this comment.
This is the same shared-SessionState concern already fixed on #18984 (each HiveDriverPool worker now owns its own SessionState, built via sequential bootstrap). This branch is currently rebased on #18984's pre-fix commit, so the fix isn't visible here yet — it'll show up once #18984 merges and this branch rebases onto it. Leaving open until then rather than resolving.
There was a problem hiding this comment.
🤖 In the current HiveDriverPool design each worker actually gets its own SessionState rather than sharing one: DefaultDriverFactory.newDriver() does new SessionState(new HiveConf(hiveConf), ...) + SessionState.start(...) on the worker's own single-thread executor, and every statement for that slot runs pinned to that one thread (the Driver/SessionState are never touched from another thread). Each worker also gets its own HiveConf copy specifically so per-query keys like HIVEQUERYID don't collide. So there's no shared SessionState across threads here, and the concurrent USE/conf-mutation concern shouldn't apply — the per-worker isolation you suggested is what the code does.
| // 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(); |
There was a problem hiding this comment.
🤖 nit: the comment says this field is 'kept for reference only — close() is delegated through ddlExecutor.close()', which means it's never read after being passed into HiveQueryDDLExecutor. Could you make it a local variable in the constructor instead? A field that isn't read after construction will make future readers wonder what lifecycle role it plays.
- AI-generated; verify before applying. React 👍/👎 to flag quality.
There was a problem hiding this comment.
Noted — this partitionDriverPool field-vs-local-variable point applies to code introduced in #18984, not this PR. Tracking it there; will pick up on rebase.
| * semantics, so each worker still iterates its chunk one partition at a time — the | ||
| * win is fanning chunks across independent Thrift clients. | ||
| */ | ||
| private void runDropBatches(String tableName, List<List<String>> batches) throws Exception { |
There was a problem hiding this comment.
🤖 nit: the parallel-path error-collection loop inside runDropBatches (Exception firstError = null; for (Future<Void> f : futures) …) is nearly identical to the one in HMSDDLExecutor.runBatches. Since HiveDriverPool already has a well-factored awaitAll(List<Future<?>>) for exactly this pattern, it might be worth adding a similar awaitAll(List<Future<Void>>) to IMetaStoreClientPool so both callers can delegate to it instead.
- AI-generated; verify before applying. React 👍/👎 to flag quality.
There was a problem hiding this comment.
Good suggestion — once #18984 merges and this branch rebases, HiveDriverPool.awaitAll(List<Future<?>>) will be available; will look at whether IMetaStoreClientPool's runDropBatches error-collection loop here can delegate to an analogous awaitAll on IMetaStoreClientPool at that point, to de-duplicate the two nearly-identical loops.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19033 +/- ##
============================================
- Coverage 70.49% 68.68% -1.81%
+ Complexity 32775 31362 -1413
============================================
Files 2700 2562 -138
Lines 152896 148467 -4429
Branches 19009 18725 -284
============================================
- Hits 107778 101974 -5804
- Misses 36596 38175 +1579
+ Partials 8522 8318 -204
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
7022e7d to
5e3c1fd
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR extends the opt-in HiveQL batching to partition drops by fanning batches across an IMetaStoreClientPool, and routes ADD/UPDATE/TOUCH through a runSQLs hook backed by HiveDriverPool. The concurrency on the drop path itself looks sound (disjoint batches, partition-row-only ops, thread-safe DateTimeFormatter-based extractors). One error-path resource-leak edge case is worth double-checking in the inline comment. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small naming and consistency nits below, otherwise the code is clean and well-documented.
| public static final ConfigProperty<Integer> HIVE_SYNC_BATCHING_THREADS = ConfigProperty | ||
| .key("hoodie.datasource.hive_sync.batching.threads") | ||
| .defaultValue(4) | ||
| .markAdvanced() |
There was a problem hiding this comment.
🤖 nit: the doc says "Pool size (number of worker resources) and worker-thread count" — could you simplify to something like "Number of worker threads (and pooled metastore clients) for parallel partition sync when hoodie.datasource.hive_sync.batching.enabled is true"? The current wording lists "pool size" and "worker-thread count" as if they're distinct, but they're the same number.
There was a problem hiding this comment.
| // SessionState is shared across workers; build it once on the first call (with | ||
| // currentDatabase already set) and attach it to each worker's thread-local on | ||
| // subsequent calls. The database is a pool-wide property and never changes | ||
| // across workers, so setting it once at construction time is sufficient. |
There was a problem hiding this comment.
🤖 Sharing one SessionState across worker threads is risky here. Hive stores the active SessionState in a ThreadLocal (SessionState.start() sets tss.set(...)), so this pattern installs the same mutable instance into N threads' ThreadLocals. The Hive Driver isn't documented as thread-safe, and Driver execution mutates session-scoped state — currentDatabase (USE), conf overrides, and the compile/execute lineage/read-entity sets — so concurrent ALTER PARTITION/TOUCH runs can interleave on those fields. Worth confirming whether any of the batched statements can issue a USE or set conf; if so, per-worker SessionState (or serializing the session-mutating bits) is the safer default even at the init cost.
Follow-up to the HiveQL partition batching introduced in the prior commit (PR apache#18984). HiveQueryDDLExecutor.dropPartitionsToTable goes through IMetaStoreClient.dropPartition (Thrift), not the Hive Driver — so it can't reuse HiveDriverPool. This change wires an IMetaStoreClientPool into HiveQueryDDLExecutor and uses it to fan drop batches across the pool's worker threads. Behavior: - batching.enabled=false (default): unchanged. dropPartitionsToTable iterates the partition list sequentially on the session metastore client, exactly as before. - batching.enabled=true: partitions are split into batches of HIVE_BATCH_SYNC_PARTITION_NUM, batches fan out across the pool's workers (one IMetaStoreClient per worker), and first-error semantics match the existing HMS-mode implementation (first failure thrown, subsequent failures logged at WARN). HoodieHiveSyncClient now builds partitionClientPool for the HIVEQL mode branches (explicit `mode=hiveql` and the legacy default-mode branch). The pool is closed in HoodieHiveSyncClient.close() before Hive.closeCurrent() so the RetryingMetaStoreClient instances release their Thrift sockets without racing the ThreadLocal Hive cleanup. IMetaStoreClientPool is brought in as part of this PR. An analogous parallelization for HMS sync mode (separately tracked) would introduce the same class; whichever lands first owns the file going forward. Tests: - New end-to-end test: testHiveQLDropPartitionsWithBatching — creates 8 partitions, drops 4 through the parallel pool path with threads=3 and batch_num=2 (so multiple drop batches race), asserts the remaining partition set matches. Configs: no new configs. Reuses hoodie.datasource.hive_sync.batching.* from apache#18984. Related: apache#18331
- Close partitionClientPool and run Hive.closeCurrent() in a finally so a throwing ddlExecutor.close() no longer skips them, leaking the pool's Thrift sockets and worker threads. - Make partitionClientPool an Option<IMetaStoreClientPool>, matching partitionDriverPool's Option<HiveDriverPool> container type. - Extract buildHiveQueryDDLExecutor(config), which builds both partition pools and the HiveQueryDDLExecutor and rolls back (closes) whichever pool(s) were already built if a later step in that sequence throws. Previously a failure in maybeBuildPartitionClientPool (after partitionDriverPool was already built) or in the executor's own constructor leaked the already-built pool, since the outer constructor catch just rethrows.
Rebasing onto master (where apache#18984 landed) surfaced three places where the DROP-side code had not absorbed feedback that apache#18984 addressed after this branch was cut. - close(): keep master's client.close()-by-identity fix, which releases the MSC that RetryingMetaStoreClient may have rebuilt, while retaining this branch's try/finally so the pool and Hive.closeCurrent() still run when ddlExecutor.close() throws. The rebase's conflict resolution would otherwise have dropped one of the two. - Constructor catch: closePartitionDriverPoolQuietly() only closed the driver pool, so a throw after maybeBuildPartitionClientPool leaked the Thrift client pool's sockets and worker threads. Renamed to closePartitionPoolsQuietly() and extended to both pools; buildHiveQueryDDLExecutor's rollback now delegates to it instead of duplicating the logic. - runDropBatches: mirror HiveDriverPool.awaitAll's abort-on-first-error. Previously every remaining drop batch was still dispatched after a failure, so a sync that was going to abort kept issuing DROPs. mayInterruptIfRunning=false matches awaitAll: in-flight batches finish rather than leaving the partition list half-dropped mid-batch.
Applies apache#18984's log-volume feedback to the DROP path. The per-partition INFO line ran inside the parallel dispatch, so N workers multiplied it across batches that can hold thousands of partitions. It also sat outside the partitionExists check, reporting a drop for partitions that were never there. Per-partition detail moves to debug; INFO now carries a per-batch dropped/total summary. Also corrects IMetaStoreClientPool's class javadoc, which still claimed the pool is built only for sync mode HMS. It is built for HIVEQL, backing DROP only, while ADD/UPDATE/TOUCH use HiveDriverPool.
9e35ec8 to
7610c51
Compare
github-actions
Bot
added
size:L
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR extends the opt-in HiveQL batching flag to cover partition DROP, fanning drop batches across a new IMetaStoreClientPool of independent Thrift clients while keeping the default sequential path unchanged. I traced the pool's borrow/return and close lifecycle, the batch dispatch and first-error/cancellation logic, and the failure-rollback paths, and didn't find a new correctness issue in this round — pool sizing avoids deadlock (threads == clients), closes are idempotent, and batches are disjoint so concurrent workers don't contend. No blocking issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review. A couple of minor naming issues worth a quick look.
cc @yihua
| return Option.empty(); | ||
| } | ||
| if (config.getBooleanOrDefault(HIVE_SYNC_USE_SPARK_CATALOG)) { | ||
| // The Spark catalog client is constructed via reflection against a Spark-side |
There was a problem hiding this comment.
🤖 nit: the config key names in this warning (hive_sync.batching.enabled, use_spark_catalog) don't match the actual property keys an operator would search for in logs. Could you use the real constant's .key() value, e.g. HIVE_SYNC_BATCHING_ENABLED.key() + "=true is not supported with " + HIVE_SYNC_USE_SPARK_CATALOG.key() + "=true"?
| } | ||
|
|
||
| private void applyDropBatch(IMetaStoreClient poolClient, String tableName, List<String> batch) throws Exception { | ||
| int dropped = 0; |
There was a problem hiding this comment.
🤖 nit: the parameter is named poolClient but the sequential code path calls this same method with the plain session metaStoreClient — not a pool client. Something like client would read correctly in both call sites.
| // SessionState is shared across workers; build it once on the first call (with | ||
| // currentDatabase already set) and attach it to each worker's thread-local on | ||
| // subsequent calls. The database is a pool-wide property and never changes | ||
| // across workers, so setting it once at construction time is sufficient. |
There was a problem hiding this comment.
🤖 In the current HiveDriverPool design each worker actually gets its own SessionState rather than sharing one: DefaultDriverFactory.newDriver() does new SessionState(new HiveConf(hiveConf), ...) + SessionState.start(...) on the worker's own single-thread executor, and every statement for that slot runs pinned to that one thread (the Driver/SessionState are never touched from another thread). Each worker also gets its own HiveConf copy specifically so per-query keys like HIVEQUERYID don't collide. So there's no shared SessionState across threads here, and the concurrent USE/conf-mutation concern shouldn't apply — the per-worker isolation you suggested is what the code does.
danny0405
left a comment
danny0405
left a comment
There was a problem hiding this comment.
Reviewed the parallel DROP dispatch and its configuration contract. I left two inline comments that should be addressed before relying on the advertised failure semantics.
| continue; | ||
| } | ||
| try { | ||
| f.get(); |
There was a problem hiding this comment.
[P2] Stop queued DROP batches as soon as a worker fails
This waits on futures in submission order. If a later batch fails quickly while an earlier batch is slow, this thread remains blocked on the earlier f.get(), and the shared executor can keep starting every queued batch because the tasks do not observe an abort flag. By the time firstError is set, most or all "not-yet-started" DROPs may already have run, so the advertised abort-on-first-error behavior is not achieved. Please mirror HiveDriverPool.Dispatch (shared task-side abort plus completion-order notification), or use a completion service/abort flag so pending batches are stopped immediately after any worker fails.
| return Option.empty(); | ||
| } | ||
| int size = config.getIntOrDefault(HIVE_SYNC_BATCHING_THREADS); | ||
| return Option.of(new IMetaStoreClientPool(config, size)); |
There was a problem hiding this comment.
[P2] Update the public batching config contract for DROP
HIVE_SYNC_BATCHING_ENABLED in HiveSyncConfigHolder still documents that "DROP remains serial," but this pool makes DROP parallel under that exact flag. That text is the user-facing configuration contract, so it currently tells operators the opposite of the new behavior. Please update it to cover the Thrift-client DROP fan-out (including the Spark-catalog fallback) as part of this change.
5 participants
Describe the issue this Pull Request addresses
Closes #18331 (continued).
The HiveQL sync mode's
dropPartitionsToTablewalks the partition list one element at a time on the sessionIMetaStoreClient(Thrift). On tables with thousands of partitions, a single sync that needs to drop a non-trivial subset becomes a multi-minute serial Thrift loop.The companion PR #18984 introduced parallel ADD/UPDATE/TOUCH for HiveQL via
HiveDriverPool. DROP can't reuse that pool because it goes throughIMetaStoreClient.dropPartition, not the HiveDriver. This PR adds the analogous Thrift-side fan-out so the opt-in batching flag covers all four partition operations end-to-end.This PR is built on top of #18984. Once #18984 merges to master, this PR rebases trivially.
Summary and Changelog
When
hoodie.datasource.hive_sync.batching.enabled=trueand sync mode is HIVEQL, partition drops are split into batches ofhoodie.datasource.hive_sync.batch_numand dispatched across a pool of worker threads, each holding its ownIMetaStoreClient. Default off; existing behavior is unchanged.Changes:
HiveQueryDDLExecutor.dropPartitionsToTable— previously iterated the partition list sequentially on the session metastore client. Now splits into batches ofHIVE_BATCH_SYNC_PARTITION_NUMand either runs them sequentially on the session client (default) or fans them across the pool (when batching is enabled). First-error semantics: first failure thrown, subsequent failures logged at WARN. Each worker iterates its chunk one partition at a time — Hive has no batch-drop primitive matchingdropPartition's semantics, so the win comes from running chunks across independent Thrift clients in parallel.HoodieHiveSyncClient— builds aIMetaStoreClientPoolfor the HIVEQL mode branches (mode=hiveqland the legacy default-mode branch) when batching is enabled. The pool is closed inclose()beforeHive.closeCurrent()so theRetryingMetaStoreClientinstances release their Thrift sockets without racing theThreadLocalHive cleanup. Falls back to single-client sequential drop whenhoodie.datasource.hive_sync.use_spark_catalog=true(the Spark catalog client is constructed reflectively and isn't compatible with the directRetryingMetaStoreClientpool path).util/IMetaStoreClientPool— new, ~207 lines. Bounded blocking pool ofRetryingMetaStoreClientinstances backed by a fixed-sizeExecutorService. Workers borrow a client for the duration of the function (pool.run(Function)), and the pool releases it back on the slot's free list on return. UnlikeHiveDriverPool(which is constrained by Hive 2.x's thread-boundDriver/SessionState),IMetaStoreClientis a Thrift socket — the pool is a plainArrayBlockingQueue+ExecutorService, no per-slot dedicated thread required.Design constraint note: DROP and ADD/UPDATE/TOUCH live on opposite concurrency models in HiveQL mode. ADD/UPDATE/TOUCH go through the Hive
Driver(thread-bound,SessionStateisThreadLocal) and useHiveDriverPool. DROP goes through the Thrift client and usesIMetaStoreClientPool. The two pools are constructed only whenbatching.enabled=trueand torn down together inHoodieHiveSyncClient.close().Impact
User-facing impact is opt-in only.
hoodie.datasource.hive_sync.batching.enabled=false— DROP runs exactly as it does today (single client, sequential).hoodie.datasource.hive_sync.batching.enabled,hoodie.datasource.hive_sync.batching.threads, andhoodie.datasource.hive_sync.batch_numintroduced by feat(hive-sync): batch and parallelize HiveQL partition operations #18984.Risk Level
Medium.
Mitigation:
testHiveQLDropPartitionsWithBatchingcreates 8 partitions and drops 4 through the parallel pool withthreads=3andbatch_num=2(so multiple drop batches race against each other on independent Thrift clients) and asserts the remaining partition set.batching.enabled=false, the newrunDropBatcheshelper falls through to a sequentialapplyDropBatchagainst the session client — semantically equivalent to the prior loop.Documentation Update
No new configs; no docsite change required.
Contributor's checklist