fix: database base ttl (#4926)

* main:
 Add common-meta dependency and implement SchemaMetadataManager

 - Introduce `common-meta` as a new dependency in `mito2`.
 - Implement `SchemaMetadataManager` for managing schema-level metadata.
 - Update `DatanodeBuilder` and `MitoEngine` to pass `KvBackendRef` for schema metadata management.
 - Add `SchemaMetadataManager` to `RegionWorkerLoop` for compaction handling.
 - Include `SchemaNameKey` usage in compaction-related code.
 - Add `database_metadata_manager` module with `SchemaMetadataManager` struct and associated logic.

* fix/database-base-ttl:
 Refactor metadata management and update compaction logic

 - Remove `database_metadata_manager` and introduce `schema_metadata_manager`
 - Update compaction logic to handle TTL based on schema metadata
 - Adjust tests to use `schema_metadata_manager` for setting up schema options
 - Fix engine creation in tests to pass `kv_backend` explicitly
 - Remove unused imports and apply minor code cleanups

* fix/database-base-ttl:
 Extend CREATE TABLE LIKE to inherit schema options

 - Implement inheritance of database level options for CREATE TABLE LIKE
 - Add schema options to SHOW CREATE TABLE output
 - Refactor create_table_stmt to include schema_options in SQL generation
 - Update error handling to include TableMetadataManagerSnafu

* fix/database-base-ttl:
 Refactor error handling and remove schema dependency in table creation

 - Replace expect with the ? operator for error handling in open_compaction_region
 - Simplify create_logical_tables by removing catalog and schema name parameters
 - Remove unnecessary schema retrieval and merging of schema options in create_table_info
 - Clean up unused imports and redundant code

* fix/database-base-ttl:
 Refactor error handling and update documentation comments

 - Update comment to reflect retrieval of schema options instead of metadata
 - Introduce new error type `GetSchemaMetadataSnafu` for schema metadata retrieval failures
 - Implement error handling for schema metadata retrieval in `find_ttl` function

* fix: toml

* fix/database-base-ttl:
 Refactor SchemaMetadataManager and adjust Cargo.toml dependencies

 - Remove unused imports in schema_metadata_manager.rs
 - Add conditional compilation for SchemaMetadataManager::new
 - Update Cargo.toml to remove "testing" feature from common-meta dependency in main section and add it to dev-dependencies

* fix/database-base-ttl:
 Fix typos in comments and function names across multiple modules

 - Correct spelling of 'parallelism' in region_server, engine, and scan_region modules
 - Amend typo in TODO comment from 'persisent' to 'persistent' in server module
 - Update incorrect test query from 'versiona' to 'version' in federated module tests

* fix/database-base-ttl: Add schema existence check in StatementExecutor for CREATE TABLE operation

* fix/database-base-ttl: Add warning log for failed TTL retrieval in compaction region open function

* fix/database-base-ttl:
 Refactor to use SchemaMetadataManagerRef in Datanode and MitoEngine

 - Replace KvBackendRef with SchemaMetadataManagerRef across various components.
 - Update DatanodeBuilder and MitoEngine to pass SchemaMetadataManagerRef instead of KvBackendRef.
 - Adjust test cases to use get_schema_metadata_manager method for consistency.

src/operator/src/insert.rs

@@ -760,10 +760,8 @@ impl Inserter {
         ctx: &QueryContextRef,
         statement_executor: &StatementExecutor,
     ) -> Result<Vec<TableRef>> {
-        let catalog_name = ctx.current_catalog();
-        let schema_name = ctx.current_schema();
         let res = statement_executor
-            .create_logical_tables(catalog_name, &schema_name, &create_table_exprs, ctx.clone())
+            .create_logical_tables(&create_table_exprs, ctx.clone())
             .await;
 
         match res {

src/operator/src/statement/ddl.rs

@@ -26,7 +26,7 @@ use common_error::ext::BoxedError;
 use common_meta::cache_invalidator::Context;
 use common_meta::ddl::ExecutorContext;
 use common_meta::instruction::CacheIdent;
-use common_meta::key::schema_name::{SchemaNameKey, SchemaNameValue};
+use common_meta::key::schema_name::SchemaNameKey;
 use common_meta::key::NAME_PATTERN;
 use common_meta::rpc::ddl::{
     CreateFlowTask, DdlTask, DropFlowTask, DropViewTask, SubmitDdlTaskRequest,
@@ -116,9 +116,21 @@ impl StatementExecutor {
             .await
             .context(error::FindTablePartitionRuleSnafu { table_name: table })?;
 
+        // CREATE TABLE LIKE also inherits database level options.
+        let schema_options = self
+            .table_metadata_manager
+            .schema_manager()
+            .get(SchemaNameKey {
+                catalog: &catalog,
+                schema: &schema,
+            })
+            .await
+            .context(TableMetadataManagerSnafu)?;
+
         let quote_style = ctx.quote_style();
-        let mut create_stmt = create_table_stmt(&table_ref.table_info(), quote_style)
-            .context(error::ParseQuerySnafu)?;
+        let mut create_stmt =
+            create_table_stmt(&table_ref.table_info(), schema_options, quote_style)
+                .context(error::ParseQuerySnafu)?;
         create_stmt.name = stmt.table_name;
         create_stmt.if_not_exists = false;
 
@@ -165,15 +177,8 @@ impl StatementExecutor {
                 .table_options
                 .contains_key(LOGICAL_TABLE_METADATA_KEY)
         {
-            let catalog_name = &create_table.catalog_name;
-            let schema_name = &create_table.schema_name;
             return self
-                .create_logical_tables(
-                    catalog_name,
-                    schema_name,
-                    &[create_table.clone()],
-                    query_ctx,
-                )
+                .create_logical_tables(&[create_table.clone()], query_ctx)
                 .await?
                 .into_iter()
                 .next()
@@ -183,6 +188,7 @@ impl StatementExecutor {
         }
 
         let _timer = crate::metrics::DIST_CREATE_TABLE.start_timer();
+
         let schema = self
             .table_metadata_manager
             .schema_manager()
@@ -193,12 +199,12 @@ impl StatementExecutor {
             .await
             .context(TableMetadataManagerSnafu)?;
 
-        let Some(schema_opts) = schema else {
-            return SchemaNotFoundSnafu {
+        ensure!(
+            schema.is_some(),
+            SchemaNotFoundSnafu {
                 schema_info: &create_table.schema_name,
             }
-            .fail();
-        };
+        );
 
         // if table exists.
         if let Some(table) = self
@@ -240,7 +246,7 @@ impl StatementExecutor {
         );
 
         let (partitions, partition_cols) = parse_partitions(create_table, partitions, &query_ctx)?;
-        let mut table_info = create_table_info(create_table, partition_cols, schema_opts)?;
+        let mut table_info = create_table_info(create_table, partition_cols)?;
 
         let resp = self
             .create_table_procedure(
@@ -273,8 +279,6 @@ impl StatementExecutor {
     #[tracing::instrument(skip_all)]
     pub async fn create_logical_tables(
         &self,
-        catalog_name: &str,
-        schema_name: &str,
         create_table_exprs: &[CreateTableExpr],
         query_context: QueryContextRef,
     ) -> Result<Vec<TableRef>> {
@@ -296,19 +300,9 @@ impl StatementExecutor {
             );
         }
 
-        let schema = self
-            .table_metadata_manager
-            .schema_manager()
-            .get(SchemaNameKey::new(catalog_name, schema_name))
-            .await
-            .context(TableMetadataManagerSnafu)?
-            .context(SchemaNotFoundSnafu {
-                schema_info: schema_name,
-            })?;
-
         let mut raw_tables_info = create_table_exprs
             .iter()
-            .map(|create| create_table_info(create, vec![], schema.clone()))
+            .map(|create| create_table_info(create, vec![]))
             .collect::<Result<Vec<_>>>()?;
         let tables_data = create_table_exprs
             .iter()
@@ -1261,7 +1255,6 @@ fn parse_partitions(
 fn create_table_info(
     create_table: &CreateTableExpr,
     partition_columns: Vec<String>,
-    schema_opts: SchemaNameValue,
 ) -> Result<RawTableInfo> {
     let mut column_schemas = Vec::with_capacity(create_table.column_defs.len());
     let mut column_name_to_index_map = HashMap::new();
@@ -1310,7 +1303,6 @@ fn create_table_info(
 
     let table_options = TableOptions::try_from_iter(&create_table.table_options)
         .context(UnrecognizedTableOptionSnafu)?;
-    let table_options = merge_options(table_options, schema_opts);
 
     let meta = RawTableMeta {
         schema: raw_schema,
@@ -1495,12 +1487,6 @@ fn convert_value(
         .context(ParseSqlValueSnafu)
 }
 
-/// Merge table level table options with schema level table options.
-fn merge_options(mut table_opts: TableOptions, schema_opts: SchemaNameValue) -> TableOptions {
-    table_opts.ttl = table_opts.ttl.or(schema_opts.ttl);
-    table_opts
-}
-
 #[cfg(test)]
 mod test {
     use session::context::{QueryContext, QueryContextBuilder};

src/operator/src/statement/show.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use common_error::ext::BoxedError;
+use common_meta::key::schema_name::SchemaNameKey;
 use common_query::Output;
 use common_telemetry::tracing;
 use partition::manager::PartitionInfo;
@@ -33,7 +34,7 @@ use table::TableRef;
 
 use crate::error::{
     self, CatalogSnafu, ExecuteStatementSnafu, ExternalSnafu, FindViewInfoSnafu, InvalidSqlSnafu,
-    Result, ViewInfoNotFoundSnafu, ViewNotFoundSnafu,
+    Result, TableMetadataManagerSnafu, ViewInfoNotFoundSnafu, ViewNotFoundSnafu,
 };
 use crate::statement::StatementExecutor;
 
@@ -118,6 +119,16 @@ impl StatementExecutor {
             .fail();
         }
 
+        let schema_options = self
+            .table_metadata_manager
+            .schema_manager()
+            .get(SchemaNameKey {
+                catalog: &table_name.catalog_name,
+                schema: &table_name.schema_name,
+            })
+            .await
+            .context(TableMetadataManagerSnafu)?;
+
         let partitions = self
             .partition_manager
             .find_table_partitions(table.table_info().table_id())
@@ -128,7 +139,8 @@ impl StatementExecutor {
 
         let partitions = create_partitions_stmt(partitions)?;
 
-        query::sql::show_create_table(table, partitions, query_ctx).context(ExecuteStatementSnafu)
+        query::sql::show_create_table(table, schema_options, partitions, query_ctx)
+            .context(ExecuteStatementSnafu)
     }
 
     #[tracing::instrument(skip_all)]