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

common_meta/reconciliation/
reconcile_logical_tables.rs
