feat: introduce reconciliation interface (#6614)

* feat: introduce reconcile interface

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: apply suggestions from CR

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: apply suggestions from CR

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: upgrade proto

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>

src/common/function/src/admin.rs

@@ -16,6 +16,9 @@ mod add_region_follower;
 mod flush_compact_region;
 mod flush_compact_table;
 mod migrate_region;
+mod reconcile_catalog;
+mod reconcile_database;
+mod reconcile_table;
 mod remove_region_follower;
 
 use std::sync::Arc;
@@ -24,6 +27,9 @@ use add_region_follower::AddRegionFollowerFunction;
 use flush_compact_region::{CompactRegionFunction, FlushRegionFunction};
 use flush_compact_table::{CompactTableFunction, FlushTableFunction};
 use migrate_region::MigrateRegionFunction;
+use reconcile_catalog::ReconcileCatalogFunction;
+use reconcile_database::ReconcileDatabaseFunction;
+use reconcile_table::ReconcileTableFunction;
 use remove_region_follower::RemoveRegionFollowerFunction;
 
 use crate::flush_flow::FlushFlowFunction;
@@ -43,5 +49,8 @@ impl AdminFunction {
         registry.register_async(Arc::new(FlushTableFunction));
         registry.register_async(Arc::new(CompactTableFunction));
         registry.register_async(Arc::new(FlushFlowFunction));
+        registry.register_async(Arc::new(ReconcileCatalogFunction));
+        registry.register_async(Arc::new(ReconcileDatabaseFunction));
+        registry.register_async(Arc::new(ReconcileTableFunction));
     }
 }

src/common/function/src/admin/reconcile_catalog.rs

@@ -0,0 +1,179 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use api::v1::meta::reconcile_request::Target;
+use api::v1::meta::{ReconcileCatalog, ReconcileRequest};
+use common_macro::admin_fn;
+use common_query::error::{
+    InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result,
+    UnsupportedInputDataTypeSnafu,
+};
+use common_query::prelude::{Signature, TypeSignature, Volatility};
+use common_telemetry::info;
+use datatypes::prelude::*;
+use session::context::QueryContextRef;
+
+use crate::handlers::ProcedureServiceHandlerRef;
+use crate::helper::{
+    cast_u32, default_parallelism, default_resolve_strategy, get_string_from_params,
+    parse_resolve_strategy,
+};
+
+const FN_NAME: &str = "reconcile_catalog";
+
+/// A function to reconcile a catalog.
+/// Returns the procedure id if success.
+///
+/// - `reconcile_catalog(resolve_strategy)`.
+/// - `reconcile_catalog(resolve_strategy, parallelism)`.
+///
+/// - `reconcile_catalog()`.
+#[admin_fn(
+    name = ReconcileCatalogFunction,
+    display_name = reconcile_catalog,
+    sig_fn = signature,
+    ret = string
+)]
+pub(crate) async fn reconcile_catalog(
+    procedure_service_handler: &ProcedureServiceHandlerRef,
+    query_ctx: &QueryContextRef,
+    params: &[ValueRef<'_>],
+) -> Result<Value> {
+    let (resolve_strategy, parallelism) = match params.len() {
+        0 => (default_resolve_strategy(), default_parallelism()),
+        1 => (
+            parse_resolve_strategy(get_string_from_params(params, 0, FN_NAME)?)?,
+            default_parallelism(),
+        ),
+        2 => {
+            let Some(parallelism) = cast_u32(&params[1])? else {
+                return UnsupportedInputDataTypeSnafu {
+                    function: FN_NAME,
+                    datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
+                }
+                .fail();
+            };
+            (
+                parse_resolve_strategy(get_string_from_params(params, 0, FN_NAME)?)?,
+                parallelism,
+            )
+        }
+        size => {
+            return InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect 0, 1 or 2, have: {}",
+                    size
+                ),
+            }
+            .fail();
+        }
+    };
+    info!(
+        "Reconciling catalog with resolve_strategy: {:?}, parallelism: {}",
+        resolve_strategy, parallelism
+    );
+    let pid = procedure_service_handler
+        .reconcile(ReconcileRequest {
+            target: Some(Target::ReconcileCatalog(ReconcileCatalog {
+                catalog_name: query_ctx.current_catalog().to_string(),
+                parallelism,
+                resolve_strategy: resolve_strategy as i32,
+            })),
+            ..Default::default()
+        })
+        .await?;
+    match pid {
+        Some(pid) => Ok(Value::from(pid)),
+        None => Ok(Value::Null),
+    }
+}
+
+fn signature() -> Signature {
+    let nums = ConcreteDataType::numerics();
+    let mut signs = Vec::with_capacity(2 + nums.len());
+    signs.extend([
+        // reconcile_catalog()
+        TypeSignature::NullAry,
+        // reconcile_catalog(resolve_strategy)
+        TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
+    ]);
+    for sign in nums {
+        // reconcile_catalog(resolve_strategy, parallelism)
+        signs.push(TypeSignature::Exact(vec![
+            ConcreteDataType::string_datatype(),
+            sign,
+        ]));
+    }
+    Signature::one_of(signs, Volatility::Immutable)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::assert_matches::assert_matches;
+    use std::sync::Arc;
+
+    use common_query::error::Error;
+    use datatypes::vectors::{StringVector, UInt64Vector, VectorRef};
+
+    use crate::admin::reconcile_catalog::ReconcileCatalogFunction;
+    use crate::function::{AsyncFunction, FunctionContext};
+
+    #[tokio::test]
+    async fn test_reconcile_catalog() {
+        common_telemetry::init_default_ut_logging();
+
+        // reconcile_catalog()
+        let f = ReconcileCatalogFunction;
+        let args = vec![];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // reconcile_catalog(resolve_strategy)
+        let f = ReconcileCatalogFunction;
+        let args = vec![Arc::new(StringVector::from(vec!["UseMetasrv"])) as _];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // reconcile_catalog(resolve_strategy, parallelism)
+        let f = ReconcileCatalogFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
+            Arc::new(UInt64Vector::from_slice([10])) as _,
+        ];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // unsupported input data type
+        let f = ReconcileCatalogFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
+            Arc::new(StringVector::from(vec!["test"])) as _,
+        ];
+        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
+        assert_matches!(err, Error::UnsupportedInputDataType { .. });
+
+        // invalid function args
+        let f = ReconcileCatalogFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
+            Arc::new(UInt64Vector::from_slice([10])) as _,
+            Arc::new(StringVector::from(vec!["10"])) as _,
+        ];
+        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
+        assert_matches!(err, Error::InvalidFuncArgs { .. });
+    }
+}

