[Enhancement] Support Lake autonomous compaction#72236
Open
meegoo wants to merge 43 commits into
Open
Conversation
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 16af3fe5a1
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a Lake “autonomous compaction” flow where BE can persist compaction outputs locally and FE can later trigger a publish by collecting those local results into TxnLogs (instead of always having BE write TxnLogs to remote storage during compaction).
Changes:
- Extend Lake compaction RPC/protos to support a new COLLECT_AND_PUBLISH mode and BE-local compaction result persistence.
- Add FE-side autonomous publish triggering logic and new FE configs/state tracking for publish triggers.
- Add BE-side CompactionResultManager + skeleton LakeCompactionManager, plus COLLECT_AND_PUBLISH handling and related unit tests.
Reviewed changes
Copilot reviewed 29 out of 29 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| gensrc/proto/lake_types.proto | Add CompactionResultPB to represent locally cached compaction results. |
| gensrc/proto/lake_service.proto | Add CompactionMode and new CompactRequest fields for autonomous flows. |
| fe/fe-core/src/test/java/com/starrocks/lake/compaction/AutonomousPublishTriggerTest.java | Add unit tests for autonomous publish trigger predicate and job type wiring. |
| fe/fe-core/src/main/java/com/starrocks/lake/compaction/PartitionStatistics.java | Persist new fields to track last autonomous publish version/time/score. |
| fe/fe-core/src/main/java/com/starrocks/lake/compaction/CompactionScheduler.java | Add autonomous publish scheduling and new publish-only task construction; redirect compaction output to local results when enabled. |
| fe/fe-core/src/main/java/com/starrocks/lake/compaction/CompactionJob.java | Add JobType to distinguish legacy vs publish-only autonomous jobs. |
| fe/fe-core/src/main/java/com/starrocks/common/Config.java | Add FE configs for enabling autonomous compaction and trigger thresholds/timeouts. |
| be/test/storage/lake/lake_compaction_manager_test.cpp | Add unit tests for LakeCompactionManager queue/counter behavior (skeleton mode). |
| be/test/storage/lake/compaction_result_manager_test.cpp | Add unit tests for local compaction result persistence/merge/delete behaviors. |
| be/test/CMakeLists.txt | Register new lake compaction unit test sources. |
| be/src/storage/lake/vertical_compaction_task.cpp | Persist compaction results locally when write_to_local_result is set. |
| be/src/storage/lake/transactions.cpp | Notify autonomous manager on publish_version completion. |
| be/src/storage/lake/lake_compaction_manager.h | Introduce LakeCompactionManager API and internal bookkeeping. |
| be/src/storage/lake/lake_compaction_manager.cpp | Implement skeleton autonomous manager (queue + counters + dispatch loop placeholder). |
| be/src/storage/lake/horizontal_compaction_task.cpp | Persist compaction results locally when write_to_local_result is set. |
| be/src/storage/lake/compaction_task_context.h | Add autonomous-compaction fields to task context (local result persistence plumbing). |
| be/src/storage/lake/compaction_scheduler.cpp | Wire write_to_local_result into task contexts and notify LakeCompactionManager on completion. |
| be/src/storage/lake/compaction_result_manager.h | Define CompactionResultManager interface and helper functions. |
| be/src/storage/lake/compaction_result_manager.cpp | Implement local result file scanning, append/load/delete, and merge-to-TxnLog helpers. |
| be/src/storage/lake/compaction_policy.h | Add pick_rowsets_with_limit() API for autonomous compaction selection constraints. |
| be/src/storage/lake/compaction_policy.cpp | Provide default pick_rowsets_with_limit() implementation. |
| be/src/storage/CMakeLists.txt | Include new BE lake compaction sources in build. |
| be/src/service/service_be/lake_service.h | Declare compact_collect_and_publish() handler. |
| be/src/service/service_be/lake_service.cpp | Implement COLLECT_AND_PUBLISH: load local results, write merged TxnLogs, delete local results. |
| be/src/runtime/exec_env.h | Add ExecEnv accessor/storage for CompactionResultManager. |
| be/src/runtime/exec_env.cpp | Initialize CompactionResultManager, scan-on-startup, start/stop LakeCompactionManager. |
| be/src/common/config_lake_fwd.h | Add BE configs for autonomous compaction limits/thresholds and local result cap. |
| be/src/common/config_fwd_headers_manifest.json | Register new BE config keys in manifest. |
| be/src/common/config.h | Add BE config definitions for autonomous compaction. |
meegoo
force-pushed
the
lake-autonomous-compaction
branch
3 times, most recently
from
16b9451 to
92a154b
Compare
Contributor
🌎 Translation Required?Thanks for your doc contribution! The following languages are missing or outdated for your changes. 🤖 Automated TranslationsIf you are fluent in any of the missing languages you can add them. If not, a maintainer can generate the translations below. Maintainer: Check the ones you wish to generate:
|
github-actions
Bot
added
the
docs-maintainer
meegoo
force-pushed
the
lake-autonomous-compaction
branch
from
3d27f87 to
91bb383
Compare
Adds the foundation for BE-driven autonomous lake compaction with FE-side periodic publish triggers. BE: - gensrc proto: CompactRequest.mode (EXECUTE / COLLECT_AND_PUBLISH), visible_version, new_version, write_to_local_result; new CompactionResultPB embedding OpCompaction. - CompactionResultManager: per-store-path local persistence, scan_on_startup index rebuild, pending_inputs tracking, capacity cap, helper persist_compaction_result_from_txn_log, helper merge_results_to_txn_log (always wraps as OpParallelCompaction so apply path takes the existing graceful-skip branch). - LakeService::compact: routes COLLECT_AND_PUBLISH to a new compact_collect_and_publish that drains local results into one OpParallelCompaction TxnLog per tablet; FE drives the actual publish via the existing partial-success force_publish path. - compaction_scheduler.cpp: plumbs write_to_local_result and result_manager into CompactionTaskContext; horizontal/vertical task implementations switch output path based on the flag. - LakeCompactionManager skeleton: priority queue, dedup, per-tablet and global concurrency caps, bvar metrics. Live task submission is left as a follow-up and clearly TODO'd. - pick_rowsets_with_limit(excluded, max_bytes) on CompactionPolicy with a default post-filter implementation. - Module-boundary clean: result manager handle is plumbed through the context / LakeCompactionManager singleton, so lake task files do not need ExecEnv::GetInstance. FE: - CompactionJob.JobType (COMPACT_AND_PUBLISH / PUBLISH_ONLY) + PartitionStatistics.lastPublishVisibleVersion / TimeMs / Score. - CompactionScheduler: new schedulePartitionPublish() with three trigger strategies (version_delta / high_score+min_delta / max_interval), createPublishOnlyTasks builds CompactRequest with mode=COLLECT_AND_PUBLISH; PUBLISH_ONLY jobs always forceCommit=true so tablets without local results take the ignore_txn_log + observe_empty_compaction path on the BE. - Aggregate-publish (file_bundling) is mutually exclusive with autonomous; warning logged and autonomous publish skipped when both are enabled. Configs (BE & FE) all default OFF (enable_lake_autonomous_compaction). UTs: - compaction_result_manager_test: append/load/delete/scan/corrupt/ capacity/persist-helper/merge-helper. - lake_compaction_manager_test: dedup/disabled/notify counters. - AutonomousPublishTriggerTest (FE): three trigger predicates + JobType plumbing. Phase 2.3 / 4.3 fixture-heavy integration UTs are listed as TODOs in the test files for follow-up PRs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <meegoo.sr@gmail.com>
- CompactionScheduler.java:125 long line broken into 3 strings. - AutonomousPublishTriggerTest.java: convert if-without-braces to braced. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <meegoo.sr@gmail.com>
Use FileSystemFactory::CreateSharedFromString (per fs_util.h pattern) instead of the non-existent FileSystem::CreateSharedFromString. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <meegoo.sr@gmail.com>
…E_TEST init The dispatch thread was started unconditionally in ExecEnv::init() under BE_TEST. Even though dispatch_loop only does cv.wait_for(1s), the thread launches were apparently interfering with other test fixtures and causing init to hang past result_buffer_mgr setup. Tests that exercise the manager (lake_compaction_manager_test) call start() explicitly in their SetUp; production paths still start it from the USE_STAROS branch. Plain test suites that never touch autonomous compaction now no longer pay for a background dispatch thread. scan_on_startup() is still called so pending_inputs is rebuilt before any test fixture might enable the feature. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <meegoo.sr@gmail.com>
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <meegoo.sr@gmail.com>
P2: add 5 new BE configs and 5 new FE configs introduced by the lake autonomous compaction PR to the canonical Configuration parameter pages, with bilingual coverage: BE side (be/src/common/config.h): - enable_lake_autonomous_compaction - lake_autonomous_compaction_max_concurrent_tasks - lake_autonomous_compaction_max_tasks_per_tablet - lake_autonomous_compaction_score_threshold - lake_autonomous_compaction_local_result_dir_max_bytes FE side (fe-core Config.java): - enable_lake_autonomous_compaction - lake_compaction_version_delta_threshold - lake_compaction_high_score_threshold - lake_compaction_min_version_delta_for_high_score - lake_compaction_max_interval_ms - lake_compaction_publish_timeout_seconds Added to shared_lake_other.md (the existing lake-compaction parameter section) so they sit next to lake_compaction_allow_partial_success and lake_pk_compaction_max_input_rowsets. Signed-off-by: meegoo <meegoo.sr@gmail.com>
…score
P1: replace try_dispatch_one_locked's skeleton (which only logged and
eagerly released the slot) with a real submission to the existing
CompactionScheduler:
- Synthesize a CompactRequest with write_to_local_result=true and a
negative txn_id from a per-manager atomic counter (kept negative so
they can never collide with FE-allocated positive ids).
- Resolve base_version under the lock from
TabletManager::get_latest_cached_tablet_metadata(); cold tablets
(no cached metadata) are skipped rather than dispatched with version=0.
- Drop _mu around the actual scheduler->compact() call so worker
threads completing in parallel can re-enter notify_task_finished.
- Use a self-deleting AutonomousDispatchClosure that owns the heap
request/response and runs once finish_task fires the existing
notify_task_finished + update_tablet_async wiring (already in
compaction_scheduler.cpp). This means slot release happens on real
completion, not at dispatch time.
compute_score_locked: replace the constant-above-threshold stub with
the latest cached metadata's rowsets_size(). Returning 0 for cold
tablets deliberately suppresses dispatch until publish_version warms
their metadata.
g_autonomous_completed_tasks moved into notify_task_finished so the
counter only ticks on real completion (the skeleton previously ticked
both dispatched and completed in the same call).
Existing UTs unchanged: they call start(nullptr, nullptr); compute_score
short-circuits to 0 with a null tablet_mgr so update_tablet_async is a
no-op below threshold and the tests still verify lifecycle/idempotency
contracts (queue dedup, notify-without-reservation safety, singleton).
Signed-off-by: meegoo <meegoo.sr@gmail.com>
Adds 4 SQL-Tester cases that exercise the COLLECT_AND_PUBLISH path
end-to-end against a real shared-data cluster, raising coverage on:
- lake_compaction_manager.cpp dispatch_loop / try_dispatch_one_locked
real path (not the test-time short-circuit)
- lake_service.cpp::compact_collect_and_publish (only reachable from
a real PUBLISH_ONLY transaction)
- compaction_scheduler.cpp autonomous integration points
(write_to_local_result branch, notify_task_finished real path)
- horizontal_compaction_task / vertical_compaction_task
write_to_local_result branches
- FE: schedulePartitionPublish, startPublishOnly, the PUBLISH_ONLY
completion handler in scheduleNewCompaction
Cases:
1. test_lake_autonomous_compaction_basic
Duplicate-key table; flips enable_lake_autonomous_compaction +
enable_file_bundling=false, lowers thresholds, runs 5 inserts and
waits ~15s for the autonomous publish to fire. Asserts row count
and ordered scan are stable; the partition's CURRENT_VERSION is
marked [UC] because it depends on how many publish rounds fit in
the wait window.
2. test_lake_autonomous_compaction_pk
Primary-key variant. Inserts overlap keys 1-3 with newer values;
asserts post-compaction PK semantics (latest row wins).
3. test_lake_autonomous_compaction_disabled
Negative case: master switch off → autonomous code path is a
complete no-op; verifies data integrity.
4. test_lake_autonomous_compaction_bundling_skip
With enable_file_bundling=true (the default) and the master
switch on, schedulePartitionPublish must skip (the aggregate-
publish RPC path is incompatible). Visible behavior is identical
to "disabled"; we assert data integrity only since the skip is
a log-only side effect.
All four cases are tagged @cloud (shared-data only) @Sequential
because they mutate global FE/BE configs. Configs are restored to
their documented defaults before DROP DATABASE so subsequent tests
in the same cluster are unaffected.
Signed-off-by: meegoo <meegoo.sr@gmail.com>
… trigger
Adds 7 unit tests to lift CompactionScheduler.java coverage from 29.31%
toward the 80% target. The previous coverage report (FE 41.84%, BE
69.64%) was bottlenecked by the autonomous-publish path being exercised
only by SQL-Tester, which runs against a deployed cluster without
JaCoCo / gcov instrumentation and therefore contributes nothing to the
incremental-coverage gate.
These cover the schedulePartitionPublish entry into startPublishOnly
and every early-return branch of startPublishOnly that does not require
a real OlapTable / TabletInvertedIndex fixture:
- testSchedulePartitionPublishDrivesThroughStartPublishOnly:
full sweep through schedulePartitionPublish with a registered
partition that fires the version-delta trigger; startPublishOnly
immediately hits the db-not-found branch and removePartition is
asserted to have run.
- testStartPublishOnlyDbNotFound: db == null → null + removePartition.
- testStartPublishOnlyWarehouseResolutionFails:
getCompactionComputeResource throws ErrorReportException
(ERR_WAREHOUSE_UNAVAILABLE) → null without taking the locker.
- testStartPublishOnlySkipsSchemaChange: table.getState() ==
SCHEMA_CHANGE → null. Exercises the cross-version-apply guard
that prevents alter_metadata races.
- testStartPublishOnlyPartitionGone: physical partition dropped
concurrently → null + removePartition.
- testStartPublishOnlyEmptyBeToTablets: collectPartitionTablets
returns empty (no shards assigned) → null without beginTransaction.
- testStartPublishOnlySuccess: full happy path; mocks
collectPartitionTablets, beginTransaction, BrpcProxy.getLakeService
so the test stops at sendRequest. Asserts the returned CompactionJob
has JobType=PUBLISH_ONLY and the expected txn_id.
All seven mock GlobalStateMgr/LocalMetastore via JMockit MockUp blocks
following the testStartCompaction/testStartCompactionWithFileBundling
pattern already in this file. Verified locally: 79/79 lake.compaction
package tests green (was 72/72).
Signed-off-by: meegoo <meegoo.sr@gmail.com>
… paths Push FE coverage past 80%. The previous run (806f497) hit 79.43% (112/141), 0.57 percentage points short. The two newly-uncovered lines in schedulePartitionPublish are easy targets: - testSchedulePartitionPublishSkipsDisabledTable: Constructs the scheduler with disableIdsStr containing the partition's tableId so disabledIds.contains(...) is true at line 177; verifies the partition is skipped without calling startPublishOnly. - testSchedulePartitionPublishSkipsWhenNoTrigger: Sets thresholds high enough that none of the trigger strategies match (versionDelta=1 < threshold=1000, score below high-score cutoff, lastPublishTimeMs=0 disables max-interval). evaluatePublishTrigger returns null and line 189's "continue" fires. Both verified locally. Together they should close the gap to 80%+. Signed-off-by: meegoo <meegoo.sr@gmail.com>
…enqueue Lift BE coverage past 80%. Previous coverage report (e02d583) flagged lake_service.cpp at 1.96% (1/51 lines on compact_collect_and_publish) and lake_compaction_manager.cpp at 49.65% — both because the existing tests short-circuit before the dispatch path. These tests drive the real LakeServiceImpl::compact_collect_and_publish RPC handler with the live ExecEnv fixture (which has CompactionResultManager initialized by BE_TEST init), and exercise the update_tablet_async enqueue branch by lowering score_threshold to 0. LakeServiceTest additions (6 cases): - test_compact_collect_and_publish_missing_visible_version (validation) - test_compact_collect_and_publish_missing_new_version (validation) - test_compact_collect_and_publish_missing_txn_id (validation) - test_compact_collect_and_publish_no_local_results (empty load, status_code = 0 → covers the line that fixed the FE-side null Future bug) - test_compact_collect_and_publish_with_local_results (full load → merge_results_to_txn_log → put_txn_log → delete_results; asserts the seeded result is gone afterward) - test_compact_collect_and_publish_via_compact_dispatch (verifies the mode-dispatch in compact() at line 1421 routes COLLECT_AND_PUBLISH straight through) LakeCompactionManagerTest additions (3 cases): - update_enqueues_when_threshold_is_zero (with threshold=0 even score=0 passes the gate, exercising lines 75-84 push + cv.notify_one; dispatch loop then takes the null-tablet_mgr skip branch at 167-177) - update_dedupes_via_enqueued_set (50 calls for one tablet, asserts no underflow and clean counters; covers dedup branch at line 70) - notify_consumed_rowsets_without_reservation_is_safe (multiple rowsets passed without prior reservation; covers lines 110-116 defensive cleanup) Verified on remote (lake-auto worktree, ASAN build): starrocks_test LakeServiceTest.test_compact_collect_and_publish* : 6/6 PASS starrocks_dw_test LakeCompactionManagerTest.* : 12/12 PASS CompactionResultManagerTest.* : 25/25 PASS Signed-off-by: meegoo <meegoo.sr@gmail.com>
Three new error-path tests that close the remaining uncovered lines in
compaction_result_manager.cpp (lines 127, 129-131, 249-251):
- scan_renames_corrupt_file_with_missing_required_fields:
Write a CompactionResultPB that parses cleanly but omits the
required tablet_id/base_version/op_compaction fields. scan_on_startup
-> load_one_file's manual required-fields check (lines 129-131)
returns Status::Corruption; scan_on_startup renames the .pb with a
.corrupt suffix so a future scan does not re-attempt parsing.
- scan_renames_garbage_pb_payload:
A .pb file containing pure 0xFF bytes makes ParseFromArray return
false outright (line 127), triggering the same rename-with-.corrupt
treatment.
- load_results_returns_corruption_on_mid_flight_garbage:
After a successful append_result, overwrite the on-disk file with
garbage. load_results re-reads from disk (no parsed-PB cache) and
must surface Status::Corruption from line 250, not silently swallow it.
Signed-off-by: meegoo <meegoo.sr@gmail.com>
Rebase conflict resolution accidentally kept this include from the old
HEAD of the lake-autonomous-compaction branch. Upstream/main removed it
as part of unrelated cleanup, and the BE Architecture module-boundary
check now forbids non-service production code from including top-level
Service headers (see be/AGENTS.md and module_boundary_manifest.json).
Verified locally: python3 build-support/check_be_module_boundaries.py
--mode changed --base upstream/main now passes ("OK: BE module
boundaries clean for common."). The symbol is not referenced anywhere
in exec_env.cpp, so removing the include is purely dead-weight cleanup.
Signed-off-by: meegoo <meegoo.sr@gmail.com>
…us compaction files build-support/check_common_config_header_includes.sh forbids new C++ files from including the monolithic common/config.h directly; new files must use the appropriate forward-header (config_<domain>_fwd.h). All the config symbols referenced by the autonomous-compaction code paths (enable_lake_autonomous_compaction, lake_autonomous_compaction_*) are already exported from common/config_lake_fwd.h, so switching the includes is a drop-in replacement. Files updated: - be/src/storage/lake/lake_compaction_manager.cpp - be/src/storage/lake/compaction_result_manager.cpp - be/test/storage/lake/lake_compaction_manager_test.cpp - be/test/storage/lake/compaction_result_manager_test.cpp Verified locally: check_common_config_header_includes.sh now passes with "OK: all direct C++ includes of common/config.h are within allowlist (2 files)." Signed-off-by: meegoo <meegoo.sr@gmail.com>
Adds 7 new SQL-Tester cases on top of the existing 4, covering scenarios
the unit tests cannot reach (real TabletManager + real publish_version
flow against a deployed cluster):
- test_lake_autonomous_compaction_agg
AGGREGATE-key table. Multiple inserts for same keys must collapse
via SUM/MAX aggregation when autonomous compaction merges rowsets.
- test_lake_autonomous_compaction_unique
UNIQUE-key table. Duplicate keys must collapse to last-write-wins.
- test_lake_autonomous_compaction_pk_delete
PK + DELETE. Tombstones must be honored across autonomous compaction +
force_publish; deleted keys must not reappear.
- test_lake_autonomous_compaction_pk_update
PK + UPDATE. Chained UPDATEs (including IN-list multi-key) must take
effect after autonomous compaction merges the per-update rowsets.
- test_lake_autonomous_compaction_multi_bucket
Multi-bucket (4) partition with 200 rows. Exercises the per-BE
grouping in CompactionScheduler.collectPartitionTablets/
createPublishOnlyTasks and verifies version advances uniformly
across tablets even when some have no local results.
- test_lake_autonomous_compaction_disable_table
lake_compaction_disable_ids: tableId listed → schedulePartitionPublish
must skip via the disabledIds branch. Data integrity unchanged.
- test_lake_autonomous_compaction_high_score_trigger
Tunes lake_compaction_high_score_threshold=0.5 +
lake_compaction_min_version_delta_for_high_score=2 while keeping
version_delta_threshold high. Exercises the second leg of
evaluatePublishTrigger (high-score path) that the version-delta
path tests would never hit.
- test_lake_autonomous_compaction_many_rowsets
15 separate INSERTs into one tablet; verifies merge_results_to_txn_log
handles a large OpParallelCompaction with many subtasks correctly.
Also fixes a config-restore bug in the existing _bundling_skip case:
enable_file_bundling default is true but the previous restore set it
to "false". Restore now matches the documented default.
All cases @cloud @Sequential to avoid race with global FE/BE config
mutation between concurrent tests. Each restores its config triplet
+ enable_file_bundling before DROP DATABASE.
Signed-off-by: meegoo <meegoo.sr@gmail.com>
meegoo
force-pushed
the
lake-autonomous-compaction
branch
from
c34370e to
902c1cd
Compare
…tion
Replace the cluster-wide skip in CompactionScheduler.runOneCycle()
(if Config.enable_file_bundling: skip autonomous publish entirely)
with a per-table check inside schedulePartitionPublish():
if (isPartitionTableFileBundling(partition)) continue;
Why: the prior global skip disabled autonomous publish whenever the
cluster default enable_file_bundling was true — which is the documented
default. Effectively autonomous compaction was unusable unless an
operator manually flipped the cluster default off. With the per-table
check, only the bundled tables are skipped; mixed databases (some
bundled, some not) work correctly out of the box.
Level 2 will replace the per-table skip with an aggregator-based
autonomous publish path so bundled tables also benefit. Until then,
bundled tables continue using the existing ALTER TABLE ... COMPACT
manual trigger which produces aggregate_compact + aggregate_publish_version.
Test changes:
- CompactionSchedulerTest.testSchedulePartitionPublishSkipsBundledTable
replaces testRunOneCycleSkipsAutonomousWhenFileBundlingEnabled. The
new test actually mocks a LakeTable with setFileBundling(true) and
verifies the schedulePartitionPublish loop skips it.
- SQL test_lake_autonomous_compaction_bundling_skip now creates one
bundled table + one non-bundled table in the same DB with
PROPERTIES("file_bundling"=...) and asserts both keep correct data
(the bundled table is skipped, the non-bundled one is published).
enable_file_bundling cluster config stays at the default false in
this test to validate per-table dispatch.
Verified locally: lake.compaction Maven test package runs 81/81 green
(was 79 + 2 — replaced the old skip test, added the per-table skip
test, total +2 cases vs the previous run).
Signed-off-by: meegoo <meegoo.sr@gmail.com>
Bundled tables (table.isFileBundling()) now route through an
AggregateCompactRequest-based PUBLISH_ONLY job that produces a single
CombinedTxnLogPB the existing aggregate_publish_version path on the BE
can read. Level 1 left bundled tables out of autonomous compaction
entirely; Level 2 brings them in without conflicting with the
combined_txn_log layout publish-side expects.
Flow:
FE schedulePartitionPublish -> startPublishOnly
startPublishOnly:
if table.isFileBundling():
createAggregatePublishOnlyTask ----> one AggregateCompactionTask
else:
createPublishOnlyTasks ----> per-BE CompactionTask list
Aggregator BE receives AggregateCompactRequest with sub-CompactRequest[]:
mode=COLLECT_AND_PUBLISH, skip_write_txnlog=true, visible_version,
new_version per sub
Each owning BE handles sub-request via compact() -> compact_collect_and_publish:
skip_write_txnlog=true -> response->add_txn_logs(merged TxnLog)
delete_results(consumed_result_ids)
Aggregator BE collects all txn_logs into CombinedTxnLogPB and writes
via put_combined_txn_log (anchor tablet's location).
FE commits txn (forceCommit=true -> TxnInfoPB.force_publish=true).
PublishVersionDaemon: useAggregatePublish=table.isFileBundling()=true,
so aggregate_publish_version reads the combined log + per-tablet
applies + writes bundle metadata.
BE side (lake_service.cpp::compact_collect_and_publish):
- New branch on request.skip_write_txnlog(): instead of put_txn_log(),
pack the merged TxnLog into response.txn_logs[]. Local results are
still deleted after the response is built (same failure-recovery
semantics as the per-BE flow: if the aggregator fails to write the
combined log, the autonomous compaction's output rowsets are still
on remote storage and the next autonomous round will re-compact
from them).
- Log line includes skip_write_txnlog for observability.
FE side (CompactionScheduler.java):
- schedulePartitionPublish no longer skips bundled tables (the Level 1
per-table skip is now per-route dispatch).
- startPublishOnly branches on table.isFileBundling():
- createAggregatePublishOnlyTask: mirrors the existing
createAggregateCompactionTask pattern (LakeAggregator.chooseAggregatorNode
picks aggregator from candidate owning nodes) but builds sub-requests
with mode=COLLECT_AND_PUBLISH + skip_write_txnlog=true.
- createPublishOnlyTasks: unchanged.
Test changes:
- CompactionSchedulerTest.testStartPublishOnlyRoutesBundledTableToAggregator
replaces the Level 1-era testSchedulePartitionPublishSkipsBundledTable.
Mocks GlobalStateMgr/LocalMetastore/OlapTable.getPhysicalPartition +
collectPartitionTablets/beginTransaction/BrpcProxy and asserts the
resulting CompactionJob's single task is an AggregateCompactionTask.
Risk notes:
- Aggregator BE failing mid-write loses the merged TxnLog metadata for
that cycle; compaction output rowsets survive and re-compact next
cycle. Same failure semantic as the per-BE flow's put_txn_log failure.
- Local result cleanup happens immediately on per-BE response; if
aggregator fails to write combined log, we lose the merged TxnLog
but rowsets remain and force-publish next round picks them up.
Verified locally:
- lake.compaction package: 81/81 tests green, including the new aggregator
routing test.
- BE lake_service.cpp compiles cleanly (final starrocks_test link has
upstream-induced symbol issues unrelated to this change).
Signed-off-by: meegoo <meegoo.sr@gmail.com>
…spatch failure When startPublishOnly throws after beginTransaction (sendRequest / aggregate dispatch failure), the job is never registered in runningCompactions, so the setMinRetainVersion(0) reset in scheduleNewCompaction never runs for it. The partition's version GC/vacuum stays pinned at visibleVersion indefinitely. Reset it in the catch, mirroring the legacy startCompaction failure path. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <mervin.hu@celerdata.com>
|
|
Contributor
|
No new undocumented parameters detected by the param-drift check. |
…p publish concurrency
B1 (correctness): the autonomous EXECUTE dispatch ran the plain pick_rowsets()
path, so the pending_inputs/running_inputs exclusion set (and
pick_rowsets_with_limit) were never consulted in production — only by tests.
Between an EXECUTE compaction and the FE COLLECT_AND_PUBLISH the tablet metadata
is unchanged, so finish_task self-continuation kept re-picking the SAME rowsets,
producing duplicate CompactionResultPB. Merged into one OpParallelCompaction
those duplicate input_rowsets either fail apply on non-PK tables ("input rowset
not found" -> partition publish permanently stuck) or are silently skipped on PK
tables (orphan rowsets + busy-loop).
Fix: TabletManager::compact, for the write_to_local_result path, now selects
inputs via LakeCompactionManager::pick_and_reserve_inputs. It lets the policy
pick its natural contiguous block and reserves it all-or-nothing into
running_inputs only if the block is fully disjoint from running_inputs U
pending_inputs. This guarantees (a) no rowset is compacted twice before publish
and (b) each reserved block stays adjacent in metadata (required by
apply_compaction_log_single_output). The reserved set is stored on the context
and released in finish_task on ANY outcome (success or failure), so the
reservation can never leak; self-continuation only fires when real work was
done. An empty pick is a benign no-op (do_compaction maps it to success).
B3 (stability): schedulePartitionPublish now respects the same per-warehouse
compactionTaskLimit as scheduleNewCompaction, seeded from in-flight jobs, so
enabling the feature cannot open an unbounded number of transactions in one tick.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: meegoo <mervin.hu@celerdata.com>
… gate testSchedulePartitionPublishDrivesThroughStartPublishOnly now must pass the new per-warehouse budget check before reaching startPublishOnly. Set a positive lake_compaction_max_tasks and stub getCompactionComputeResource so the gate is satisfied deterministically (independent of alive-node counts). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <mervin.hu@celerdata.com>
# Conflicts: # be/src/runtime/exec_env.cpp # be/src/runtime/exec_env.h # be/src/storage/lake/tablet_manager.cpp
Contributor
[BE Incremental Coverage Report]❌ fail : 22 / 28 (78.57%) file detail
|
… local results Coexistence fix (FE): when enable_lake_autonomous_compaction is on, the BE drives compaction and FE only publishes via schedulePartitionPublish(). scheduleNewCompaction now returns right after its completion-processing loop instead of scheduling NEW legacy FE-driven jobs. Previously legacy ran concurrently on the same tablets and could (a) compact+publish first, orphaning an autonomous pending local result whose input rowsets then no longer exist, and (b) abort with 'empty CombinedTxnLogPB' once autonomous had already drained the work. The completion loop still runs, so in-flight legacy jobs AND PUBLISH_ONLY jobs (both in runningCompactions) finish normally. Defensive GC (BE): compact_collect_and_publish now drops+deletes any local CompactionResultPB whose input rowsets are absent from the tablet's current visible metadata before merging. Such a stale result (crash/restart/external compaction leftover) can never publish — non-PK apply fails 'input rowset not found' and would wedge the partition — so we delete it rather than let it block. Guarded on a successful metadata read so a transient failure never drops a valid result. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <mervin.hu@celerdata.com>
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7a66fcd8be
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+137
to
+139
| if (versionDelta <= 0) { | ||
| return null; | ||
| } |
There was a problem hiding this comment.
When a PUBLISH_ONLY job becomes visible, the completion path records lastPublishVisibleVersion as the partition's current visible version; the BE autonomous compaction that is triggered by that publish only writes a local CompactionResultPB and does not advance FE currentVersion. On an idle partition, versionDelta therefore stays 0 forever, so this early return prevents even the max-interval trigger from draining and publishing the pending local result until another load happens.
Useful? React with 👍 / 👎.
Contributor
Author
There was a problem hiding this comment.
This concern does not apply. A PUBLISH_ONLY job is a LAKE_COMPACTION transaction, so when it becomes VISIBLE LakeTableTxnLogApplier calls handleCompactionFinished, which advances the statistics' currentVersion to the newly published version (CompactionMgr L158-163). The completion handler then sets lastPublishVisibleVersion to that same version (CompactionScheduler L432-433).
So versionDelta only returns to 0 after a successful drain — and BE cannot produce a new pending local result without new input rowsets, which arrive via loads that re-raise currentVersion (and therefore versionDelta). versionDelta == 0 is thus equivalent to "nothing pending to publish," and the max-interval safety net remains reachable exactly for the case it targets: a small positive delta below the eager thresholds. In addition, the benign no-op completion path (L399-407) intentionally leaves lastPublishVisibleVersion untouched, so a premature publish keeps versionDelta > 0 and retries. There is no idle partition that holds an undrained local result while versionDelta == 0.
Generated by Claude Code
Comment on lines
+1531
to
+1533
| if (results.empty()) { | ||
| ++without_compaction; | ||
| continue; |
There was a problem hiding this comment.
In the file-bundling path the aggregator writes a CombinedTxnLogPB containing only the txn_logs returned here; for tablets with no local result this branch returns no entry. Later load_txn_log() for a combined transaction treats a combined log that exists but lacks the tablet's log as InternalError, so force_publish cannot take the missing-log no-op path and aggregate_publish_version fails whenever a bundled partition has any tablet without a local compaction result.
Useful? React with 👍 / 👎.
Comment on lines
+178
to
+179
| if (disabledIds.contains(partition.getTableId())) { | ||
| continue; |
There was a problem hiding this comment.
disabledIds is documented and exposed as table id or partition id, and the legacy scheduler passes it through choosePartitionsToCompact() so both are excluded. This autonomous publish loop only checks the table id, so setting lake_compaction_disable_ids to a specific partition id no longer prevents PUBLISH_ONLY transactions from being opened for that partition when autonomous compaction is enabled.
Useful? React with 👍 / 👎.
Comment on lines
+256
to
+257
| Locker locker = new Locker(); | ||
| locker.lockDatabase(db.getId(), LockType.READ); |
There was a problem hiding this comment.
This new critical section only reads/mutates one known table and partition before opening the compaction transaction, but a DB-wide READ lock blocks unrelated table ALTER/DDL in the same database. That violates the fe/AGENTS.md Database and Table Locks guidance that new code should default to the intensive path for known-table work; use the same lockTableWithIntensiveDbLock(..., tableId, READ) pattern as startCompaction().
Useful? React with 👍 / 👎.
FE's version_delta publish trigger is more eager than the BE EXECUTE score threshold and can race ahead of it, so a file-bundling partition may be asked to publish before any local compaction result exists. The aggregator rejects an empty CombinedTxnLogPB, which scheduleNewCompaction logged as a job failure (ERROR spam + failure-history churn + on-failure penalty interval) and aborted the txn. This is not a real failure — there was simply nothing to publish yet. Now a PUBLISH_ONLY whose failure is 'empty CombinedTxnLogPB' is dropped quietly and retried on the next trigger once results accumulate. No data impact (versions still advance once EXECUTE produces results); this only removes the benign-abort noise. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <mervin.hu@celerdata.com>
The prior FE-scope build uploaded the OSS package keyed by commit 1b266e0, so a subsequent full build is skipped as 'already exists' and only the FE-only artifact remains. New hash -> new package -> full FE+BE build. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: meegoo <mervin.hu@celerdata.com>
Contributor
[Java-Extensions Incremental Coverage Report]✅ pass : 0 / 0 (0%) |
Contributor
[FE Incremental Coverage Report]❌ fail : 164 / 210 (78.10%) file detail
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why I'm doing:
What I'm doing:
Fixes #issue
What type of PR is this:
Does this PR entail a change in behavior?
If yes, please specify the type of change:
Checklist:
Bugfix cherry-pick branch check: