feat: add manual series index reconciliation (#9323)

* feat(mito): support manual series index reconciliation Signed-off-by: evenyag <realevenyag@gmail.com> * feat(storage): route series index build requests Signed-off-by: evenyag <realevenyag@gmail.com> * feat(admin): add BUILD_SERIES_INDEX Signed-off-by: evenyag <realevenyag@gmail.com> * test: specify compaction type in series index fixtures Signed-off-by: evenyag <realevenyag@gmail.com> * test: correct series index SQL fixtures and error assertions Signed-off-by: evenyag <realevenyag@gmail.com> * test(compat): preserve legacy index rebuild across upgrades Signed-off-by: evenyag <realevenyag@gmail.com> * test(sql): cover series index admin validation Signed-off-by: evenyag <realevenyag@gmail.com> * refactor(mito): bound series index maintenance queue Signed-off-by: evenyag <realevenyag@gmail.com> * docs: remove series index how-to guide Signed-off-by: evenyag <realevenyag@gmail.com> * test: remove index build upgrade compatibility case Signed-off-by: evenyag <realevenyag@gmail.com> * fix: address series index reconciliation review feedback Signed-off-by: evenyag <realevenyag@gmail.com> * fix: bound manual series index reconciliation admission Signed-off-by: evenyag <realevenyag@gmail.com> * chore: update greptime-proto to merged index build options Signed-off-by: evenyag <realevenyag@gmail.com> --------- Signed-off-by: evenyag <realevenyag@gmail.com>
2026-10-19 18:32:33 +00:00 · 2026-09-24 09:39:21 +00:00
parent b547a3e369
commit 3c2aac0a55
16 changed files with 1111 additions and 107 deletions
@@ -34,7 +34,7 @@ use reconcile_catalog::ReconcileCatalogFunction;
 use reconcile_database::ReconcileDatabaseFunction;
 use reconcile_table::ReconcileTableFunction;

-use crate::admin::build_index_table::BuildIndexFunction;
+use crate::admin::build_index_table::{BuildIndexFunction, BuildSeriesIndexFunction};
 use crate::admin::discard_unflushed_data::DiscardUnflushedDataFunction;
 use crate::flush_flow::FlushFlowFunction;
 use crate::function_registry::FunctionRegistry;
@@ -53,6 +53,7 @@ impl AdminFunction {
        registry.register(GcRegionsFunction::factory());
        registry.register(GcTableFunction::factory());
        registry.register(BuildIndexFunction::factory());
+        registry.register(BuildSeriesIndexFunction::factory());
        registry.register(FlushFlowFunction::factory());
        registry.register(ReconcileCatalogFunction::factory());
        registry.register(ReconcileDatabaseFunction::factory());
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use api::v1::region::build_index_request;
 use arrow::datatypes::DataType as ArrowDataType;
 use common_error::ext::BoxedError;
 use common_macro::admin_fn;
@@ -38,6 +39,45 @@ pub(crate) async fn build_index(
    table_mutation_handler: &TableMutationHandlerRef,
    query_ctx: &QueryContextRef,
    params: &[ValueRef<'_>],
+) -> Result<Value> {
+    build_index_impl(
+        table_mutation_handler,
+        query_ctx,
+        params,
+        "build_index",
+        None,
+    )
+    .await
+}
+
+/// Reconciles series indexes on every physical data-region leader and waits for publication.
+#[admin_fn(
+    name = BuildSeriesIndexFunction,
+    display_name = build_series_index,
+    sig_fn = build_index_signature,
+    ret = uint64
+)]
+pub(crate) async fn build_series_index(
+    table_mutation_handler: &TableMutationHandlerRef,
+    query_ctx: &QueryContextRef,
+    params: &[ValueRef<'_>],
+) -> Result<Value> {
+    build_index_impl(
+        table_mutation_handler,
+        query_ctx,
+        params,
+        "build_series_index",
+        Some(build_index_request::Options::SeriesIndex(Default::default())),
+    )
+    .await
+}
+
+async fn build_index_impl(
+    table_mutation_handler: &TableMutationHandlerRef,
+    query_ctx: &QueryContextRef,
+    params: &[ValueRef<'_>],
+    function: &str,
+    options: Option<build_index_request::Options>,
 ) -> Result<Value> {
    ensure!(
        params.len() == 1,
@@ -51,7 +91,7 @@ pub(crate) async fn build_index(

    let ValueRef::String(table_name) = params[0] else {
        return UnsupportedInputDataTypeSnafu {
-            function: "build_index",
+            function,
            datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
        }
        .fail();
@@ -64,6 +104,7 @@ pub(crate) async fn build_index(
    let affected_rows = table_mutation_handler
        .build_index(
            BuildIndexTableRequest {
+                options,
                catalog_name,
                schema_name,
                table_name,
@@ -78,3 +119,45 @@ pub(crate) async fn build_index(
 fn build_index_signature() -> Signature {
    Signature::uniform(1, vec![ArrowDataType::Utf8], Volatility::Immutable)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::function::FunctionContext;
+
+    #[tokio::test]
+    async fn test_build_series_index_arguments() {
+        let ctx = FunctionContext::mock();
+        let handler = ctx.state.table_mutation_handler.as_ref().unwrap();
+        for params in [
+            vec![],
+            vec![ValueRef::Int32(1)],
+            vec![ValueRef::Null],
+            vec![ValueRef::String("a"), ValueRef::String("b")],
+        ] {
+            assert!(
+                build_index_impl(
+                    handler,
+                    &ctx.query_ctx,
+                    &params,
+                    "build_series_index",
+                    Some(build_index_request::Options::SeriesIndex(Default::default()))
+                )
+                .await
+                .is_err()
+            );
+        }
+        assert_eq!(
+            Value::UInt64(42),
+            build_index_impl(
+                handler,
+                &ctx.query_ctx,
+                &[ValueRef::String("greptime.public.test")],
+                "build_series_index",
+                Some(build_index_request::Options::SeriesIndex(Default::default())),
+            )
+            .await
+            .unwrap()
+        );
+    }
+}