src/common/function/src/admin/reconcile_database.rs

@@ -0,0 +1,198 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use api::v1::meta::reconcile_request::Target;
+use api::v1::meta::{ReconcileDatabase, ReconcileRequest};
+use common_macro::admin_fn;
+use common_query::error::{
+    InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result,
+    UnsupportedInputDataTypeSnafu,
+};
+use common_query::prelude::{Signature, TypeSignature, Volatility};
+use common_telemetry::info;
+use datatypes::prelude::*;
+use session::context::QueryContextRef;
+
+use crate::handlers::ProcedureServiceHandlerRef;
+use crate::helper::{
+    cast_u32, default_parallelism, default_resolve_strategy, get_string_from_params,
+    parse_resolve_strategy,
+};
+
+const FN_NAME: &str = "reconcile_database";
+
+/// A function to reconcile a database.
+/// Returns the procedure id if success.
+///
+/// - `reconcile_database(database_name)`.
+/// - `reconcile_database(database_name, resolve_strategy)`.
+/// - `reconcile_database(database_name, resolve_strategy, parallelism)`.
+///
+/// The parameters:
+/// - `database_name`:  the database name
+#[admin_fn(
+    name = ReconcileDatabaseFunction,
+    display_name = reconcile_database,
+    sig_fn = signature,
+    ret = string
+)]
+pub(crate) async fn reconcile_database(
+    procedure_service_handler: &ProcedureServiceHandlerRef,
+    query_ctx: &QueryContextRef,
+    params: &[ValueRef<'_>],
+) -> Result<Value> {
+    let (database_name, resolve_strategy, parallelism) = match params.len() {
+        1 => (
+            get_string_from_params(params, 0, FN_NAME)?,
+            default_resolve_strategy(),
+            default_parallelism(),
+        ),
+        2 => (
+            get_string_from_params(params, 0, FN_NAME)?,
+            parse_resolve_strategy(get_string_from_params(params, 1, FN_NAME)?)?,
+            default_parallelism(),
+        ),
+        3 => {
+            let Some(parallelism) = cast_u32(&params[2])? else {
+                return UnsupportedInputDataTypeSnafu {
+                    function: FN_NAME,
+                    datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
+                }
+                .fail();
+            };
+            (
+                get_string_from_params(params, 0, FN_NAME)?,
+                parse_resolve_strategy(get_string_from_params(params, 1, FN_NAME)?)?,
+                parallelism,
+            )
+        }
+        size => {
+            return InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect 1, 2 or 3, have: {}",
+                    size
+                ),
+            }
+            .fail();
+        }
+    };
+    info!(
+        "Reconciling database: {}, resolve_strategy: {:?}, parallelism: {}",
+        database_name, resolve_strategy, parallelism
+    );
+    let pid = procedure_service_handler
+        .reconcile(ReconcileRequest {
+            target: Some(Target::ReconcileDatabase(ReconcileDatabase {
+                catalog_name: query_ctx.current_catalog().to_string(),
+                database_name: database_name.to_string(),
+                parallelism,
+                resolve_strategy: resolve_strategy as i32,
+            })),
+            ..Default::default()
+        })
+        .await?;
+    match pid {
+        Some(pid) => Ok(Value::from(pid)),
+        None => Ok(Value::Null),
+    }
+}
+
+fn signature() -> Signature {
+    let nums = ConcreteDataType::numerics();
+    let mut signs = Vec::with_capacity(2 + nums.len());
+    signs.extend([
+        // reconcile_database(datanode_name)
+        TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
+        // reconcile_database(database_name, resolve_strategy)
+        TypeSignature::Exact(vec![
+            ConcreteDataType::string_datatype(),
+            ConcreteDataType::string_datatype(),
+        ]),
+    ]);
+    for sign in nums {
+        // reconcile_database(database_name, resolve_strategy, parallelism)
+        signs.push(TypeSignature::Exact(vec![
+            ConcreteDataType::string_datatype(),
+            ConcreteDataType::string_datatype(),
+            sign,
+        ]));
+    }
+    Signature::one_of(signs, Volatility::Immutable)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::assert_matches::assert_matches;
+    use std::sync::Arc;
+
+    use common_query::error::Error;
+    use datatypes::vectors::{StringVector, UInt32Vector, VectorRef};
+
+    use crate::admin::reconcile_database::ReconcileDatabaseFunction;
+    use crate::function::{AsyncFunction, FunctionContext};
+
+    #[tokio::test]
+    async fn test_reconcile_catalog() {
+        common_telemetry::init_default_ut_logging();
+
+        // reconcile_database(database_name)
+        let f = ReconcileDatabaseFunction;
+        let args = vec![Arc::new(StringVector::from(vec!["test"])) as _];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // reconcile_database(database_name, resolve_strategy)
+        let f = ReconcileDatabaseFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["test"])) as _,
+            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
+        ];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // reconcile_database(database_name, resolve_strategy, parallelism)
+        let f = ReconcileDatabaseFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["test"])) as _,
+            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
+            Arc::new(UInt32Vector::from_slice([10])) as _,
+        ];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // invalid function args
+        let f = ReconcileDatabaseFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
+            Arc::new(UInt32Vector::from_slice([10])) as _,
+            Arc::new(StringVector::from(vec!["v1"])) as _,
+            Arc::new(StringVector::from(vec!["v2"])) as _,
+        ];
+        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
+        assert_matches!(err, Error::InvalidFuncArgs { .. });
+
+        // unsupported input data type
+        let f = ReconcileDatabaseFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
+            Arc::new(UInt32Vector::from_slice([10])) as _,
+            Arc::new(StringVector::from(vec!["v1"])) as _,
+        ];
+        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
+        assert_matches!(err, Error::UnsupportedInputDataType { .. });
+    }
+}

