common_meta/reconciliation/
reconcile_table.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_column_metadata;
19pub(crate) mod update_table_info;
20
21use std::fmt::Debug;
22
23use common_error::ext::ErrorExt;
24use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
25use common_procedure::{
26    Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
27    Procedure, Result as ProcedureResult, Status,
28};
29use serde::{Deserialize, Serialize};
30use snafu::ResultExt;
31use store_api::metadata::ColumnMetadata;
32use store_api::storage::TableId;
33use table::metadata::TableMeta;
34use table::table_name::TableName;
35use tonic::async_trait;
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, TableNameLock};
43use crate::metrics;
44use crate::node_manager::NodeManagerRef;
45use crate::reconciliation::event::{
46    RECONCILE_TABLE_EVENT_TYPE, ReconcileTableEvent, ReconciliationLocator,
47};
48use crate::reconciliation::reconcile_table::reconciliation_start::ReconciliationStart;
49use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy;
50use crate::reconciliation::utils::{
51    Context, ReconcileTableMetrics, build_table_meta_from_column_metadatas,
52};
53
54pub struct ReconcileTableContext {
55    pub node_manager: NodeManagerRef,
56    pub table_metadata_manager: TableMetadataManagerRef,
57    pub cache_invalidator: CacheInvalidatorRef,
58    pub persistent_ctx: PersistentContext,
59    pub volatile_ctx: VolatileContext,
60}
61
62#[derive(Debug, Clone, Copy)]
63enum TablePhase {
64    Start,
65    ResolveColumnMetadata,
66    ReconcileRegions,
67    UpdateTableInfo,
68}
69
70impl TablePhase {
71    const fn as_event_value(self) -> &'static str {
72        match self {
73            Self::Start => "start",
74            Self::ResolveColumnMetadata => "resolve_column_metadata",
75            Self::ReconcileRegions => "reconcile_regions",
76            Self::UpdateTableInfo => "update_table_info",
77        }
78    }
79}
80
81#[derive(Debug, Clone, Copy)]
82enum TableMetadataState {
83    Consistent,
84    Inconsistent,
85}
86
87impl TableMetadataState {
88    const fn as_event_value(self) -> &'static str {
89        match self {
90            Self::Consistent => "consistent",
91            Self::Inconsistent => "inconsistent",
92        }
93    }
94}
95
96#[derive(Debug, Default)]
97struct ReconcileTableResultSummary {
98    metadata_state: Option<TableMetadataState>,
99    resolution_strategy_applied: Option<ResolveStrategy>,
100    resolved_column_count: Option<usize>,
101    scanned_region_count: usize,
102    updated_region_count: usize,
103    table_info_updated: bool,
104    last_completed_phase: Option<TablePhase>,
105}
106
107impl ReconcileTableResultSummary {
108    fn record_scanned_regions(&mut self, scanned_region_count: usize) {
109        self.scanned_region_count = scanned_region_count;
110    }
111
112    fn mark_start_completed(&mut self) {
113        self.last_completed_phase = Some(TablePhase::Start);
114    }
115
116    fn record_metadata_state(&mut self, metadata_state: TableMetadataState) {
117        self.metadata_state = Some(metadata_state);
118    }
119
120    fn record_resolution_strategy(&mut self, resolve_strategy: ResolveStrategy) {
121        self.resolution_strategy_applied = Some(resolve_strategy);
122    }
123
124    fn record_resolved_columns(
125        &mut self,
126        metadata_state: TableMetadataState,
127        resolution_strategy_applied: Option<ResolveStrategy>,
128        resolved_column_count: Option<usize>,
129    ) {
130        self.metadata_state = Some(metadata_state);
131        self.resolution_strategy_applied = resolution_strategy_applied;
132        self.resolved_column_count = resolved_column_count;
133        self.last_completed_phase = Some(TablePhase::ResolveColumnMetadata);
134    }
135
136    fn record_updated_regions(&mut self, updated_region_count: usize) {
137        self.updated_region_count = updated_region_count;
138    }
139
140    fn mark_region_phase_completed(&mut self) {
141        self.last_completed_phase = Some(TablePhase::ReconcileRegions);
142    }
143
144    fn mark_table_info_updated(&mut self) {
145        self.table_info_updated = true;
146    }
147
148    fn mark_table_info_phase_completed(&mut self) {
149        self.last_completed_phase = Some(TablePhase::UpdateTableInfo);
150    }
151
152    fn metadata_state(&self) -> Option<&'static str> {
153        self.metadata_state.map(TableMetadataState::as_event_value)
154    }
155
156    fn last_completed_phase(&self) -> Option<&'static str> {
157        self.last_completed_phase.map(TablePhase::as_event_value)
158    }
159}
160
161impl ReconcileTableContext {
162    /// Creates a new [`ReconcileTableContext`] with the given [`Context`] and [`PersistentContext`].
163    pub fn new(ctx: Context, persistent_ctx: PersistentContext) -> Self {
164        Self {
165            node_manager: ctx.node_manager,
166            table_metadata_manager: ctx.table_metadata_manager,
167            cache_invalidator: ctx.cache_invalidator,
168            persistent_ctx,
169            volatile_ctx: VolatileContext::default(),
170        }
171    }
172
173    /// Returns the physical table name.
174    pub(crate) fn table_name(&self) -> &TableName {
175        &self.persistent_ctx.table_name
176    }
177
178    /// Returns the physical table id.
179    pub(crate) fn table_id(&self) -> TableId {
180        self.persistent_ctx.table_id
181    }
182
183    /// Builds a [`TableMeta`] from the provided [`ColumnMetadata`]s.
184    pub(crate) fn build_table_meta(
185        &self,
186        column_metadatas: &[ColumnMetadata],
187    ) -> Result<TableMeta> {
188        // Safety: The table info value is set in `ReconciliationStart` state.
189        let table_info_value = self.persistent_ctx.table_info_value.as_ref().unwrap();
190        let table_id = self.table_id();
191        let table_ref = self.table_name().table_ref();
192        let name_to_ids = table_info_value.table_info.name_to_ids();
193        let table_meta = build_table_meta_from_column_metadatas(
194            table_id,
195            table_ref,
196            &table_info_value.table_info.meta,
197            name_to_ids,
198            column_metadatas,
199        )?;
200
201        Ok(table_meta)
202    }
203
204    /// Returns a mutable reference to the metrics.
205    pub(crate) fn mut_metrics(&mut self) -> &mut ReconcileTableMetrics {
206        &mut self.volatile_ctx.metrics
207    }
208
209    /// Returns a reference to the metrics.
210    pub(crate) fn metrics(&self) -> &ReconcileTableMetrics {
211        &self.volatile_ctx.metrics
212    }
213}
214
215#[derive(Debug, Serialize, Deserialize)]
216pub(crate) struct PersistentContext {
217    pub(crate) table_id: TableId,
218    pub(crate) table_name: TableName,
219    pub(crate) resolve_strategy: ResolveStrategy,
220    /// The table info value.
221    /// The value will be set in `ReconciliationStart` state.
222    pub(crate) table_info_value: Option<DeserializedValueWithBytes<TableInfoValue>>,
223    // The physical table route.
224    // The value will be set in `ReconciliationStart` state.
225    pub(crate) physical_table_route: Option<PhysicalTableRouteValue>,
226    // Whether the procedure is a subprocedure.
227    pub(crate) is_subprocedure: bool,
228}
229
230impl PersistentContext {
231    pub(crate) fn new(
232        table_id: TableId,
233        table_name: TableName,
234        resolve_strategy: ResolveStrategy,
235        is_subprocedure: bool,
236    ) -> Self {
237        Self {
238            table_id,
239            table_name,
240            resolve_strategy,
241            table_info_value: None,
242            physical_table_route: None,
243            is_subprocedure,
244        }
245    }
246}
247
248#[derive(Default)]
249pub(crate) struct VolatileContext {
250    pub(crate) table_meta: Option<TableMeta>,
251    pub(crate) metrics: ReconcileTableMetrics,
252    result_summary: ReconcileTableResultSummary,
253}
254
255pub struct ReconcileTableProcedure {
256    pub context: ReconcileTableContext,
257    state: Box<dyn State>,
258}
259
260impl ReconcileTableProcedure {
261    /// Creates a new [`ReconcileTableProcedure`] with the given [`Context`] and [`PersistentContext`].
262    pub fn new(
263        ctx: Context,
264        table_id: TableId,
265        table_name: TableName,
266        resolve_strategy: ResolveStrategy,
267        is_subprocedure: bool,
268    ) -> Self {
269        let persistent_ctx =
270            PersistentContext::new(table_id, table_name, resolve_strategy, is_subprocedure);
271        let context = ReconcileTableContext::new(ctx, persistent_ctx);
272        let state = Box::new(ReconciliationStart);
273        Self { context, state }
274    }
275}
276
277impl ReconcileTableProcedure {
278    pub const TYPE_NAME: &'static str = "metasrv-procedure::ReconcileTable";
279
280    pub(crate) fn from_json(ctx: Context, json: &str) -> ProcedureResult<Self> {
281        let ProcedureDataOwned {
282            state,
283            persistent_ctx,
284        } = serde_json::from_str(json).context(FromJsonSnafu)?;
285        let context = ReconcileTableContext::new(ctx, persistent_ctx);
286        Ok(Self { context, state })
287    }
288}
289
290#[derive(Debug, Serialize)]
291struct ProcedureData<'a> {
292    state: &'a dyn State,
293    persistent_ctx: &'a PersistentContext,
294}
295
296#[derive(Debug, Deserialize)]
297struct ProcedureDataOwned {
298    state: Box<dyn State>,
299    persistent_ctx: PersistentContext,
300}
301
302#[async_trait]
303impl Procedure for ReconcileTableProcedure {
304    fn type_name(&self) -> &str {
305        Self::TYPE_NAME
306    }
307
308    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
309        let state = &mut self.state;
310
311        let procedure_name = Self::TYPE_NAME;
312        let step = state.name();
313        let _timer = metrics::METRIC_META_RECONCILIATION_PROCEDURE
314            .with_label_values(&[procedure_name, step])
315            .start_timer();
316        match state.next(&mut self.context, _ctx).await {
317            Ok((next, status)) => {
318                *state = next;
319                Ok(status)
320            }
321            Err(e) => {
322                if e.retry_hint().is_retryable() {
323                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
324                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
325                        .inc();
326                    Err(ProcedureError::retry_later(e))
327                } else {
328                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
329                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_EXTERNAL])
330                        .inc();
331                    Err(ProcedureError::external(e))
332                }
333            }
334        }
335    }
336
337    fn dump(&self) -> ProcedureResult<String> {
338        let data = ProcedureData {
339            state: self.state.as_ref(),
340            persistent_ctx: &self.context.persistent_ctx,
341        };
342        serde_json::to_string(&data).context(ToJsonSnafu)
343    }
344
345    fn lock_key(&self) -> LockKey {
346        let table_ref = &self.context.table_name().table_ref();
347
348        if self.context.persistent_ctx.is_subprocedure {
349            // The catalog and schema are already locked by the parent procedure.
350            // Only lock the table name.
351            return LockKey::new(vec![
352                TableNameLock::new(table_ref.catalog, table_ref.schema, table_ref.table).into(),
353            ]);
354        }
355
356        LockKey::new(vec![
357            CatalogLock::Read(table_ref.catalog).into(),
358            SchemaLock::read(table_ref.catalog, table_ref.schema).into(),
359            TableNameLock::new(table_ref.catalog, table_ref.schema, table_ref.table).into(),
360        ])
361    }
362
363    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
364        if !ctx.event_type_filter.allows(RECONCILE_TABLE_EVENT_TYPE) {
365            return None;
366        }
367
368        let persistent_ctx = &self.context.persistent_ctx;
369        let table_name = &persistent_ctx.table_name;
370        let locator = ReconciliationLocator::physical_table(
371            &table_name.catalog_name,
372            &table_name.schema_name,
373            &table_name.table_name,
374            persistent_ctx.table_id,
375        );
376        let event = match ctx.trigger {
377            EventTrigger::Submitted => ReconcileTableEvent::table_submitted(
378                locator,
379                persistent_ctx.resolve_strategy,
380                persistent_ctx.is_subprocedure,
381            ),
382            EventTrigger::Succeeded => {
383                Self::result_event(locator, &self.context.volatile_ctx.result_summary, true)
384            }
385            EventTrigger::Failed | EventTrigger::Poisoned => {
386                Self::result_event(locator, &self.context.volatile_ctx.result_summary, false)
387            }
388            _ => ReconcileTableEvent::table_lifecycle(locator),
389        };
390        Some(Box::new(event))
391    }
392}
393
394impl ReconcileTableProcedure {
395    fn result_event(
396        locator: ReconciliationLocator,
397        summary: &ReconcileTableResultSummary,
398        complete: bool,
399    ) -> ReconcileTableEvent {
400        ReconcileTableEvent::table_result(
401            locator,
402            complete,
403            summary.metadata_state(),
404            summary.resolution_strategy_applied,
405            summary.resolved_column_count,
406            summary.scanned_region_count,
407            summary.updated_region_count,
408            summary.table_info_updated,
409            summary.last_completed_phase(),
410        )
411    }
412}
413
414#[async_trait::async_trait]
415#[typetag::serde(tag = "reconcile_table_state")]
416pub(crate) trait State: Sync + Send + Debug {
417    fn name(&self) -> &'static str {
418        let type_name = std::any::type_name::<Self>();
419        // short name
420        type_name.split("::").last().unwrap_or(type_name)
421    }
422
423    async fn next(
424        &mut self,
425        ctx: &mut ReconcileTableContext,
426        procedure_ctx: &ProcedureContext,
427    ) -> Result<(Box<dyn State>, Status)>;
428}
429
430#[cfg(test)]
431mod tests {
432    use std::sync::Arc;
433
434    use common_event_recorder::{EventTypeFilter, EventTypeFilterRef};
435    use common_procedure::{
436        ChildSubmissionOutcome, EventContext, EventTrigger, Procedure, ProcedureId, ProcedureState,
437        RetryPhase,
438    };
439    use common_procedure_test::MockContextProvider;
440    use serde_json::{Value, json};
441    use store_api::storage::RegionId;
442
443    use super::*;
444    use crate::ddl::test_util::datanode_handler::PartialSuccessDatanodeHandler;
445    use crate::key::DeserializedValueWithBytes;
446    use crate::key::table_info::TableInfoValue;
447    use crate::key::table_route::PhysicalTableRouteValue;
448    use crate::key::test_utils::new_test_table_info_with_name;
449    use crate::peer::Peer;
450    use crate::reconciliation::reconcile_table::reconcile_regions::ReconcileRegions;
451    use crate::reconciliation::utils::build_column_metadata_from_table_info;
452    use crate::rpc::router::{Region, RegionRoute};
453    use crate::test_util::{MockDatanodeManager, new_ddl_context};
454
455    struct TableEventHarness {
456        procedure_id: ProcedureId,
457        lifecycle_state: ProcedureState,
458        event_type_filter: EventTypeFilterRef,
459    }
460
461    impl TableEventHarness {
462        fn all() -> Self {
463            Self {
464                procedure_id: ProcedureId::random(),
465                lifecycle_state: ProcedureState::Running,
466                event_type_filter: Arc::new(EventTypeFilter::All),
467            }
468        }
469
470        fn selected(event_types: impl IntoIterator<Item = &'static str>) -> Self {
471            Self {
472                event_type_filter: Arc::new(EventTypeFilter::Only(
473                    event_types.into_iter().map(str::to_string).collect(),
474                )),
475                ..Self::all()
476            }
477        }
478
479        fn event(
480            &self,
481            procedure: &dyn Procedure,
482            trigger: EventTrigger,
483        ) -> Option<Box<dyn common_event_recorder::Event>> {
484            procedure.event(&EventContext {
485                procedure_id: self.procedure_id,
486                lifecycle_state: &self.lifecycle_state,
487                trigger,
488                event_type_filter: self.event_type_filter.clone(),
489                event_context: None,
490            })
491        }
492    }
493
494    #[test]
495    fn table_submitted_events_cover_root_and_child_intent() {
496        let events = TableEventHarness::all();
497        let root = test_procedure(false);
498        let child = test_procedure(true);
499        for (procedure, is_subprocedure) in [(&root, false), (&child, true)] {
500            let submitted = events.event(procedure, EventTrigger::Submitted).unwrap();
501            assert_eq!(submitted.event_type(), RECONCILE_TABLE_EVENT_TYPE);
502            assert_eq!(
503                submitted.json_payload().unwrap(),
504                json!({
505                    "version": 1,
506                    "resolve_strategy": "use_latest",
507                    "is_subprocedure": is_subprocedure,
508                })
509            );
510        }
511    }
512
513    #[test]
514    fn table_non_terminal_lifecycle_events_have_null_payloads() {
515        let events = TableEventHarness::all();
516        let mut procedure = test_procedure(true);
517        procedure.context.volatile_ctx.result_summary = populated_summary();
518
519        for trigger in [
520            EventTrigger::Recovered,
521            EventTrigger::ChildSubmitted {
522                procedure_id: ProcedureId::random(),
523                outcome: ChildSubmissionOutcome::Accepted,
524            },
525            EventTrigger::Retrying {
526                phase: RetryPhase::Execute,
527                attempt: 2,
528            },
529            EventTrigger::RollingBack,
530        ] {
531            assert_eq!(
532                events
533                    .event(&procedure, trigger)
534                    .unwrap()
535                    .json_payload()
536                    .unwrap(),
537                Value::Null
538            );
539        }
540    }
541
542    #[test]
543    fn table_terminal_events_report_bounded_results() {
544        let events = TableEventHarness::all();
545        let mut procedure = test_procedure(true);
546        procedure.context.volatile_ctx.result_summary = populated_summary();
547
548        for (trigger, complete) in [
549            (EventTrigger::Succeeded, true),
550            (EventTrigger::Failed, false),
551            (EventTrigger::Poisoned, false),
552        ] {
553            assert_eq!(
554                events
555                    .event(&procedure, trigger)
556                    .unwrap()
557                    .json_payload()
558                    .unwrap(),
559                json!({
560                    "version": 1,
561                    "complete": complete,
562                    "metadata_state": "inconsistent",
563                    "resolution_strategy_applied": "use_latest",
564                    "resolved_column_count": 3,
565                    "scanned_region_count": 2,
566                    "updated_region_count": 1,
567                    "table_info_updated": true,
568                    "last_completed_phase": "update_table_info",
569                })
570            );
571        }
572    }
573
574    #[test]
575    fn table_event_filtering_uses_the_reconciliation_event_type() {
576        let procedure = test_procedure(false);
577        assert!(
578            TableEventHarness::selected([RECONCILE_TABLE_EVENT_TYPE])
579                .event(&procedure, EventTrigger::Submitted)
580                .is_some()
581        );
582        assert!(
583            TableEventHarness::selected(["create_table"])
584                .event(&procedure, EventTrigger::Submitted)
585                .is_none()
586        );
587        assert!(
588            TableEventHarness::selected([])
589                .event(&procedure, EventTrigger::Submitted)
590                .is_none()
591        );
592    }
593
594    #[test]
595    fn table_result_summary_is_not_persisted() {
596        let events = TableEventHarness::all();
597        let mut procedure = test_procedure(false);
598        procedure.context.volatile_ctx.result_summary = populated_summary();
599        let value: Value = serde_json::from_str(&procedure.dump().unwrap()).unwrap();
600        assert!(value["persistent_ctx"].get("result_summary").is_none());
601
602        let loaded = ReconcileTableProcedure::from_json(
603            test_context(),
604            &serde_json::to_string(&value).unwrap(),
605        )
606        .unwrap();
607        assert_eq!(
608            events
609                .event(&loaded, EventTrigger::Succeeded)
610                .unwrap()
611                .json_payload()
612                .unwrap(),
613            json!({
614                "version": 1,
615                "complete": true,
616                "metadata_state": null,
617                "resolution_strategy_applied": null,
618                "resolved_column_count": null,
619                "scanned_region_count": 0,
620                "updated_region_count": 0,
621                "table_info_updated": false,
622                "last_completed_phase": null,
623            })
624        );
625    }
626
627    #[test]
628    fn table_partial_summary_preserves_completed_work_without_advancing_phases() {
629        let events = TableEventHarness::all();
630        let mut procedure = test_procedure(false);
631        let summary = &mut procedure.context.volatile_ctx.result_summary;
632        summary.record_scanned_regions(2);
633
634        assert_eq!(
635            events
636                .event(&procedure, EventTrigger::Failed)
637                .unwrap()
638                .json_payload()
639                .unwrap(),
640            json!({
641                "version": 1,
642                "complete": false,
643                "metadata_state": null,
644                "resolution_strategy_applied": null,
645                "resolved_column_count": null,
646                "scanned_region_count": 2,
647                "updated_region_count": 0,
648                "table_info_updated": false,
649                "last_completed_phase": null,
650            })
651        );
652
653        let summary = &mut procedure.context.volatile_ctx.result_summary;
654        summary.mark_start_completed();
655        summary.record_metadata_state(TableMetadataState::Inconsistent);
656        assert_eq!(
657            events
658                .event(&procedure, EventTrigger::Failed)
659                .unwrap()
660                .json_payload()
661                .unwrap(),
662            json!({
663                "version": 1,
664                "complete": false,
665                "metadata_state": "inconsistent",
666                "resolution_strategy_applied": null,
667                "resolved_column_count": null,
668                "scanned_region_count": 2,
669                "updated_region_count": 0,
670                "table_info_updated": false,
671                "last_completed_phase": "start",
672            })
673        );
674
675        let summary = &mut procedure.context.volatile_ctx.result_summary;
676        summary.record_resolved_columns(
677            TableMetadataState::Inconsistent,
678            Some(ResolveStrategy::UseLatest),
679            Some(3),
680        );
681        summary.record_updated_regions(1);
682        assert_eq!(
683            events
684                .event(&procedure, EventTrigger::Failed)
685                .unwrap()
686                .json_payload()
687                .unwrap(),
688            json!({
689                "version": 1,
690                "complete": false,
691                "metadata_state": "inconsistent",
692                "resolution_strategy_applied": "use_latest",
693                "resolved_column_count": 3,
694                "scanned_region_count": 2,
695                "updated_region_count": 1,
696                "table_info_updated": false,
697                "last_completed_phase": "resolve_column_metadata",
698            })
699        );
700    }
701
702    #[tokio::test]
703    async fn table_mixed_region_outcome_preserves_partial_result() {
704        let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(
705            PartialSuccessDatanodeHandler { retryable: false },
706        )));
707        let context = Context {
708            node_manager: ddl_context.node_manager,
709            table_metadata_manager: ddl_context.table_metadata_manager,
710            cache_invalidator: ddl_context.cache_invalidator,
711        };
712        let mut procedure = ReconcileTableProcedure::new(
713            context,
714            42,
715            TableName::new("greptime", "public", "metrics"),
716            ResolveStrategy::UseLatest,
717            false,
718        );
719
720        let mut table_info = new_test_table_info_with_name(42, "metrics");
721        table_info.meta.column_ids = vec![0, 1, 2];
722        let name_to_ids = table_info.name_to_ids().unwrap();
723        let column_metadatas = build_column_metadata_from_table_info(
724            table_info.meta.schema.column_schemas(),
725            &table_info.meta.primary_key_indices,
726            &name_to_ids,
727        )
728        .unwrap();
729        let region_ids = [RegionId::new(42, 1), RegionId::new(42, 2)];
730        procedure.context.persistent_ctx.table_info_value = Some(
731            DeserializedValueWithBytes::from_inner(TableInfoValue::new(table_info)),
732        );
733        procedure.context.persistent_ctx.physical_table_route =
734            Some(PhysicalTableRouteValue::new(vec![
735                RegionRoute {
736                    region: Region::new_test(region_ids[0]),
737                    leader_peer: Some(Peer::empty(1)),
738                    ..Default::default()
739                },
740                RegionRoute {
741                    region: Region::new_test(region_ids[1]),
742                    leader_peer: Some(Peer::empty(2)),
743                    ..Default::default()
744                },
745            ]));
746
747        let summary = &mut procedure.context.volatile_ctx.result_summary;
748        summary.record_scanned_regions(region_ids.len());
749        summary.mark_start_completed();
750        summary.record_resolved_columns(
751            TableMetadataState::Inconsistent,
752            Some(ResolveStrategy::UseLatest),
753            Some(column_metadatas.len()),
754        );
755        procedure.state = Box::new(ReconcileRegions::new(column_metadatas, region_ids.to_vec()));
756
757        let procedure_ctx = ProcedureContext {
758            procedure_id: ProcedureId::random(),
759            provider: Arc::new(MockContextProvider::default()),
760            event_context: None,
761        };
762        assert!(procedure.execute(&procedure_ctx).await.is_err());
763
764        assert_eq!(
765            TableEventHarness::all()
766                .event(&procedure, EventTrigger::Failed)
767                .unwrap()
768                .json_payload()
769                .unwrap(),
770            json!({
771                "version": 1,
772                "complete": false,
773                "metadata_state": "inconsistent",
774                "resolution_strategy_applied": "use_latest",
775                "resolved_column_count": 3,
776                "scanned_region_count": 2,
777                "updated_region_count": 1,
778                "table_info_updated": false,
779                "last_completed_phase": "resolve_column_metadata",
780            })
781        );
782    }
783
784    fn populated_summary() -> ReconcileTableResultSummary {
785        ReconcileTableResultSummary {
786            metadata_state: Some(TableMetadataState::Inconsistent),
787            resolution_strategy_applied: Some(ResolveStrategy::UseLatest),
788            resolved_column_count: Some(3),
789            scanned_region_count: 2,
790            updated_region_count: 1,
791            table_info_updated: true,
792            last_completed_phase: Some(TablePhase::UpdateTableInfo),
793        }
794    }
795
796    fn test_procedure(is_subprocedure: bool) -> ReconcileTableProcedure {
797        ReconcileTableProcedure::new(
798            test_context(),
799            42,
800            TableName::new("greptime", "public", "metrics"),
801            ResolveStrategy::UseLatest,
802            is_subprocedure,
803        )
804    }
805
806    fn test_context() -> Context {
807        let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
808        Context {
809            node_manager: ddl_context.node_manager,
810            table_metadata_manager: ddl_context.table_metadata_manager,
811            cache_invalidator: ddl_context.cache_invalidator,
812        }
813    }
814}
common_meta/reconciliation/reconcile_table.rs

common_meta/reconciliation/
reconcile_table.rs
