feat(mito2): publish table compaction results conditionally (#10310)

* feat(mito2): publish table compaction results conditionally

Publish one region from a live input snapshot using an ordinary file-only edit. Permit repeated submissions and partial success without manifest task state, while preserving sequence, GC protection and uncertain-write recovery.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* refactor(mito2): write table compaction outputs to level 0

Route split outputs into the ordinary L0 pipeline so window merges and semantic folding stay with region compaction. Update publication validation and tests accordingly.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): release verified footers before manifest publication

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* refactor(mito2): compare typed table compaction options

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* test(mito2): cover table compaction output validation failures

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* docs(mito2): clarify series count verification limits

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): reject index declarations without output index objects

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* docs(mito2): document table compaction outputs as L0

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* docs(mito2): map the table compaction publication module

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

licenserc-enterprise.toml

@@ -26,6 +26,10 @@ includes = [
     "src/sql/src/parsers/create_parser/trigger.rs",
     "src/sql/src/parsers/show_parser/trigger.rs",
     "src/mito2/src/extension.rs",
+    "src/mito2/src/engine/table_compaction.rs",
+    "src/mito2/src/engine/table_compaction/output.rs",
+    "src/mito2/src/engine/table_compaction/tests.rs",
+    "src/mito2/src/worker/handle_table_compaction.rs",
     "src/mito2/src/compaction/multi_region.rs",
     "src/mito2/src/compaction/multi_region/source.rs",
     "src/mito2/src/compaction/multi_region/memory.rs",

licenserc.toml

@@ -42,6 +42,10 @@ excludes = [
     "src/sql/src/parsers/create_parser/trigger.rs",
     "src/sql/src/parsers/show_parser/trigger.rs",
     "src/mito2/src/extension.rs",
+    "src/mito2/src/engine/table_compaction.rs",
+    "src/mito2/src/engine/table_compaction/output.rs",
+    "src/mito2/src/engine/table_compaction/tests.rs",
+    "src/mito2/src/worker/handle_table_compaction.rs",
     "src/mito2/src/compaction/multi_region.rs",
     "src/mito2/src/compaction/multi_region/source.rs",
     "src/mito2/src/compaction/multi_region/memory.rs",

src/mito2/AGENTS.md

@@ -25,6 +25,7 @@ snapshot isolation). It implements the `RegionEngine` trait from `store-api`.
 | `flush` | `src/mito2/src/flush.rs` | `FlushScheduler`, `WriteBufferManager`, memtable → SST |
 | `compaction` | `src/mito2/src/compaction/` | Compaction scheduler (`scheduler.rs` + `scheduler/`), TWCS picker, strict-window manual picker, compactor, memory control |
 | `compaction/multi_region` | `src/mito2/src/compaction/multi_region.rs` | Enterprise lossless routing by a captured table partition rule; selected SSTs to per-region outputs, without manifest publication |
+| `engine/table_compaction` | `src/mito2/src/engine/table_compaction.rs` | Enterprise conditional publication of table-compaction results: shared-input snapshot, verified-output checks, and the manifest-locked swap of shared inputs for L0 outputs |
 | `access_layer` | `src/mito2/src/access_layer.rs` | SST read/write over the object store |
 | `sst` | `src/mito2/src/sst/` | Parquet format, file metadata, index layout |
 | `series_index` | `src/mito2/src/series_index/` | Series index writer/searcher, bucket planning, catalogs, worker-owned reconciliation (`maintenance.rs`), shared approximate usage (`task.rs`), catalogs, and index lifecycle |

src/mito2/src/compaction/multi_region.rs

@@ -102,10 +102,10 @@ pub struct MultiRegionCompactionStats {
 
 /// Complete per-target outputs, including explicit empty partitions.
 pub struct MultiRegionCompactionOutput {
-    /// Output files, initially assigned to level 1 (L1).
+    /// Output files, already assigned to region-local L0.
     ///
-    /// The caller owns publication and must set [FileMeta::level] before
-    /// publishing if its intended output level differs from L1.
+    /// Window merging and semantic folding stay with ordinary region compaction,
+    /// which runs after publication; the caller does not adjust the level.
     pub files: BTreeMap<RegionId, Vec<FileMeta>>,
     /// Complete footer/index statistics for target hook delivery; not fabricated from FileMeta.
     pub sst_infos: BTreeMap<RegionId, Vec<SstInfo>>,

src/mito2/src/compaction/multi_region/tests.rs

@@ -543,7 +543,7 @@ async fn test_single_scan_routes_all_versions_without_folding(
             }
             for file in files {
                 assert!(protected.files.lock().unwrap().contains(&file.file_id()));
-                assert_eq!(1, file.level);
+                assert_eq!(0, file.level);
                 assert_eq!(id, file.region_id);
                 assert_eq!(id != fixture.metadata.region_id, file.preserve_row_sequence);
                 let rows = fixture
@@ -1665,7 +1665,7 @@ async fn test_streaming_merge_preserves_overlapping_versions_and_field_routing(
         for (file, info) in files.into_iter().zip(&output.sst_infos[&id]) {
             assert!(protection.files.lock().unwrap().contains(&file.file_id()));
             assert_eq!(id, file.region_id);
-            assert_eq!(1, file.level);
+            assert_eq!(0, file.level);
             assert_eq!(info.file_id, file.file_id);
             assert_eq!(info.num_rows as u64, file.num_rows);
             assert_eq!(info.file_size, file.file_size);

src/mito2/src/compaction/multi_region/writer.rs

@@ -183,7 +183,9 @@ impl<'a> TargetWriter<'a> {
             region_id: self.target.metadata.region_id,
             file_id: info.file_id,
             time_range: info.time_range,
-            level: 1,
+            // Region-local L0: window merges and semantic folding stay with
+            // ordinary region compaction.
+            level: 0,
             file_size: info.file_size,
             max_row_group_uncompressed_size: info.max_row_group_uncompressed_size,
             available_indexes: info.index_metadata.build_available_indexes(),

src/mito2/src/engine.rs

@@ -87,6 +87,8 @@ mod skip_wal_test;
 mod staging_test;
 #[cfg(test)]
 mod sync_test;
+#[cfg(feature = "enterprise")]
+pub mod table_compaction;
 #[cfg(test)]
 mod truncate_test;
 

src/mito2/src/engine/region_hook.rs

@@ -157,6 +157,7 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use parquet::file::metadata::ParquetMetaData;
 use store_api::ManifestVersion;
 use store_api::metadata::RegionMetadataRef;
 use store_api::storage::RegionId;
@@ -269,6 +270,13 @@ pub struct SstFileInfo<'a> {
     pub file_meta: &'a FileMeta,
 }
 
+/// Externally produced SST with its verified footer. Index-builder statistics
+/// are deliberately absent: they cannot be reconstructed from FileMeta.
+pub struct VerifiedSstFileInfo<'a> {
+    pub file_meta: &'a FileMeta,
+    pub footer: &'a Arc<ParquetMetaData>,
+}
+
 /// Hook for observing region mutations in mito2.
 ///
 /// Implementations can be registered via the `Plugins` system:
@@ -315,6 +323,21 @@ pub trait RegionHook: Send + Sync + Debug {
         Ok(HashSet::new())
     }
 
+    /// Called for verified table-compaction outputs before manifest publication.
+    /// Unlike `on_sst_files_written`, this carries actual remote footers without
+    /// fabricating unavailable index-builder statistics. Retries may repeat it;
+    /// no manifest lock is held and implementations must be idempotent. Publication
+    /// may still fail after this callback; it is not a publication acknowledgement.
+    /// Use `on_manifest_updated` to observe a successful manifest commit.
+    async fn on_sst_files_verified(
+        &self,
+        region_id: RegionId,
+        region_metadata: &RegionMetadataRef,
+        files: &[VerifiedSstFileInfo<'_>],
+    ) {
+        let _ = (region_id, region_metadata, files);
+    }
+
     /// Called after the region manifest is successfully committed to the **live**
     /// (normal) manifest directory.
     ///

src/mito2/src/engine/table_compaction.rs

@@ -0,0 +1,131 @@
+// Copyright 2023-2026 GrepTime Inc.
+//
+// This file is part of the GreptimeDB Enterprise Edition and is licensed under
+// the GreptimeDB Enterprise License. You may not use this file except in
+// compliance with that license. A copy of the license is available at the root
+// of this repository in the file LICENSE-ENTERPRISE.
+//
+// Unless required by applicable law or agreed to in writing, this software is
+// distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+// either express or implied.
+
+//! One-shot publication of already-admitted shared inputs. Task tracking is external.
+
+pub(crate) mod output;
+#[cfg(test)]
+mod tests;
+
+use serde::{Deserialize, Serialize};
+use snafu::{OptionExt, ResultExt, ensure};
+use store_api::metadata::RegionMetadataRef;
+use store_api::storage::RegionId;
+use tokio::sync::oneshot;
+
+use crate::engine::MitoEngine;
+use crate::error::{
+    Error, InvalidRequestSnafu, RecvSnafu, RegionBusySnafu, RegionNotFoundSnafu, Result,
+    SerdeJsonSnafu,
+};
+use crate::manifest::action::RegionEdit;
+use crate::region::options::RegionOptions;
+use crate::region::{RegionLeaderState, RegionRoleState};
+use crate::request::{RegionEditKind, RegionEditRequest, WorkerRequest};
+use crate::sst::file::{FileMeta, SHARED_LEVEL_SLOT};
+
+/// A query-visible input snapshot. Callers may select a subset of `inputs` for a job.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TableCompactionSnapshot {
+    pub metadata: RegionMetadataRef,
+    pub options: serde_json::Value,
+    pub inputs: Vec<FileMeta>,
+}
+
+/// Transient publication preconditions, never stored in a manifest.
+#[derive(Debug)]
+pub(crate) struct PublicationContext {
+    pub metadata: RegionMetadataRef,
+    pub options: RegionOptions,
+}
+
+/// A result for this publication attempt, not a durable task receipt.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PublicationOutcome {
+    Applied,
+    /// The captured inputs/schema/options no longer match. Re-snapshot and replan.
+    StaleInputs,
+}
+
+impl MitoEngine {
+    pub fn table_compaction_snapshot(
+        &self,
+        region_id: RegionId,
+    ) -> Result<TableCompactionSnapshot> {
+        let region = self
+            .find_region(region_id)
+            .context(RegionNotFoundSnafu { region_id })?;
+        ensure!(
+            region.state() == RegionRoleState::Leader(RegionLeaderState::Writable),
+            RegionBusySnafu { region_id }
+        );
+        let current = region.version_control.current();
+        Ok(TableCompactionSnapshot {
+            metadata: current.version.metadata.clone(),
+            options: serde_json::to_value(&current.version.options).context(SerdeJsonSnafu)?,
+            inputs: current.version.ssts.levels()[SHARED_LEVEL_SLOT]
+                .files()
+                .map(|file| file.meta_ref().clone())
+                .collect(),
+        })
+    }
+
+    /// Publishes one region independently, using compaction rather than import sequences.
+    ///
+    /// Duplicate submissions are allowed. A stale outcome does NOT prove these
+    /// outputs were never published (an earlier ACK may have been lost). Never
+    /// blindly delete outputs on stale/error: keep their protection until recovery
+    /// or manifest-aware GC determines liveness. No input registration is performed.
+    /// Input admission must be complete/quiescent; a compaction retry must not replay
+    /// imports. The caller must produce a logically equivalent replacement under this snapshot.
+    pub async fn publish_table_compaction(
+        &self,
+        region_id: RegionId,
+        snapshot: TableCompactionSnapshot,
+        outputs: Vec<FileMeta>,
+    ) -> Result<PublicationOutcome> {
+        ensure!(
+            snapshot.metadata.region_id == region_id && !snapshot.inputs.is_empty(),
+            InvalidRequestSnafu {
+                region_id,
+                reason: "Publication requires this region's nonempty input snapshot"
+            }
+        );
+        // The public snapshot keeps JSON options; parse them once here so the
+        // worker compares typed options and reads typed flags without reparsing.
+        let options = serde_json::from_value(snapshot.options).context(SerdeJsonSnafu)?;
+        let edit = RegionEdit {
+            files_to_remove: snapshot.inputs,
+            files_to_add: outputs,
+            timestamp_ms: Some(chrono::Utc::now().timestamp_millis()),
+            ..Default::default()
+        };
+        let (tx, rx) = oneshot::channel();
+        let mut request = RegionEditRequest::new(region_id, edit, false, tx);
+        request.kind = RegionEditKind::Compaction(Box::new(PublicationContext {
+            metadata: snapshot.metadata,
+            options,
+        }));
+        self.inner
+            .workers
+            .submit_to_worker(region_id, WorkerRequest::EditRegion(request))
+            .await?;
+        match rx.await.context(RecvSnafu)? {
+            Ok(()) => Ok(PublicationOutcome::Applied),
+            Err(Error::EditRegion { source, .. })
+                if matches!(source.as_ref(), Error::StaleCompactionExecution { .. }) =>
+            {
+                Ok(PublicationOutcome::StaleInputs)
+            }
+            Err(error) => Err(error),
+        }
+    }
+}

src/mito2/src/engine/table_compaction/output.rs

@@ -0,0 +1,207 @@
+// Copyright 2023-2026 GrepTime Inc.
+//
+// This file is part of the GreptimeDB Enterprise Edition and is licensed under
+// the GreptimeDB Enterprise License. You may not use this file except in
+// compliance with that license. A copy of the license is available at the root
+// of this repository in the file LICENSE-ENTERPRISE.
+//
+// Unless required by applicable law or agreed to in writing, this software is
+// distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+// either express or implied.
+
+use std::sync::Arc;
+
+use parquet::file::metadata::ParquetMetaData;
+use parquet::file::statistics::Statistics;
+use snafu::{OptionExt, ResultExt, ensure};
+use store_api::metadata::RegionMetadataRef;
+
+use crate::access_layer::AccessLayer;
+use crate::cache::CachedSstMeta;
+use crate::engine::region_hook::VerifiedSstFileInfo;
+use crate::error::{InvalidRequestSnafu, OpenDalSnafu, Result};
+use crate::region::MitoRegionRef;
+use crate::sst::file::FileMeta;
+use crate::sst::location;
+use crate::sst::parquet::metadata::{MetadataLoader, extract_primary_key_range};
+
+/// Verifies every output, then notifies the region hook once if one is registered.
+///
+/// Without a hook no footer is needed after its own verification, so each one is
+/// dropped before the next output is read. With a hook every footer is kept until
+/// all outputs verify, because the hook observes them in one call; footers are then
+/// released when this returns, before the caller takes the manifest lock. A failed
+/// verification notifies nothing.
+pub(crate) async fn verify_and_notify(
+    region: &MitoRegionRef,
+    metadata: &RegionMetadataRef,
+    outputs: &[FileMeta],
+) -> Result<()> {
+    let Some(hook) = region.manifest_ctx.hook() else {
+        for file in outputs {
+            verify_file(&region.access_layer, metadata, file).await?;
+        }
+        return Ok(());
+    };
+
+    let mut footers = Vec::with_capacity(outputs.len());
+    for file in outputs {
+        footers.push(verify_file(&region.access_layer, metadata, file).await?);
+    }
+    let verified = outputs
+        .iter()
+        .zip(&footers)
+        .map(|(file_meta, footer)| VerifiedSstFileInfo { file_meta, footer })
+        .collect::<Vec<_>>();
+    hook.on_sst_files_verified(region.region_id, metadata, &verified)
+        .await;
+    Ok(())
+}
+
+async fn verify_file(
+    access: &AccessLayer,
+    metadata: &RegionMetadataRef,
+    file: &FileMeta,
+) -> Result<Arc<ParquetMetaData>> {
+    let region_id = metadata.region_id;
+    let path = location::sst_file_path(access.table_dir(), file.file_id(), access.path_type());
+    let size = access
+        .object_store()
+        .stat(&path)
+        .await
+        .context(OpenDalSnafu)?
+        .content_length();
+    ensure!(
+        size == file.file_size,
+        InvalidRequestSnafu {
+            region_id,
+            reason: "Output SST size mismatch"
+        }
+    );
+    let footer = MetadataLoader::new(access.object_store().clone(), &path, size)
+        .load(&mut Default::default())
+        .await?;
+    let cached = CachedSstMeta::try_new_with_region_metadata(&path, footer, None)?;
+    ensure!(
+        cached.region_metadata() == *metadata,
+        InvalidRequestSnafu {
+            region_id,
+            reason: "Output footer schema or partition mismatch"
+        }
+    );
+    let footer = cached.parquet_metadata();
+    verify_statistics(&footer, metadata, file)?;
+    if file.index_file_size > 0 {
+        let path =
+            location::index_file_path(access.table_dir(), file.index_id(), access.path_type());
+        let index_size = access
+            .object_store()
+            .stat(&path)
+            .await
+            .context(OpenDalSnafu)?
+            .content_length();
+        ensure!(
+            index_size == file.index_file_size,
+            InvalidRequestSnafu {
+                region_id,
+                reason: "Output index size mismatch"
+            }
+        );
+    } else {
+        ensure!(
+            file.available_indexes.is_empty() && file.indexes.is_empty(),
+            InvalidRequestSnafu {
+                region_id,
+                reason: "Output advertises indexes without an index object",
+            }
+        );
+    }
+    Ok(footer)
+}
+
+fn verify_statistics(
+    footer: &ParquetMetaData,
+    metadata: &RegionMetadataRef,
+    file: &FileMeta,
+) -> Result<()> {
+    let max_uncompressed = footer
+        .row_groups()
+        .iter()
+        .map(|group| {
+            group
+                .columns()
+                .iter()
+                .map(|column| column.uncompressed_size() as u64)
+                .sum::<u64>()
+        })
+        .max()
+        .unwrap_or(0);
+    let pk_range = extract_primary_key_range(footer, metadata);
+    ensure!(
+        footer.file_metadata().num_rows() > 0
+            && footer.file_metadata().num_rows() as u64 == file.num_rows
+            && footer.num_row_groups() as u64 == file.num_row_groups
+            && max_uncompressed == file.max_row_group_uncompressed_size
+            && pk_range.as_ref().map(|range| &range.0) == file.primary_key_min.as_ref()
+            && pk_range.as_ref().map(|range| &range.1) == file.primary_key_max.as_ref()
+            // `num_series` is the writer's caller-reported estimate derived from
+            // timestamp boundaries, not a footer statistic, so it only bounds the
+            // count against rows instead of being verified against the footer.
+            && file.num_series <= file.num_rows,
+        InvalidRequestSnafu {
+            region_id: metadata.region_id,
+            reason: "Output statistics differ from its footer"
+        }
+    );
+    verify_time_range(footer, metadata, file)
+}
+
+fn verify_time_range(
+    footer: &ParquetMetaData,
+    metadata: &RegionMetadataRef,
+    file: &FileMeta,
+) -> Result<()> {
+    let region_id = metadata.region_id;
+    let column_name = &metadata.time_index_column().column_schema.name;
+    let index = footer
+        .file_metadata()
+        .schema_descr()
+        .columns()
+        .iter()
+        .position(|column| column.path().parts().len() == 1 && column.name() == column_name)
+        .context(InvalidRequestSnafu {
+            region_id,
+            reason: "Output has no timestamp column",
+        })?;
+    let mut minimum = i64::MAX;
+    let mut maximum = i64::MIN;
+    for group in footer.row_groups() {
+        let Some(Statistics::Int64(stats)) = group.column(index).statistics() else {
+            return InvalidRequestSnafu {
+                region_id,
+                reason: "Output timestamp statistics missing",
+            }
+            .fail();
+        };
+        minimum = minimum.min(*stats.min_opt().context(InvalidRequestSnafu {
+            region_id,
+            reason: "Output timestamp minimum missing",
+        })?);
+        maximum = maximum.max(*stats.max_opt().context(InvalidRequestSnafu {
+            region_id,
+            reason: "Output timestamp maximum missing",
+        })?);
+    }
+    let unit = metadata.time_index_type().unit();
+    ensure!(
+        file.time_range.0.value() == minimum
+            && file.time_range.1.value() == maximum
+            && file.time_range.0.unit() == unit
+            && file.time_range.1.unit() == unit,
+        InvalidRequestSnafu {
+            region_id,
+            reason: "Output timestamp range mismatch"
+        }
+    );
+    Ok(())
+}

src/mito2/src/request.rs

@@ -1373,10 +1373,19 @@ impl Waiters {
     }
 }
 
+/// Transient execution semantics; this is never serialized into the manifest.
+#[derive(Debug)]
+pub(crate) enum RegionEditKind {
+    Import,
+    #[cfg(feature = "enterprise")]
+    Compaction(Box<crate::engine::table_compaction::PublicationContext>),
+}
+
 /// Request to edit a region directly.
 #[derive(Debug)]
 pub(crate) struct RegionEditRequest {
     pub(crate) region_id: RegionId,
+    pub(crate) kind: RegionEditKind,
     pub(crate) edit: RegionEdit,
     /// Whether to preload SST files into the write cache.
     pub(crate) preload_sst_cache: bool,
@@ -1398,6 +1407,7 @@ impl RegionEditRequest {
             region_id,
             edit,
             preload_sst_cache,
+            kind: RegionEditKind::Import,
             waiters: Waiters::one(waiter),
             staging_partition_expr: None,
         }

src/mito2/src/worker.rs

@@ -29,6 +29,8 @@ mod handle_manifest;
 mod handle_open;
 mod handle_rebuild_index;
 mod handle_remap;
+#[cfg(feature = "enterprise")]
+mod handle_table_compaction;
 mod handle_truncate;
 mod handle_write;
 

src/mito2/src/worker/handle_manifest.rs

@@ -45,8 +45,8 @@ use crate::region::version::VersionControlRef;
 use crate::region::{MitoRegionRef, RegionLeaderState, RegionRoleState};
 use crate::request::{
     BackgroundNotify, BuildIndexRequest, DiscardUnflushedResult, OptionOutputTx,
-    RegionChangeResult, RegionEditRequest, RegionEditResult, RegionSyncRequest, TruncateResult,
-    WorkerRequest, WorkerRequestWithTime,
+    RegionChangeResult, RegionEditKind, RegionEditRequest, RegionEditResult, RegionSyncRequest,
+    TruncateResult, WorkerRequest, WorkerRequestWithTime,
 };
 use crate::sst::index::IndexBuildType;
 use crate::sst::{file, location};
@@ -99,7 +99,11 @@ impl RegionEditQueue {
     }
 
     fn dequeue(&mut self) -> Option<RegionEditRequest> {
-        fn can_merge(edit: &RegionEdit) -> bool {
+        fn can_merge(request: &RegionEditRequest) -> bool {
+            if !matches!(request.kind, RegionEditKind::Import) {
+                return false;
+            }
+            let edit = &request.edit;
             // Only the `RegionEdit`:
             // 1. contains the "raw" (file without a sequence) files to add,
             // 2. and no `committed_sequence`,
@@ -120,14 +124,11 @@ impl RegionEditQueue {
         }
 
         let mut merged = self.requests.pop_front()?;
-        if !can_merge(&merged.edit) {
+        if !can_merge(&merged) {
             return Some(merged);
         }
 
-        while let Some(request) = self
-            .requests
-            .pop_front_if(|request| can_merge(&request.edit))
-        {
+        while let Some(request) = self.requests.pop_front_if(|request| can_merge(request)) {
             merged.edit.files_to_add.extend(request.edit.files_to_add);
             merged.waiters.merge(request.waiters);
         }
@@ -365,12 +366,20 @@ impl<S: LogStore> RegionWorkerLoop<S> {
 
         let RegionEditRequest {
             region_id: _,
+            kind,
             mut edit,
             waiters,
             preload_sst_cache,
             staging_partition_expr: _,
         } = request;
-        if let Err(error) = prepare_region_edit(&region, &mut edit) {
+        #[cfg(feature = "enterprise")]
+        if matches!(kind, RegionEditKind::Compaction(_)) && region.is_staging() {
+            waiters.reply_with(|| RegionBusySnafu { region_id }.fail());
+            return RegionEditDispatch::Rejected;
+        }
+        if matches!(kind, RegionEditKind::Import)
+            && let Err(error) = prepare_region_edit(&region, &mut edit)
+        {
             let error = Arc::new(error);
             waiters.reply_with(|| Err(error.clone()).context(EditRegionSnafu { region_id }));
             return RegionEditDispatch::Rejected;
@@ -399,15 +408,24 @@ impl<S: LogStore> RegionWorkerLoop<S> {
         // Now the region is in editing state.
         // Updates manifest in background.
         common_runtime::spawn_global(async move {
-            let result = edit_region(
-                &region,
-                edit.clone(),
-                cache_manager,
-                listener,
-                is_staging,
-                preload_sst_cache,
-            )
-            .await
+            let result = match kind {
+                RegionEditKind::Import => {
+                    edit_region(
+                        &region,
+                        edit.clone(),
+                        cache_manager,
+                        listener,
+                        is_staging,
+                        preload_sst_cache,
+                    )
+                    .await
+                }
+                #[cfg(feature = "enterprise")]
+                RegionEditKind::Compaction(context) => {
+                    crate::worker::handle_table_compaction::publish(&region, &context, edit.clone())
+                        .await
+                }
+            }
             .map_err(Arc::new);
             let notify = WorkerRequest::Background {
                 region_id,

src/mito2/src/worker/handle_table_compaction.rs

@@ -0,0 +1,151 @@
+// Copyright 2023-2026 GrepTime Inc.
+//
+// This file is part of the GreptimeDB Enterprise Edition and is licensed under
+// the GreptimeDB Enterprise License. You may not use this file except in
+// compliance with that license. A copy of the license is available at the root
+// of this repository in the file LICENSE-ENTERPRISE.
+//
+// Unless required by applicable law or agreed to in writing, this software is
+// distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+// either express or implied.
+
+//! At-least-once requests, conditional file publication, no durable task state.
+
+use std::collections::HashSet;
+
+use snafu::ensure;
+
+use crate::engine::table_compaction::{PublicationContext, output};
+use crate::error::{InvalidRequestSnafu, RegionBusySnafu, Result, StaleCompactionExecutionSnafu};
+use crate::manifest::action::{RegionEdit, RegionManifest, RegionMetaAction, RegionMetaActionList};
+use crate::region::{MitoRegionRef, RegionLeaderState, RegionRoleState};
+use crate::sst::file::LEVEL_SHARED;
+
+pub(crate) async fn publish(
+    region: &MitoRegionRef,
+    context: &PublicationContext,
+    edit: RegionEdit,
+) -> Result<()> {
+    let snapshot = region.manifest_ctx.manifest().await;
+    // An old result can name outputs already compacted away. Reject stale inputs
+    // before inspecting those objects; absence of an output is not a task receipt.
+    ensure!(
+        is_current(region, context, &edit, &snapshot),
+        StaleCompactionExecutionSnafu {
+            region_id: region.region_id
+        }
+    );
+    validate_files(region, context, &edit, &snapshot)?;
+    output::verify_and_notify(region, &context.metadata, &edit.files_to_add).await?;
+
+    let mut manager = region.manifest_ctx.manifest_manager.write().await;
+    ensure!(
+        matches!(
+            region.state(),
+            RegionRoleState::Leader(RegionLeaderState::Editing | RegionLeaderState::Downgrading)
+        ) && !manager.is_stopped(),
+        RegionBusySnafu {
+            region_id: region.region_id
+        }
+    );
+    let manifest = manager.manifest();
+    ensure!(
+        is_current(region, context, &edit, &manifest),
+        StaleCompactionExecutionSnafu {
+            region_id: region.region_id
+        }
+    );
+    validate_files(region, context, &edit, &manifest)?;
+    let pending = match region
+        .manifest_ctx
+        .update_locked(
+            &mut manager,
+            RegionMetaActionList::with_action(RegionMetaAction::Edit(edit)),
+            false,
+        )
+        .await
+    {
+        Ok(pending) => pending,
+        Err(error) => {
+            if error.may_have_persisted_manifest_update() {
+                // Do not accept another publication against a potentially stale
+                // in-memory file set. Completion removes this incarnation.
+                manager.stop().await;
+            }
+            return Err(error);
+        }
+    };
+    drop(manager);
+    pending.fire().await;
+    Ok(())
+}
+
+fn is_current(
+    region: &MitoRegionRef,
+    context: &PublicationContext,
+    edit: &RegionEdit,
+    manifest: &RegionManifest,
+) -> bool {
+    let current = region.version_control.current();
+    context.metadata == manifest.metadata
+        && context.options == current.version.options
+        && !edit.files_to_remove.is_empty()
+        && edit
+            .files_to_remove
+            .iter()
+            .all(|file| manifest.files.get(&file.file_id) == Some(file))
+}
+
+fn validate_files(
+    region: &MitoRegionRef,
+    context: &PublicationContext,
+    edit: &RegionEdit,
+    manifest: &RegionManifest,
+) -> Result<()> {
+    let region_id = region.region_id;
+    let mut inputs = HashSet::new();
+    ensure!(
+        edit.files_to_remove
+            .iter()
+            .all(|file| file.level == LEVEL_SHARED
+                && file.region_id.table_id() == region_id.table_id()
+                && inputs.insert(file.file_id())),
+        InvalidRequestSnafu {
+            region_id,
+            reason: "Expected distinct, already-admitted shared inputs"
+        }
+    );
+    let trusted = context.options.preserve_row_sequence
+        && edit.files_to_remove.iter().all(|file| {
+            if file.region_id == region_id {
+                file.preserve_row_sequence
+            } else {
+                file.sequence.is_some()
+            }
+        });
+    let partition =
+        crate::region::parse_partition_expr(context.metadata.partition_expr.as_deref())?;
+    let sequence = edit
+        .files_to_remove
+        .iter()
+        .map(|file| file.sequence)
+        .collect::<Option<Vec<_>>>()
+        .and_then(|sequences| sequences.into_iter().max());
+    let mut outputs = HashSet::new();
+    ensure!(
+        edit.files_to_add
+            .iter()
+            .all(|file| file.region_id == region_id
+                && file.level == 0
+                && outputs.insert(file.file_id)
+                && !manifest.files.contains_key(&file.file_id)
+                && file.partition_expr == partition
+                && file.preserve_row_sequence == trusted
+                && file.sequence == sequence),
+        InvalidRequestSnafu {
+            region_id,
+            reason: "Invalid region-local compaction output"
+        }
+    );
+    Ok(())
+}

tests/compatibility/manifest/table_compaction/README.md

@@ -7,6 +7,18 @@ released binary. They preserve L0/L1 metadata, sequences and flush boundaries.
 V1 publishes compaction results using ordinary file edits. It adds no task IDs,
 phases, receipts, request epochs or extension-state fields to the manifest.
 Task retries may be submitted more than once; publication checks current inputs.
+The Enterprise `publish_table_compaction` API consumes a query-visible input
+snapshot and returns `Applied` or `StaleInputs`. A stale attempt must re-snapshot
+and replan, never re-import its old inputs. Even stale/error outcomes must not
+trigger blind output deletion: an earlier attempt may already have published
+those objects. Callers retain output protection until recovery or GC resolves
+liveness.
+
+Publication tests (including partial success, lost ACKs and reopened regions):
+
+```sh
+cargo nextest run -p mito2 --features enterprise -E 'test(engine::table_compaction)'
+```
 
 ```sh
 cargo nextest run -p mito2 -E 'test(manifest::tests::compatibility)'