src/common/function/src/admin/reconcile_table.rs

@@ -0,0 +1,149 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use api::v1::meta::reconcile_request::Target;
+use api::v1::meta::{ReconcileRequest, ReconcileTable, ResolveStrategy};
+use common_catalog::format_full_table_name;
+use common_error::ext::BoxedError;
+use common_macro::admin_fn;
+use common_query::error::{
+    MissingProcedureServiceHandlerSnafu, Result, TableMutationSnafu, UnsupportedInputDataTypeSnafu,
+};
+use common_query::prelude::{Signature, TypeSignature, Volatility};
+use common_telemetry::info;
+use datatypes::prelude::*;
+use session::context::QueryContextRef;
+use session::table_name::table_name_to_full_name;
+use snafu::ResultExt;
+
+use crate::handlers::ProcedureServiceHandlerRef;
+use crate::helper::parse_resolve_strategy;
+
+const FN_NAME: &str = "reconcile_table";
+
+/// A function to reconcile a table.
+/// Returns the procedure id if success.
+///
+/// - `reconcile_table(table_name)`.
+/// - `reconcile_table(table_name, resolve_strategy)`.
+///
+/// The parameters:
+/// - `table_name`:  the table name
+#[admin_fn(
+    name = ReconcileTableFunction,
+    display_name = reconcile_table,
+    sig_fn = signature,
+    ret = string
+)]
+pub(crate) async fn reconcile_table(
+    procedure_service_handler: &ProcedureServiceHandlerRef,
+    query_ctx: &QueryContextRef,
+    params: &[ValueRef<'_>],
+) -> Result<Value> {
+    let (table_name, resolve_strategy) = match params {
+        [ValueRef::String(table_name)] => (table_name, ResolveStrategy::UseLatest),
+        [ValueRef::String(table_name), ValueRef::String(resolve_strategy)] => {
+            (table_name, parse_resolve_strategy(resolve_strategy)?)
+        }
+        _ => {
+            return UnsupportedInputDataTypeSnafu {
+                function: FN_NAME,
+                datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
+            }
+            .fail()
+        }
+    };
+    let (catalog_name, schema_name, table_name) = table_name_to_full_name(table_name, query_ctx)
+        .map_err(BoxedError::new)
+        .context(TableMutationSnafu)?;
+    info!(
+        "Reconciling table: {} with resolve_strategy: {:?}",
+        format_full_table_name(&catalog_name, &schema_name, &table_name),
+        resolve_strategy
+    );
+    let pid = procedure_service_handler
+        .reconcile(ReconcileRequest {
+            target: Some(Target::ReconcileTable(ReconcileTable {
+                catalog_name,
+                schema_name,
+                table_name,
+                resolve_strategy: resolve_strategy as i32,
+            })),
+            ..Default::default()
+        })
+        .await?;
+    match pid {
+        Some(pid) => Ok(Value::from(pid)),
+        None => Ok(Value::Null),
+    }
+}
+
+fn signature() -> Signature {
+    Signature::one_of(
+        vec![
+            // reconcile_table(table_name)
+            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
+            // reconcile_table(table_name, resolve_strategy)
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+        ],
+        Volatility::Immutable,
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use std::assert_matches::assert_matches;
+    use std::sync::Arc;
+
+    use common_query::error::Error;
+    use datatypes::vectors::{StringVector, VectorRef};
+
+    use crate::admin::reconcile_table::ReconcileTableFunction;
+    use crate::function::{AsyncFunction, FunctionContext};
+
+    #[tokio::test]
+    async fn test_reconcile_table() {
+        common_telemetry::init_default_ut_logging();
+
+        // reconcile_table(table_name)
+        let f = ReconcileTableFunction;
+        let args = vec![Arc::new(StringVector::from(vec!["test"])) as _];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // reconcile_table(table_name, resolve_strategy)
+        let f = ReconcileTableFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["test"])) as _,
+            Arc::new(StringVector::from(vec!["UseMetasrv"])) as _,
+        ];
+        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
+        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
+        assert_eq!(expect, result);
+
+        // unsupported input data type
+        let f = ReconcileTableFunction;
+        let args = vec![
+            Arc::new(StringVector::from(vec!["test"])) as _,
+            Arc::new(StringVector::from(vec!["UseMetasrv"])) as _,
+            Arc::new(StringVector::from(vec!["10"])) as _,
+        ];
+        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
+        assert_matches!(err, Error::UnsupportedInputDataType { .. });
+    }
+}

