meta_srv/event/
repartition.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
15use std::any::Any;
16use std::collections::HashMap;
17use std::time::Duration;
18
19use api::v1::value::ValueData;
20use api::v1::{ColumnSchema, Row};
21use common_event_recorder::Event;
22use common_event_recorder::error::{Result, SerializeEventSnafu};
23use common_event_recorder::event_table::{
24    CATALOG_NAME_COLUMN, PARENT_PROCEDURE_ID_COLUMN, REPARTITION_GROUP_ID_COLUMN,
25    SCHEMA_NAME_COLUMN, SOURCE_PARTITION_EXPR_COLUMN, SOURCE_REGION_ID_COLUMN,
26    SOURCE_REGION_NUMBER_COLUMN, TABLE_ID_COLUMN, TABLE_NAME_COLUMN, TARGET_PARTITION_EXPR_COLUMN,
27    TARGET_REGION_ID_COLUMN, TARGET_REGION_NUMBER_COLUMN, column_schemas, nullable_string,
28    nullable_value,
29};
30use serde::Serialize;
31use snafu::ResultExt;
32use store_api::storage::{RegionId, TableId};
33
34use crate::procedure::repartition::PersistentContext as RepartitionPersistentContext;
35use crate::procedure::repartition::group::PersistentContext as GroupPersistentContext;
36use crate::procedure::repartition::plan::{SourceRegionDescriptor, TargetRegionDescriptor};
37use crate::procedure::repartition::repartition_start::RepartitionStart;
38
39pub(crate) const REPARTITION_EVENT_TYPE: &str = "repartition";
40pub(crate) const REPARTITION_GROUP_EVENT_TYPE: &str = "repartition_group";
41
42const PAYLOAD_VERSION: u8 = 2;
43
44#[derive(Debug, Serialize)]
45struct RepartitionSubmittedPayload {
46    version: u8,
47    source_type: &'static str,
48    #[serde(skip_serializing_if = "Vec::is_empty")]
49    source_partition_exprs: Vec<String>,
50    target_partition_exprs: Vec<String>,
51    #[serde(skip_serializing_if = "Option::is_none")]
52    target_partition_columns: Option<Vec<String>>,
53    #[serde(with = "humantime_serde")]
54    timeout: Duration,
55}
56
57#[derive(Debug)]
58pub(crate) struct RepartitionEvent {
59    catalog_name: Option<String>,
60    schema_name: Option<String>,
61    table_name: Option<String>,
62    table_id: Option<TableId>,
63    payload: Option<RepartitionSubmittedPayload>,
64}
65
66impl RepartitionEvent {
67    pub(crate) fn submitted(
68        persistent_ctx: &RepartitionPersistentContext,
69        start: &RepartitionStart,
70    ) -> Self {
71        let intent = start.submitted_intent();
72        Self {
73            catalog_name: Some(persistent_ctx.catalog_name.clone()),
74            schema_name: Some(persistent_ctx.schema_name.clone()),
75            table_name: Some(persistent_ctx.table_name.clone()),
76            table_id: Some(persistent_ctx.table_id),
77            payload: Some(RepartitionSubmittedPayload {
78                version: PAYLOAD_VERSION,
79                source_type: intent.source_type(),
80                source_partition_exprs: intent
81                    .source_partition_exprs()
82                    .iter()
83                    .map(|expr| expr.to_string())
84                    .collect(),
85                target_partition_exprs: intent
86                    .target_partition_exprs()
87                    .iter()
88                    .map(|expr| expr.to_string())
89                    .collect(),
90                target_partition_columns: intent.target_partition_columns().map(ToOwned::to_owned),
91                timeout: persistent_ctx.timeout,
92            }),
93        }
94    }
95
96    pub(crate) fn lifecycle(persistent_ctx: &RepartitionPersistentContext) -> Self {
97        Self {
98            catalog_name: Some(persistent_ctx.catalog_name.clone()),
99            schema_name: Some(persistent_ctx.schema_name.clone()),
100            table_name: Some(persistent_ctx.table_name.clone()),
101            table_id: Some(persistent_ctx.table_id),
102            payload: None,
103        }
104    }
105
106    fn schema() -> Vec<ColumnSchema> {
107        column_schemas([
108            &CATALOG_NAME_COLUMN,
109            &SCHEMA_NAME_COLUMN,
110            &TABLE_NAME_COLUMN,
111            &TABLE_ID_COLUMN,
112        ])
113    }
114}
115
116impl Event for RepartitionEvent {
117    fn event_type(&self) -> &str {
118        REPARTITION_EVENT_TYPE
119    }
120
121    fn json_payload(&self) -> Result<serde_json::Value> {
122        self.payload
123            .as_ref()
124            .map(serde_json::to_value)
125            .transpose()
126            .context(SerializeEventSnafu)
127            .map(|payload| payload.unwrap_or(serde_json::Value::Null))
128    }
129
130    fn extra_schema(&self) -> Vec<ColumnSchema> {
131        Self::schema()
132    }
133
134    fn extra_rows(&self) -> Result<Vec<Row>> {
135        Ok(vec![Row {
136            values: vec![
137                nullable_string(self.catalog_name.as_deref()),
138                nullable_string(self.schema_name.as_deref()),
139                nullable_string(self.table_name.as_deref()),
140                nullable_value(self.table_id.map(ValueData::U32Value)),
141            ],
142        }])
143    }
144
145    fn as_any(&self) -> &dyn Any {
146        self
147    }
148}
149
150#[derive(Debug, Serialize)]
151struct RepartitionGroupSubmittedPayload {
152    version: u8,
153    sync_region: bool,
154    allocated_region_ids: Vec<u64>,
155    pending_deallocate_region_ids: Vec<u64>,
156    #[serde(with = "humantime_serde")]
157    timeout: Duration,
158}
159
160#[derive(Debug)]
161struct RepartitionTopology {
162    sources: Vec<RepartitionTopologySource>,
163    target_partition_exprs: HashMap<RegionId, String>,
164    region_mapping: HashMap<RegionId, Vec<RegionId>>,
165}
166
167#[derive(Debug)]
168struct RepartitionTopologySource {
169    region_id: RegionId,
170    partition_expr: Option<String>,
171}
172
173#[derive(Debug)]
174struct RepartitionTopologyRow<'a> {
175    source_region_id: RegionId,
176    source_partition_expr: Option<&'a str>,
177    target_region_id: RegionId,
178    target_partition_expr: Option<&'a str>,
179}
180
181impl RepartitionTopology {
182    fn new(
183        sources: &[SourceRegionDescriptor],
184        targets: &[TargetRegionDescriptor],
185        region_mapping: &HashMap<RegionId, Vec<RegionId>>,
186    ) -> Self {
187        Self {
188            sources: sources
189                .iter()
190                .map(|source| RepartitionTopologySource {
191                    region_id: source.region_id(),
192                    partition_expr: source.partition_expr().map(|expr| expr.to_string()),
193                })
194                .collect(),
195            target_partition_exprs: targets
196                .iter()
197                .map(|target| (target.region_id, target.partition_expr.to_string()))
198                .collect(),
199            region_mapping: region_mapping.clone(),
200        }
201    }
202
203    fn rows(&self) -> impl Iterator<Item = RepartitionTopologyRow<'_>> {
204        self.sources.iter().flat_map(|source| {
205            self.region_mapping
206                .get(&source.region_id)
207                .into_iter()
208                .flatten()
209                .map(|target_region_id| RepartitionTopologyRow {
210                    source_region_id: source.region_id,
211                    source_partition_expr: source.partition_expr.as_deref(),
212                    target_region_id: *target_region_id,
213                    target_partition_expr: self
214                        .target_partition_exprs
215                        .get(target_region_id)
216                        .map(String::as_str),
217                })
218        })
219    }
220}
221
222#[derive(Debug)]
223pub(crate) struct RepartitionGroupEvent {
224    catalog_name: Option<String>,
225    schema_name: Option<String>,
226    table_name: Option<String>,
227    table_id: Option<TableId>,
228    parent_procedure_id: Option<String>,
229    group_id: Option<String>,
230    topology: Option<RepartitionTopology>,
231    payload: Option<RepartitionGroupSubmittedPayload>,
232}
233
234impl RepartitionGroupEvent {
235    pub(crate) fn submitted(persistent_ctx: &GroupPersistentContext) -> Self {
236        Self {
237            catalog_name: Some(persistent_ctx.catalog_name.clone()),
238            schema_name: Some(persistent_ctx.schema_name.clone()),
239            table_name: persistent_ctx.table_name.clone(),
240            table_id: Some(persistent_ctx.table_id),
241            parent_procedure_id: persistent_ctx.parent_procedure_id.map(|id| id.to_string()),
242            group_id: Some(persistent_ctx.group_id.to_string()),
243            topology: Some(RepartitionTopology::new(
244                &persistent_ctx.sources,
245                &persistent_ctx.targets,
246                &persistent_ctx.region_mapping,
247            )),
248            payload: Some(RepartitionGroupSubmittedPayload {
249                version: PAYLOAD_VERSION,
250                sync_region: persistent_ctx.sync_region,
251                allocated_region_ids: persistent_ctx
252                    .allocated_region_ids
253                    .iter()
254                    .map(|region_id| region_id.as_u64())
255                    .collect(),
256                pending_deallocate_region_ids: persistent_ctx
257                    .pending_deallocate_region_ids
258                    .iter()
259                    .map(|region_id| region_id.as_u64())
260                    .collect(),
261                timeout: persistent_ctx.timeout,
262            }),
263        }
264    }
265
266    pub(crate) fn lifecycle(persistent_ctx: &GroupPersistentContext) -> Self {
267        Self {
268            catalog_name: Some(persistent_ctx.catalog_name.clone()),
269            schema_name: Some(persistent_ctx.schema_name.clone()),
270            table_name: persistent_ctx.table_name.clone(),
271            table_id: Some(persistent_ctx.table_id),
272            parent_procedure_id: persistent_ctx.parent_procedure_id.map(|id| id.to_string()),
273            group_id: Some(persistent_ctx.group_id.to_string()),
274            topology: None,
275            payload: None,
276        }
277    }
278
279    fn extra_row(&self, topology: Option<RepartitionTopologyRow<'_>>) -> Row {
280        let (source_region_id, source_partition_expr, target_region_id, target_partition_expr) =
281            match topology {
282                Some(topology) => (
283                    Some(topology.source_region_id),
284                    topology.source_partition_expr,
285                    Some(topology.target_region_id),
286                    topology.target_partition_expr,
287                ),
288                None => (None, None, None, None),
289            };
290
291        Row {
292            values: vec![
293                nullable_string(self.catalog_name.as_deref()),
294                nullable_string(self.schema_name.as_deref()),
295                nullable_string(self.table_name.as_deref()),
296                nullable_value(self.table_id.map(ValueData::U32Value)),
297                nullable_string(self.parent_procedure_id.as_deref()),
298                nullable_string(self.group_id.as_deref()),
299                nullable_value(source_region_id.map(|id| ValueData::U64Value(id.as_u64()))),
300                nullable_value(source_region_id.map(|id| ValueData::U32Value(id.region_number()))),
301                nullable_string(source_partition_expr),
302                nullable_value(target_region_id.map(|id| ValueData::U64Value(id.as_u64()))),
303                nullable_value(target_region_id.map(|id| ValueData::U32Value(id.region_number()))),
304                nullable_string(target_partition_expr),
305            ],
306        }
307    }
308
309    fn schema() -> Vec<ColumnSchema> {
310        let mut schema = RepartitionEvent::schema();
311        schema.extend(column_schemas([
312            &PARENT_PROCEDURE_ID_COLUMN,
313            &REPARTITION_GROUP_ID_COLUMN,
314            &SOURCE_REGION_ID_COLUMN,
315            &SOURCE_REGION_NUMBER_COLUMN,
316            &SOURCE_PARTITION_EXPR_COLUMN,
317            &TARGET_REGION_ID_COLUMN,
318            &TARGET_REGION_NUMBER_COLUMN,
319            &TARGET_PARTITION_EXPR_COLUMN,
320        ]));
321        schema
322    }
323}
324
325impl Event for RepartitionGroupEvent {
326    fn event_type(&self) -> &str {
327        REPARTITION_GROUP_EVENT_TYPE
328    }
329
330    fn json_payload(&self) -> Result<serde_json::Value> {
331        self.payload
332            .as_ref()
333            .map(serde_json::to_value)
334            .transpose()
335            .context(SerializeEventSnafu)
336            .map(|payload| payload.unwrap_or(serde_json::Value::Null))
337    }
338
339    fn extra_schema(&self) -> Vec<ColumnSchema> {
340        Self::schema()
341    }
342
343    fn extra_rows(&self) -> Result<Vec<Row>> {
344        match &self.topology {
345            Some(topology) => Ok(topology
346                .rows()
347                .map(|row| self.extra_row(Some(row)))
348                .collect()),
349            None => Ok(vec![self.extra_row(None)]),
350        }
351    }
352
353    fn as_any(&self) -> &dyn Any {
354        self
355    }
356}
357
358#[cfg(test)]
359mod tests {
360    use std::collections::HashMap;
361    use std::time::Duration;
362
363    use api::v1::value::ValueData;
364    use api::v1::{ColumnSchema, Row, Value};
365    use common_event_recorder::Event;
366    use common_event_recorder::event_table::{
367        ACTOR_COLUMN, EVENT_CONTEXT_COLUMN, PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN,
368        PROCEDURE_STATE_COLUMN, PROCEDURE_TRIGGER_COLUMN, jsonb_value,
369    };
370    use common_event_recorder::testing::assert_event_contract;
371    use common_procedure::{EventTrigger, ProcedureEvent, ProcedureId, ProcedureState};
372    use table::table_name::TableName;
373    use uuid::Uuid;
374
375    use super::*;
376    use crate::procedure::repartition::repartition_start::RepartitionFrom;
377    use crate::procedure::repartition::test_util::{new_persistent_context, range_expr};
378
379    fn expr(start: i64, end: i64) -> partition::expr::PartitionExpr {
380        range_expr("host", start, end)
381    }
382
383    fn parent_persistent_ctx() -> RepartitionPersistentContext {
384        RepartitionPersistentContext::new(
385            TableName::new("greptime", "public", "repartition_events"),
386            1024,
387            Some(Duration::from_secs(30)),
388        )
389    }
390
391    #[test]
392    fn test_parent_submitted_payload_preserves_repartition_semantics() {
393        let unpartitioned = RepartitionEvent::submitted(
394            &parent_persistent_ctx(),
395            &RepartitionStart::new(
396                RepartitionFrom::Unpartitioned {
397                    partition_columns: vec!["host".to_string()],
398                },
399                vec![expr(0, 100)],
400            ),
401        );
402        let partitioned = RepartitionEvent::submitted(
403            &parent_persistent_ctx(),
404            &RepartitionStart::new(
405                RepartitionFrom::Partitioned {
406                    exprs: vec![expr(0, 100)],
407                    target_partition_columns: Some(vec!["host".to_string()]),
408                },
409                vec![expr(0, 50), expr(50, 100)],
410            ),
411        );
412
413        assert_event_contract(
414            &unpartitioned,
415            REPARTITION_EVENT_TYPE,
416            &parent_schema(),
417            &[Row {
418                values: vec![
419                    ValueData::StringValue("greptime".to_string()).into(),
420                    ValueData::StringValue("public".to_string()).into(),
421                    ValueData::StringValue("repartition_events".to_string()).into(),
422                    ValueData::U32Value(1024).into(),
423                ],
424            }],
425        );
426
427        let unpartitioned_payload = unpartitioned.json_payload().unwrap();
428        let partitioned_payload = partitioned.json_payload().unwrap();
429        assert_eq!(unpartitioned_payload["version"], PAYLOAD_VERSION);
430        assert_eq!(unpartitioned_payload["source_type"], "unpartitioned");
431        assert!(
432            unpartitioned_payload
433                .get("source_partition_exprs")
434                .is_none()
435        );
436        assert_eq!(
437            unpartitioned_payload["target_partition_columns"],
438            serde_json::json!(["host"])
439        );
440        assert_eq!(partitioned_payload["source_type"], "partitioned");
441        assert_eq!(
442            partitioned_payload["source_partition_exprs"],
443            serde_json::json!([expr(0, 100).to_string()])
444        );
445        assert_eq!(
446            partitioned_payload["target_partition_columns"],
447            serde_json::json!(["host"])
448        );
449
450        let merge = RepartitionEvent::submitted(
451            &parent_persistent_ctx(),
452            &RepartitionStart::new(
453                RepartitionFrom::Partitioned {
454                    exprs: vec![expr(0, 50), expr(50, 100)],
455                    target_partition_columns: None,
456                },
457                vec![expr(0, 100)],
458            ),
459        );
460        assert!(
461            merge
462                .json_payload()
463                .unwrap()
464                .get("target_partition_columns")
465                .is_none()
466        );
467    }
468
469    #[test]
470    fn test_topology_rows_expand_one_source_to_multiple_targets() {
471        let table_id = 1024;
472        let source = RegionId::new(table_id, 1);
473        let left = RegionId::new(table_id, 2);
474        let right = RegionId::new(table_id, 3);
475        let source_expr = expr(0, 100);
476        let targets = vec![
477            TargetRegionDescriptor {
478                region_id: left,
479                partition_expr: expr(0, 50),
480            },
481            TargetRegionDescriptor {
482                region_id: right,
483                partition_expr: expr(50, 100),
484            },
485        ];
486        let topology = RepartitionTopology::new(
487            &[SourceRegionDescriptor::partitioned(
488                source,
489                source_expr.clone(),
490            )],
491            &targets,
492            &HashMap::from([(source, vec![left, right])]),
493        );
494        let rows = topology.rows().collect::<Vec<_>>();
495        let source_expr = source_expr.to_string();
496
497        assert_eq!(rows.len(), 2);
498        assert!(rows.iter().all(|row| row.source_region_id == source));
499        assert_eq!(rows[0].source_partition_expr, Some(source_expr.as_str()));
500        assert_eq!(rows[0].target_region_id, left);
501        assert_eq!(rows[1].target_region_id, right);
502    }
503
504    #[test]
505    fn test_group_submitted_event_contract() {
506        let table_id = 1024;
507        let source = RegionId::new(table_id, 1);
508        let left = RegionId::new(table_id, 2);
509        let right = RegionId::new(table_id, 3);
510        let source_expr = expr(0, 100);
511        let left_expr = expr(0, 50);
512        let right_expr = expr(50, 100);
513        let parent_procedure_id =
514            ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
515        let group_id = Uuid::parse_str("00000000-0000-0000-0000-000000000002").unwrap();
516        let mut persistent_ctx = new_persistent_context(
517            table_id,
518            vec![SourceRegionDescriptor::partitioned(
519                source,
520                source_expr.clone(),
521            )],
522            vec![
523                TargetRegionDescriptor {
524                    region_id: left,
525                    partition_expr: left_expr.clone(),
526                },
527                TargetRegionDescriptor {
528                    region_id: right,
529                    partition_expr: right_expr.clone(),
530                },
531            ],
532        );
533        persistent_ctx.parent_procedure_id = Some(parent_procedure_id);
534        persistent_ctx.group_id = group_id;
535        persistent_ctx.region_mapping = HashMap::from([(source, vec![left, right])]);
536
537        let event = RepartitionGroupEvent::submitted(&persistent_ctx);
538
539        assert_event_contract(
540            &event,
541            REPARTITION_GROUP_EVENT_TYPE,
542            &group_schema(),
543            &[
544                group_row(
545                    parent_procedure_id,
546                    group_id,
547                    source,
548                    &source_expr.to_string(),
549                    left,
550                    &left_expr.to_string(),
551                ),
552                group_row(
553                    parent_procedure_id,
554                    group_id,
555                    source,
556                    &source_expr.to_string(),
557                    right,
558                    &right_expr.to_string(),
559                ),
560            ],
561        );
562    }
563
564    #[test]
565    fn test_topology_rows_expand_multiple_sources_to_one_target() {
566        let table_id = 1024;
567        let left = RegionId::new(table_id, 1);
568        let right = RegionId::new(table_id, 2);
569        let merged = RegionId::new(table_id, 3);
570        let topology = RepartitionTopology::new(
571            &[
572                SourceRegionDescriptor::partitioned(left, expr(0, 50)),
573                SourceRegionDescriptor::partitioned(right, expr(50, 100)),
574            ],
575            &[TargetRegionDescriptor {
576                region_id: merged,
577                partition_expr: expr(0, 100),
578            }],
579            &HashMap::from([(left, vec![merged]), (right, vec![merged])]),
580        );
581        let rows = topology.rows().collect::<Vec<_>>();
582
583        assert_eq!(rows.len(), 2);
584        assert_eq!(rows[0].source_region_id, left);
585        assert_eq!(rows[1].source_region_id, right);
586        assert!(rows.iter().all(|row| row.target_region_id == merged));
587    }
588
589    #[test]
590    fn test_topology_rows_keep_default_source_expr_null() {
591        let table_id = 1024;
592        let source = RegionId::new(table_id, 0);
593        let target = RegionId::new(table_id, 1);
594        let topology = RepartitionTopology::new(
595            &[SourceRegionDescriptor::Default { region_id: source }],
596            &[TargetRegionDescriptor {
597                region_id: target,
598                partition_expr: expr(0, 100),
599            }],
600            &HashMap::from([(source, vec![target])]),
601        );
602        let rows = topology.rows().collect::<Vec<_>>();
603        let target_partition_expr = expr(0, 100).to_string();
604
605        assert_eq!(rows.len(), 1);
606        assert_eq!(rows[0].source_partition_expr, None);
607        assert_eq!(rows[0].source_region_id.region_number(), 0);
608        assert_eq!(rows[0].target_region_id.region_number(), 1);
609        assert_eq!(
610            rows[0].target_partition_expr,
611            Some(target_partition_expr.as_str())
612        );
613    }
614
615    #[test]
616    fn test_topology_rows_skip_empty_mapping() {
617        let topology = RepartitionTopology::new(&[], &[], &HashMap::new());
618
619        assert!(topology.rows().next().is_none());
620    }
621
622    #[test]
623    fn test_lifecycle_events_preserve_locators_and_null_payloads() {
624        let parent_ctx = parent_persistent_ctx();
625        let parent = RepartitionEvent::lifecycle(&parent_ctx);
626        assert_event_contract(
627            &parent,
628            REPARTITION_EVENT_TYPE,
629            &parent_schema(),
630            &[Row {
631                values: vec![
632                    ValueData::StringValue("greptime".to_string()).into(),
633                    ValueData::StringValue("public".to_string()).into(),
634                    ValueData::StringValue("repartition_events".to_string()).into(),
635                    ValueData::U32Value(1024).into(),
636                ],
637            }],
638        );
639        assert_eq!(parent.json_payload().unwrap(), serde_json::Value::Null);
640
641        let group_ctx = new_persistent_context(1024, vec![], vec![]);
642        let group = RepartitionGroupEvent::lifecycle(&group_ctx);
643        assert_event_contract(
644            &group,
645            REPARTITION_GROUP_EVENT_TYPE,
646            &group_schema(),
647            &[group.extra_row(None)],
648        );
649        assert_eq!(group.json_payload().unwrap(), serde_json::Value::Null);
650    }
651
652    #[test]
653    fn test_repartition_event_preserves_procedure_envelope_contract() {
654        let procedure_id = ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
655        let event = ProcedureEvent::new(
656            procedure_id,
657            Box::new(RepartitionEvent::submitted(
658                &parent_persistent_ctx(),
659                &RepartitionStart::new(
660                    RepartitionFrom::Unpartitioned {
661                        partition_columns: vec!["host".to_string()],
662                    },
663                    vec![expr(0, 100)],
664                ),
665            )),
666            ProcedureState::Running,
667            EventTrigger::Submitted,
668        );
669        let mut schema = procedure_schema();
670        schema.extend(parent_schema());
671        schema.push(ACTOR_COLUMN.column_schema());
672        schema.push(EVENT_CONTEXT_COLUMN.column_schema());
673
674        assert_event_contract(
675            &event,
676            REPARTITION_EVENT_TYPE,
677            &schema,
678            &[Row {
679                values: vec![
680                    ValueData::StringValue(procedure_id.to_string()).into(),
681                    ValueData::StringValue("Running".to_string()).into(),
682                    ValueData::StringValue(String::new()).into(),
683                    jsonb_value(&serde_json::json!({"type": "Submitted"})),
684                    ValueData::StringValue("greptime".to_string()).into(),
685                    ValueData::StringValue("public".to_string()).into(),
686                    ValueData::StringValue("repartition_events".to_string()).into(),
687                    ValueData::U32Value(1024).into(),
688                    Value { value_data: None },
689                    Value { value_data: None },
690                ],
691            }],
692        );
693    }
694
695    fn parent_schema() -> Vec<ColumnSchema> {
696        column_schemas([
697            &CATALOG_NAME_COLUMN,
698            &SCHEMA_NAME_COLUMN,
699            &TABLE_NAME_COLUMN,
700            &TABLE_ID_COLUMN,
701        ])
702    }
703
704    fn group_schema() -> Vec<ColumnSchema> {
705        let mut schema = parent_schema();
706        schema.extend(column_schemas([
707            &PARENT_PROCEDURE_ID_COLUMN,
708            &REPARTITION_GROUP_ID_COLUMN,
709            &SOURCE_REGION_ID_COLUMN,
710            &SOURCE_REGION_NUMBER_COLUMN,
711            &SOURCE_PARTITION_EXPR_COLUMN,
712            &TARGET_REGION_ID_COLUMN,
713            &TARGET_REGION_NUMBER_COLUMN,
714            &TARGET_PARTITION_EXPR_COLUMN,
715        ]));
716        schema
717    }
718
719    fn procedure_schema() -> Vec<ColumnSchema> {
720        column_schemas([
721            &PROCEDURE_ID_COLUMN,
722            &PROCEDURE_STATE_COLUMN,
723            &PROCEDURE_ERROR_COLUMN,
724            &PROCEDURE_TRIGGER_COLUMN,
725        ])
726    }
727
728    fn group_row(
729        parent_procedure_id: ProcedureId,
730        group_id: Uuid,
731        source_region_id: RegionId,
732        source_partition_expr: &str,
733        target_region_id: RegionId,
734        target_partition_expr: &str,
735    ) -> Row {
736        Row {
737            values: vec![
738                ValueData::StringValue("test_catalog".to_string()).into(),
739                ValueData::StringValue("test_schema".to_string()).into(),
740                ValueData::StringValue("test_table".to_string()).into(),
741                ValueData::U32Value(source_region_id.table_id()).into(),
742                ValueData::StringValue(parent_procedure_id.to_string()).into(),
743                ValueData::StringValue(group_id.to_string()).into(),
744                ValueData::U64Value(source_region_id.as_u64()).into(),
745                ValueData::U32Value(source_region_id.region_number()).into(),
746                ValueData::StringValue(source_partition_expr.to_string()).into(),
747                ValueData::U64Value(target_region_id.as_u64()).into(),
748                ValueData::U32Value(target_region_id.region_number()).into(),
749                ValueData::StringValue(target_partition_expr.to_string()).into(),
750            ],
751        }
752    }
753}
meta_srv/event/repartition.rs

meta_srv/event/
repartition.rs
