common_meta/reconciliation/
reconcile_logical_tables.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15pub(crate) mod reconcile_regions;
16pub(crate) mod reconciliation_end;
17pub(crate) mod reconciliation_start;
18pub(crate) mod resolve_table_metadatas;
19pub(crate) mod update_table_infos;
20
21use std::any::Any;
22use std::fmt::Debug;
23
24use async_trait::async_trait;
25use common_error::ext::ErrorExt;
26use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
27use common_procedure::{
28    Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
29    Procedure, Result as ProcedureResult, Status,
30};
31use serde::{Deserialize, Serialize};
32use snafu::ResultExt;
33use store_api::metadata::ColumnMetadata;
34use store_api::storage::TableId;
35use table::metadata::TableInfo;
36use table::table_name::TableName;
37
38use crate::cache_invalidator::CacheInvalidatorRef;
39use crate::error::Result;
40use crate::key::table_info::TableInfoValue;
41use crate::key::table_route::PhysicalTableRouteValue;
42use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
43use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
44use crate::metrics;
45use crate::node_manager::NodeManagerRef;
46use crate::reconciliation::event::{
47    RECONCILE_LOGICAL_TABLES_EVENT_TYPE, ReconcileLogicalTablesEvent, ReconciliationLocator,
48};
49use crate::reconciliation::reconcile_logical_tables::reconciliation_start::ReconciliationStart;
50use crate::reconciliation::utils::{Context, ReconcileLogicalTableMetrics};
51
52pub struct ReconcileLogicalTablesContext {
53    pub node_manager: NodeManagerRef,
54    pub table_metadata_manager: TableMetadataManagerRef,
55    pub cache_invalidator: CacheInvalidatorRef,
56    pub persistent_ctx: PersistentContext,
57    pub volatile_ctx: VolatileContext,
58}
59
60impl ReconcileLogicalTablesContext {
61    /// Creates a new [`ReconcileLogicalTablesContext`] with the given [`Context`] and [`PersistentContext`].
62    pub fn new(ctx: Context, persistent_ctx: PersistentContext) -> Self {
63        Self {
64            node_manager: ctx.node_manager,
65            table_metadata_manager: ctx.table_metadata_manager,
66            cache_invalidator: ctx.cache_invalidator,
67            persistent_ctx,
68            volatile_ctx: VolatileContext::default(),
69        }
70    }
71
72    /// Returns the physical table name.
73    pub(crate) fn table_name(&self) -> &TableName {
74        &self.persistent_ctx.table_name
75    }
76
77    /// Returns the physical table id.
78    pub(crate) fn table_id(&self) -> TableId {
79        self.persistent_ctx.table_id
80    }
81
82    /// Returns a mutable reference to the metrics.
83    pub(crate) fn mut_metrics(&mut self) -> &mut ReconcileLogicalTableMetrics {
84        &mut self.volatile_ctx.metrics
85    }
86
87    /// Returns a reference to the metrics.
88    pub(crate) fn metrics(&self) -> &ReconcileLogicalTableMetrics {
89        &self.volatile_ctx.metrics
90    }
91}
92
93#[derive(Debug, Serialize, Deserialize)]
94pub(crate) struct PersistentContext {
95    pub(crate) table_id: TableId,
96    pub(crate) table_name: TableName,
97    // The logical tables need to be reconciled.
98    // The logical tables belongs to the physical table.
99    pub(crate) logical_tables: Vec<TableName>,
100    // The logical table ids.
101    // The value will be set in `ReconciliationStart` state.
102    pub(crate) logical_table_ids: Vec<TableId>,
103    /// The table info value.
104    /// The value will be set in `ReconciliationStart` state.
105    pub(crate) table_info_value: Option<DeserializedValueWithBytes<TableInfoValue>>,
106    // The physical table route.
107    // The value will be set in `ReconciliationStart` state.
108    pub(crate) physical_table_route: Option<PhysicalTableRouteValue>,
109    // The table infos to be updated.
110    // The value will be set in `ResolveTableMetadatas` state.
111    pub(crate) update_table_infos: Vec<(TableId, Vec<ColumnMetadata>)>,
112    // The table infos to be created.
113    // The value will be set in `ResolveTableMetadatas` state.
114    pub(crate) create_tables: Vec<(TableId, TableInfo)>,
115    // Whether the procedure is a subprocedure.
116    pub(crate) is_subprocedure: bool,
117}
118
119impl PersistentContext {
120    pub(crate) fn new(
121        table_id: TableId,
122        table_name: TableName,
123        logical_tables: Vec<(TableId, TableName)>,
124        is_subprocedure: bool,
125    ) -> Self {
126        let (logical_table_ids, logical_tables) = logical_tables.into_iter().unzip();
127
128        Self {
129            table_id,
130            table_name,
131            logical_tables,
132            logical_table_ids,
133            table_info_value: None,
134            physical_table_route: None,
135            update_table_infos: vec![],
136            create_tables: vec![],
137            is_subprocedure,
138        }
139    }
140}
141
142#[derive(Default)]
143pub(crate) struct VolatileContext {
144    pub(crate) metrics: ReconcileLogicalTableMetrics,
145}
146
147pub struct ReconcileLogicalTablesProcedure {
148    pub context: ReconcileLogicalTablesContext,
149    state: Box<dyn State>,
150}
151
152#[derive(Debug, Serialize)]
153struct ProcedureData<'a> {
154    state: &'a dyn State,
155    persistent_ctx: &'a PersistentContext,
156}
157
158#[derive(Debug, Deserialize)]
159struct ProcedureDataOwned {
160    state: Box<dyn State>,
161    persistent_ctx: PersistentContext,
162}
163
164impl ReconcileLogicalTablesProcedure {
165    pub const TYPE_NAME: &'static str = "metasrv-procedure::ReconcileLogicalTables";
166
167    pub fn new(
168        ctx: Context,
169        table_id: TableId,
170        table_name: TableName,
171        logical_tables: Vec<(TableId, TableName)>,
172        is_subprocedure: bool,
173    ) -> Self {
174        let persistent_ctx =
175            PersistentContext::new(table_id, table_name, logical_tables, is_subprocedure);
176        let context = ReconcileLogicalTablesContext::new(ctx, persistent_ctx);
177        let state = Box::new(ReconciliationStart);
178        Self { context, state }
179    }
180
181    pub(crate) fn from_json(ctx: Context, json: &str) -> ProcedureResult<Self> {
182        let ProcedureDataOwned {
183            state,
184            persistent_ctx,
185        } = serde_json::from_str(json).context(FromJsonSnafu)?;
186        let context = ReconcileLogicalTablesContext::new(ctx, persistent_ctx);
187        Ok(Self { context, state })
188    }
189}
190
191#[async_trait]
192impl Procedure for ReconcileLogicalTablesProcedure {
193    fn type_name(&self) -> &str {
194        Self::TYPE_NAME
195    }
196
197    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
198        let state = &mut self.state;
199
200        let procedure_name = Self::TYPE_NAME;
201        let step = state.name();
202        let _timer = metrics::METRIC_META_RECONCILIATION_PROCEDURE
203            .with_label_values(&[procedure_name, step])
204            .start_timer();
205        match state.next(&mut self.context, _ctx).await {
206            Ok((next, status)) => {
207                *state = next;
208                Ok(status)
209            }
210            Err(e) => {
211                if e.retry_hint().is_retryable() {
212                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
213                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
214                        .inc();
215                    Err(ProcedureError::retry_later(e))
216                } else {
217                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
218                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_EXTERNAL])
219                        .inc();
220                    Err(ProcedureError::external(e))
221                }
222            }
223        }
224    }
225
226    fn dump(&self) -> ProcedureResult<String> {
227        let data = ProcedureData {
228            state: self.state.as_ref(),
229            persistent_ctx: &self.context.persistent_ctx,
230        };
231        serde_json::to_string(&data).context(ToJsonSnafu)
232    }
233
234    fn lock_key(&self) -> LockKey {
235        let table_ref = &self.context.table_name().table_ref();
236
237        let mut table_ids = self
238            .context
239            .persistent_ctx
240            .logical_table_ids
241            .iter()
242            .map(|t| TableLock::Write(*t).into())
243            .collect::<Vec<_>>();
244        table_ids.sort_unstable();
245        table_ids.push(TableLock::Read(self.context.table_id()).into());
246        if self.context.persistent_ctx.is_subprocedure {
247            // The catalog and schema are already locked by the parent procedure.
248            // Only lock the table name.
249            return LockKey::new(table_ids);
250        }
251        let mut keys = vec![
252            CatalogLock::Read(table_ref.catalog).into(),
253            SchemaLock::read(table_ref.catalog, table_ref.schema).into(),
254        ];
255        keys.extend(table_ids);
256        LockKey::new(keys)
257    }
258
259    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
260        if !ctx
261            .event_type_filter
262            .allows(RECONCILE_LOGICAL_TABLES_EVENT_TYPE)
263        {
264            return None;
265        }
266
267        let persistent_ctx = &self.context.persistent_ctx;
268        let locators = Self::event_locators(persistent_ctx);
269        let event = match ctx.trigger {
270            EventTrigger::Submitted => {
271                ReconcileLogicalTablesEvent::submitted(locators, persistent_ctx.is_subprocedure)
272            }
273            EventTrigger::Succeeded => self.result_event(locators, true),
274            EventTrigger::Failed | EventTrigger::Poisoned => self.result_event(locators, false),
275            _ => ReconcileLogicalTablesEvent::lifecycle(locators),
276        };
277        Some(Box::new(event))
278    }
279}
280
281impl ReconcileLogicalTablesProcedure {
282    fn event_locators(persistent_ctx: &PersistentContext) -> Vec<ReconciliationLocator> {
283        persistent_ctx
284            .logical_table_ids
285            .iter()
286            .zip(&persistent_ctx.logical_tables)
287            .map(|(table_id, table_name)| {
288                ReconciliationLocator::logical_table(
289                    &table_name.catalog_name,
290                    &table_name.schema_name,
291                    &table_name.table_name,
292                    *table_id,
293                    persistent_ctx.table_id,
294                )
295            })
296            .collect()
297    }
298
299    fn result_event(
300        &self,
301        locators: Vec<ReconciliationLocator>,
302        complete: bool,
303    ) -> ReconcileLogicalTablesEvent {
304        let metrics = self.context.metrics();
305        ReconcileLogicalTablesEvent::result(
306            locators,
307            complete,
308            self.context.persistent_ctx.logical_table_ids.len(),
309            metrics.column_metadata_consistent_count,
310            metrics.column_metadata_inconsistent_count,
311            metrics.create_tables_count,
312            metrics.update_table_info_count,
313        )
314    }
315}
316
317#[async_trait::async_trait]
318#[typetag::serde(tag = "reconcile_logical_tables_state")]
319pub(crate) trait State: Sync + Send + Debug {
320    fn name(&self) -> &'static str {
321        let type_name = std::any::type_name::<Self>();
322        // short name
323        type_name.split("::").last().unwrap_or(type_name)
324    }
325
326    async fn next(
327        &mut self,
328        ctx: &mut ReconcileLogicalTablesContext,
329        procedure_ctx: &ProcedureContext,
330    ) -> Result<(Box<dyn State>, Status)>;
331
332    fn as_any(&self) -> &dyn Any;
333}
334
335#[cfg(test)]
336mod tests {
337    use std::sync::Arc;
338
339    use api::v1::value::ValueData;
340    use common_event_recorder::{EventTypeFilter, EventTypeFilterRef};
341    use common_procedure::{
342        ChildSubmissionOutcome, EventContext, EventTrigger, Procedure, ProcedureId, ProcedureState,
343        RetryPhase,
344    };
345    use serde_json::{Value, json};
346
347    use super::*;
348    use crate::reconciliation::event::RECONCILE_TABLE_EVENT_TYPE;
349    use crate::test_util::{MockDatanodeManager, new_ddl_context};
350
351    struct LogicalTablesEventHarness {
352        procedure_id: ProcedureId,
353        lifecycle_state: ProcedureState,
354        event_type_filter: EventTypeFilterRef,
355    }
356
357    impl LogicalTablesEventHarness {
358        fn all() -> Self {
359            Self {
360                procedure_id: ProcedureId::random(),
361                lifecycle_state: ProcedureState::Running,
362                event_type_filter: Arc::new(EventTypeFilter::All),
363            }
364        }
365
366        fn selected(event_types: impl IntoIterator<Item = &'static str>) -> Self {
367            Self {
368                event_type_filter: Arc::new(EventTypeFilter::Only(
369                    event_types.into_iter().map(str::to_string).collect(),
370                )),
371                ..Self::all()
372            }
373        }
374
375        fn event(
376            &self,
377            procedure: &dyn Procedure,
378            trigger: EventTrigger,
379        ) -> Option<Box<dyn common_event_recorder::Event>> {
380            procedure.event(&EventContext {
381                procedure_id: self.procedure_id,
382                lifecycle_state: &self.lifecycle_state,
383                trigger,
384                event_type_filter: self.event_type_filter.clone(),
385                event_context: None,
386            })
387        }
388    }
389
390    #[test]
391    fn logical_table_submitted_events_cover_root_and_child_intent() {
392        let events = LogicalTablesEventHarness::all();
393        let root = test_procedure(false);
394        let child = test_procedure(true);
395        for (procedure, is_subprocedure) in [(&root, false), (&child, true)] {
396            let submitted = events.event(procedure, EventTrigger::Submitted).unwrap();
397            assert_eq!(submitted.event_type(), RECONCILE_LOGICAL_TABLES_EVENT_TYPE);
398            assert_eq!(
399                submitted.json_payload().unwrap(),
400                json!({
401                    "version": 1,
402                    "logical_table_count": 2,
403                    "is_subprocedure": is_subprocedure,
404                })
405            );
406            let rows = submitted.extra_rows().unwrap();
407            assert_eq!(rows.len(), 2);
408            assert_eq!(
409                rows[0]
410                    .values
411                    .iter()
412                    .map(|value| value.value_data.clone())
413                    .collect::<Vec<_>>(),
414                vec![
415                    Some(ValueData::StringValue("greptime".to_string())),
416                    Some(ValueData::StringValue("public".to_string())),
417                    Some(ValueData::StringValue("cpu".to_string())),
418                    Some(ValueData::U32Value(43)),
419                    Some(ValueData::U32Value(42)),
420                ]
421            );
422            assert_eq!(
423                rows[1]
424                    .values
425                    .iter()
426                    .map(|value| value.value_data.clone())
427                    .collect::<Vec<_>>(),
428                vec![
429                    Some(ValueData::StringValue("greptime".to_string())),
430                    Some(ValueData::StringValue("public".to_string())),
431                    Some(ValueData::StringValue("memory".to_string())),
432                    Some(ValueData::U32Value(44)),
433                    Some(ValueData::U32Value(42)),
434                ]
435            );
436        }
437    }
438
439    #[test]
440    fn logical_table_non_terminal_lifecycle_events_have_null_payloads() {
441        let events = LogicalTablesEventHarness::all();
442        let mut procedure = test_procedure(true);
443        procedure.context.volatile_ctx.metrics = populated_metrics();
444
445        for trigger in [
446            EventTrigger::Recovered,
447            EventTrigger::ChildSubmitted {
448                procedure_id: ProcedureId::random(),
449                outcome: ChildSubmissionOutcome::Accepted,
450            },
451            EventTrigger::Retrying {
452                phase: RetryPhase::Execute,
453                attempt: 2,
454            },
455            EventTrigger::RollingBack,
456        ] {
457            assert_eq!(
458                events
459                    .event(&procedure, trigger)
460                    .unwrap()
461                    .json_payload()
462                    .unwrap(),
463                Value::Null
464            );
465        }
466    }
467
468    #[test]
469    fn logical_table_terminal_events_report_existing_state_and_metrics() {
470        let events = LogicalTablesEventHarness::all();
471        let mut procedure = test_procedure(true);
472        procedure.context.volatile_ctx.metrics = populated_metrics();
473
474        for (trigger, complete) in [
475            (EventTrigger::Succeeded, true),
476            (EventTrigger::Failed, false),
477            (EventTrigger::Poisoned, false),
478        ] {
479            assert_eq!(
480                events
481                    .event(&procedure, trigger)
482                    .unwrap()
483                    .json_payload()
484                    .unwrap(),
485                expected_populated_payload(complete)
486            );
487        }
488    }
489
490    #[test]
491    fn logical_table_event_filtering_uses_the_reconciliation_event_type() {
492        let procedure = test_procedure(false);
493        assert!(
494            LogicalTablesEventHarness::selected([RECONCILE_LOGICAL_TABLES_EVENT_TYPE])
495                .event(&procedure, EventTrigger::Submitted)
496                .is_some()
497        );
498        assert!(
499            LogicalTablesEventHarness::selected([RECONCILE_TABLE_EVENT_TYPE])
500                .event(&procedure, EventTrigger::Submitted)
501                .is_none()
502        );
503        assert!(
504            LogicalTablesEventHarness::selected([])
505                .event(&procedure, EventTrigger::Submitted)
506                .is_none()
507        );
508    }
509
510    #[test]
511    fn logical_table_recovery_preserves_locators_and_resets_metrics() {
512        let events = LogicalTablesEventHarness::all();
513        let mut procedure = test_procedure(false);
514        procedure.context.volatile_ctx.metrics = populated_metrics();
515        let original_dump = procedure.dump().unwrap();
516        procedure.context.volatile_ctx.metrics = ReconcileLogicalTableMetrics::default();
517        assert_eq!(procedure.dump().unwrap(), original_dump);
518
519        let loaded =
520            ReconcileLogicalTablesProcedure::from_json(test_context(), &original_dump).unwrap();
521        assert_eq!(loaded.dump().unwrap(), original_dump);
522        assert_eq!(
523            events
524                .event(&loaded, EventTrigger::Recovered)
525                .unwrap()
526                .extra_rows()
527                .unwrap(),
528            events
529                .event(&procedure, EventTrigger::Submitted)
530                .unwrap()
531                .extra_rows()
532                .unwrap(),
533        );
534        assert_eq!(
535            events
536                .event(&loaded, EventTrigger::Succeeded)
537                .unwrap()
538                .json_payload()
539                .unwrap(),
540            json!({
541                "version": 1,
542                "complete": true,
543                "processed_table_count": 2,
544                "metadata_consistent_table_count": 0,
545                "metadata_inconsistent_table_count": 0,
546                "create_table_count": 0,
547                "update_table_info_count": 0,
548            })
549        );
550    }
551
552    #[test]
553    fn logical_table_failure_before_resolution_reports_requested_tables() {
554        let procedure = test_procedure(false);
555        let payload = LogicalTablesEventHarness::all()
556            .event(&procedure, EventTrigger::Failed)
557            .unwrap()
558            .json_payload()
559            .unwrap();
560        assert_eq!(
561            payload,
562            json!({
563                "version": 1,
564                "complete": false,
565                "processed_table_count": 2,
566                "metadata_consistent_table_count": 0,
567                "metadata_inconsistent_table_count": 0,
568                "create_table_count": 0,
569                "update_table_info_count": 0,
570            })
571        );
572    }
573
574    fn populated_metrics() -> ReconcileLogicalTableMetrics {
575        ReconcileLogicalTableMetrics {
576            column_metadata_consistent_count: 3,
577            column_metadata_inconsistent_count: 1,
578            create_tables_count: 2,
579            update_table_info_count: 4,
580            ..Default::default()
581        }
582    }
583
584    fn expected_populated_payload(complete: bool) -> Value {
585        json!({
586            "version": 1,
587            "complete": complete,
588            "processed_table_count": 2,
589            "metadata_consistent_table_count": 3,
590            "metadata_inconsistent_table_count": 1,
591            "create_table_count": 2,
592            "update_table_info_count": 4,
593        })
594    }
595
596    fn test_procedure(is_subprocedure: bool) -> ReconcileLogicalTablesProcedure {
597        ReconcileLogicalTablesProcedure::new(
598            test_context(),
599            42,
600            TableName::new("greptime", "public", "physical_metrics"),
601            vec![
602                (43, TableName::new("greptime", "public", "cpu")),
603                (44, TableName::new("greptime", "public", "memory")),
604            ],
605            is_subprocedure,
606        )
607    }
608
609    fn test_context() -> Context {
610        let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
611        Context {
612            node_manager: ddl_context.node_manager,
613            table_metadata_manager: ddl_context.table_metadata_manager,
614            cache_invalidator: ddl_context.cache_invalidator,
615        }
616    }
617}
common_meta/reconciliation/reconcile_logical_tables.rs

common_meta/reconciliation/
reconcile_logical_tables.rs