src/common/function/src/handlers.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use api::v1::meta::ReconcileRequest;
 use async_trait::async_trait;
 use catalog::CatalogManagerRef;
 use common_base::AffectedRows;
@@ -65,6 +66,9 @@ pub trait ProcedureServiceHandler: Send + Sync {
     /// Migrate a region from source peer to target peer, returns the procedure id if success.
     async fn migrate_region(&self, request: MigrateRegionRequest) -> Result<Option<String>>;
 
+    /// Reconcile a table, database or catalog, returns the procedure id if success.
+    async fn reconcile(&self, request: ReconcileRequest) -> Result<Option<String>>;
+
     /// Query the procedure' state by its id
     async fn query_procedure_state(&self, pid: &str) -> Result<ProcedureStateResponse>;
 

src/common/function/src/helper.rs

@@ -12,12 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_query::error::{InvalidInputTypeSnafu, Result};
+use api::v1::meta::ResolveStrategy;
+use common_query::error::{
+    InvalidFuncArgsSnafu, InvalidInputTypeSnafu, Result, UnsupportedInputDataTypeSnafu,
+};
 use common_query::prelude::{Signature, TypeSignature, Volatility};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::types::cast::cast;
 use datatypes::value::ValueRef;
-use snafu::ResultExt;
+use snafu::{OptionExt, ResultExt};
 
 /// Create a function signature with oneof signatures of interleaving two arguments.
 pub fn one_of_sigs2(args1: Vec<ConcreteDataType>, args2: Vec<ConcreteDataType>) -> Signature {
@@ -43,3 +46,64 @@ pub fn cast_u64(value: &ValueRef) -> Result<Option<u64>> {
         })
         .map(|v| v.as_u64())
 }
