From 1616175fe1ef26f340fd7a5d03d4a5503a99e46e Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Mon, 25 Mar 2024 20:38:44 +0800 Subject: [PATCH] feat: invalidate cache step4 --- src/cmd/src/frontend.rs | 4 +--- src/cmd/src/standalone.rs | 3 ++- .../meta/src/ddl/alter_logical_tables.rs | 18 +++++++----------- tests-integration/src/cluster.rs | 4 +--- tests-integration/src/standalone.rs | 3 ++- .../common/alter/alter_metric_table.result | 1 + .../common/create/create_metric_table.result | 9 +++++++++ .../common/create/create_metric_table.sql | 2 ++ 8 files changed, 25 insertions(+), 19 deletions(-) diff --git a/src/cmd/src/frontend.rs b/src/cmd/src/frontend.rs index 09a9eb2fdb37..39c7982eae79 100644 --- a/src/cmd/src/frontend.rs +++ b/src/cmd/src/frontend.rs @@ -253,9 +253,7 @@ impl StartCommand { let executor = HandlerGroupExecutor::new(vec![ Arc::new(ParseMailboxMessageHandler), - Arc::new(InvalidateTableCacheHandler::new( - cached_meta_backend.clone(), - )), + Arc::new(InvalidateTableCacheHandler::new(catalog_manager.clone())), ]); let heartbeat_task = HeartbeatTask::new( diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 1d08ab0e1218..5ce3c6bb88da 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -439,11 +439,12 @@ impl StartCommand { let mut frontend = FrontendBuilder::new( kv_backend, - catalog_manager, + catalog_manager.clone(), datanode_manager, ddl_task_executor, ) .with_plugin(fe_plugins.clone()) + .with_cache_invalidator(catalog_manager) .try_build() .await .context(StartFrontendSnafu)?; diff --git a/src/common/meta/src/ddl/alter_logical_tables.rs b/src/common/meta/src/ddl/alter_logical_tables.rs index 82939836714f..1ee6c080d9fd 100644 --- a/src/common/meta/src/ddl/alter_logical_tables.rs +++ b/src/common/meta/src/ddl/alter_logical_tables.rs @@ -37,7 +37,7 @@ use crate::error::{DecodeJsonSnafu, Error, MetadataCorruptionSnafu, Result}; use crate::key::table_info::TableInfoValue; use crate::key::table_route::PhysicalTableRouteValue; use crate::key::DeserializedValueWithBytes; -use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock}; +use crate::lock_key::{CatalogLock, SchemaLock, TableLock}; use crate::rpc::ddl::AlterTableTask; use crate::rpc::router::{find_leader_regions, find_leaders}; use crate::{cache_invalidator, metrics, ClusterId}; @@ -259,17 +259,13 @@ impl Procedure for AlterLogicalTablesProcedure { lock_key.push(CatalogLock::Read(table_ref.catalog).into()); lock_key.push(SchemaLock::read(table_ref.catalog, table_ref.schema).into()); lock_key.push(TableLock::Write(self.data.physical_table_id).into()); + lock_key.extend( + self.data + .table_info_values + .iter() + .map(|table| TableLock::Write(table.table_info.ident.table_id).into()), + ); - for task in &self.data.tasks { - lock_key.push( - TableNameLock::new( - &task.alter_table.catalog_name, - &task.alter_table.schema_name, - &task.alter_table.table_name, - ) - .into(), - ); - } LockKey::new(lock_key) } } diff --git a/tests-integration/src/cluster.rs b/tests-integration/src/cluster.rs index 2cf1ab5c435c..d427b2a8f27b 100644 --- a/tests-integration/src/cluster.rs +++ b/tests-integration/src/cluster.rs @@ -358,9 +358,7 @@ impl GreptimeDbClusterBuilder { let handlers_executor = HandlerGroupExecutor::new(vec![ Arc::new(ParseMailboxMessageHandler), - Arc::new(InvalidateTableCacheHandler::new( - cached_meta_backend.clone(), - )), + Arc::new(InvalidateTableCacheHandler::new(catalog_manager.clone())), ]); let heartbeat_task = HeartbeatTask::new( diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 05c66eff5334..0c9a58284cca 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -160,11 +160,12 @@ impl GreptimeDbStandaloneBuilder { let instance = FrontendBuilder::new( kv_backend.clone(), - catalog_manager, + catalog_manager.clone(), datanode_manager, ddl_task_executor, ) .with_plugin(plugins) + .with_cache_invalidator(catalog_manager) .try_build() .await .unwrap(); diff --git a/tests/cases/standalone/common/alter/alter_metric_table.result b/tests/cases/standalone/common/alter/alter_metric_table.result index d737b419d55b..8ae541b71e07 100644 --- a/tests/cases/standalone/common/alter/alter_metric_table.result +++ b/tests/cases/standalone/common/alter/alter_metric_table.result @@ -93,6 +93,7 @@ DESC TABLE phy; | __tsid | UInt64 | PRI | NO | | TAG | | host | String | PRI | YES | | TAG | | job | String | PRI | YES | | TAG | +| k | String | PRI | YES | | TAG | +------------+----------------------+-----+------+---------+---------------+ DROP TABLE t1; diff --git a/tests/cases/standalone/common/create/create_metric_table.result b/tests/cases/standalone/common/create/create_metric_table.result index b6578c5ca965..0a153ec733fc 100644 --- a/tests/cases/standalone/common/create/create_metric_table.result +++ b/tests/cases/standalone/common/create/create_metric_table.result @@ -11,6 +11,15 @@ SHOW TABLES; | phy | +---------+ +DESC TABLE phy; + ++--------+----------------------+-----+------+---------+---------------+ +| Column | Type | Key | Null | Default | Semantic Type | ++--------+----------------------+-----+------+---------+---------------+ +| ts | TimestampMillisecond | PRI | NO | | TIMESTAMP | +| val | Float64 | | YES | | FIELD | ++--------+----------------------+-----+------+---------+---------------+ + CREATE TABLE t1 (ts timestamp time index, val double, host string primary key) engine = metric with ("on_physical_table" = "phy"); Affected Rows: 0 diff --git a/tests/cases/standalone/common/create/create_metric_table.sql b/tests/cases/standalone/common/create/create_metric_table.sql index 28b3083d9037..fcc41ff11541 100644 --- a/tests/cases/standalone/common/create/create_metric_table.sql +++ b/tests/cases/standalone/common/create/create_metric_table.sql @@ -2,6 +2,8 @@ CREATE TABLE phy (ts timestamp time index, val double) engine=metric with ("phys SHOW TABLES; +DESC TABLE phy; + CREATE TABLE t1 (ts timestamp time index, val double, host string primary key) engine = metric with ("on_physical_table" = "phy"); CREATE TABLE t2 (ts timestamp time index, job string primary key, val double) engine = metric with ("on_physical_table" = "phy");