+
+/// Cast a [`ValueRef`] to u32, returns `None` if fails
+pub fn cast_u32(value: &ValueRef) -> Result<Option<u32>> {
+    cast((*value).into(), &ConcreteDataType::uint32_datatype())
+        .context(InvalidInputTypeSnafu {
+            err_msg: format!(
+                "Failed to cast input into uint32, actual type: {:#?}",
+                value.data_type(),
+            ),
+        })
+        .map(|v| v.as_u64().map(|v| v as u32))
+}
+
+/// Parse a resolve strategy from a string.
+pub fn parse_resolve_strategy(strategy: &str) -> Result<ResolveStrategy> {
+    ResolveStrategy::from_str_name(strategy).context(InvalidFuncArgsSnafu {
+        err_msg: format!("Invalid resolve strategy: {}", strategy),
+    })
+}
+
+/// Default parallelism for reconcile operations.
+pub fn default_parallelism() -> u32 {
+    64
+}
+
+/// Default resolve strategy for reconcile operations.
+pub fn default_resolve_strategy() -> ResolveStrategy {
+    ResolveStrategy::UseLatest
+}
+
+/// Get the string value from the params.
+///
+/// # Errors
+/// Returns an error if the input type is not a string.
+pub fn get_string_from_params<'a>(
+    params: &'a [ValueRef<'a>],
+    index: usize,
+    fn_name: &'a str,
+) -> Result<&'a str> {
+    let ValueRef::String(s) = &params[index] else {
+        return UnsupportedInputDataTypeSnafu {
+            function: fn_name,
+            datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
+        }
+        .fail();
+    };
+    Ok(s)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_resolve_strategy() {
+        assert_eq!(
+            parse_resolve_strategy("UseLatest").unwrap(),
+            ResolveStrategy::UseLatest
+        );
+    }
+}

src/common/function/src/lib.rs

@@ -14,6 +14,7 @@
 
 #![feature(let_chains)]
 #![feature(try_blocks)]
+#![feature(assert_matches)]
 
 mod admin;
 mod flush_flow;

src/common/function/src/state.rs

@@ -32,7 +32,7 @@ impl FunctionState {
     pub fn mock() -> Self {
         use std::sync::Arc;
 
-        use api::v1::meta::ProcedureStatus;
+        use api::v1::meta::{ProcedureStatus, ReconcileRequest};
         use async_trait::async_trait;
         use catalog::CatalogManagerRef;
         use common_base::AffectedRows;
@@ -63,6 +63,10 @@ impl FunctionState {
                 Ok(Some("test_pid".to_string()))
             }
 
+            async fn reconcile(&self, _request: ReconcileRequest) -> Result<Option<String>> {
+                Ok(Some("test_pid".to_string()))
+            }
+
             async fn query_procedure_state(&self, _pid: &str) -> Result<ProcedureStateResponse> {
                 Ok(ProcedureStateResponse {
                     status: ProcedureStatus::Done.into(),