1use std::collections::HashMap;
16use std::fmt::{self, Display};
17use std::time::Duration;
18
19use api::helper::{ColumnDataTypeWrapper, from_pb_time_ranges, from_pb_time_unit};
20use api::v1::add_column_location::LocationType;
21use api::v1::column_def::{
22 as_fulltext_option_analyzer, as_fulltext_option_backend, as_skipping_index_type,
23};
24use api::v1::region::bulk_insert_request::Body;
25use api::v1::region::{
26 AlterRequest, AlterRequests, BuildIndexRequest, BulkInsertRequest,
27 CleanUpRequest as PbCleanUpRequest, CloseRequest, CompactRequest, CreateRequest,
28 CreateRequests, DeleteRequests, DropRequest, DropRequests, FlushRequest, InsertRequests,
29 OpenRequest, TruncateRequest, alter_request, build_index_request, compact_request,
30 region_request, truncate_request,
31};
32use api::v1::{
33 self, Analyzer, ArrowIpc, FulltextBackend as PbFulltextBackend, Option as PbOption, Rows,
34 SemanticType, SkippingIndexType as PbSkippingIndexType, WriteHint,
35};
36use arrow_schema::extension::ExtensionType;
37pub use common_base::AffectedRows;
38use common_base::readable_size::ReadableSize;
39use common_grpc::flight::FlightDecoder;
40use common_recordbatch::DfRecordBatch;
41use common_time::range::TimestampRange;
42use common_time::{TimeToLive, Timestamp};
43use datatypes::error::time_index_not_widening_error;
44use datatypes::extension::json::Json2ExtensionType;
45use datatypes::json::{JsonSettings, JsonTypeHint};
46use datatypes::prelude::ConcreteDataType;
47use datatypes::schema::{FulltextOptions, SkippingIndexOptions};
48use num_enum::TryFromPrimitive;
49use serde::{Deserialize, Deserializer, Serialize, Serializer};
50use snafu::{OptionExt, ResultExt, ensure};
51use strum::{AsRefStr, IntoStaticStr};
52
53use crate::logstore::entry;
54use crate::metadata::{
55 ColumnMetadata, ConvertTimeRangesSnafu, DecodeProtoSnafu, FlightCodecSnafu,
56 InvalidIndexOptionSnafu, InvalidRawRegionRequestSnafu, InvalidRegionRequestSnafu,
57 InvalidSetRegionOptionRequestSnafu, InvalidUnsetRegionOptionRequestSnafu, MetadataError,
58 RegionMetadata, Result, UnexpectedSnafu,
59};
60use crate::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY;
61use crate::metrics;
62use crate::mito_engine_options::{
63 APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, MAX_ROW_GROUP_ROW_COUNT,
64 MAX_ROW_GROUP_ROW_COUNT_LIMIT, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY, TTL_KEY,
65 TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
66 TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM,
67 TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM, WRITE_BUFFER_SIZE_KEY,
68};
69use crate::path_utils::table_dir;
70use crate::storage::{ColumnId, RegionId};
71
72#[derive(Debug, Clone, Copy, PartialEq, TryFromPrimitive)]
74#[repr(u8)]
75pub enum PathType {
76 Bare,
80 Data,
84 Metadata,
88}
89
90#[derive(Debug, IntoStaticStr)]
91pub enum BatchRegionDdlRequest {
92 Create(Vec<(RegionId, RegionCreateRequest)>),
93 Drop(Vec<(RegionId, RegionDropRequest)>),
94 Alter(Vec<(RegionId, RegionAlterRequest)>),
95}
96
97impl BatchRegionDdlRequest {
98 pub fn try_from_request_body(body: region_request::Body) -> Result<Option<Self>> {
100 match body {
101 region_request::Body::Creates(creates) => {
102 let requests = creates
103 .requests
104 .into_iter()
105 .map(parse_region_create)
106 .collect::<Result<Vec<_>>>()?;
107 Ok(Some(Self::Create(requests)))
108 }
109 region_request::Body::Drops(drops) => {
110 let requests = drops
111 .requests
112 .into_iter()
113 .map(parse_region_drop)
114 .collect::<Result<Vec<_>>>()?;
115 Ok(Some(Self::Drop(requests)))
116 }
117 region_request::Body::Alters(alters) => {
118 let requests = alters
119 .requests
120 .into_iter()
121 .map(parse_region_alter)
122 .collect::<Result<Vec<_>>>()?;
123 Ok(Some(Self::Alter(requests)))
124 }
125 _ => Ok(None),
126 }
127 }
128
129 pub fn request_type(&self) -> &'static str {
130 self.into()
131 }
132
133 pub fn into_region_requests(self) -> Vec<(RegionId, RegionRequest)> {
134 match self {
135 Self::Create(requests) => requests
136 .into_iter()
137 .map(|(region_id, request)| (region_id, RegionRequest::Create(request)))
138 .collect(),
139 Self::Drop(requests) => requests
140 .into_iter()
141 .map(|(region_id, request)| (region_id, RegionRequest::Drop(request)))
142 .collect(),
143 Self::Alter(requests) => requests
144 .into_iter()
145 .map(|(region_id, request)| (region_id, RegionRequest::Alter(request)))
146 .collect(),
147 }
148 }
149}
150
151#[derive(Debug, IntoStaticStr)]
152pub enum RegionRequest {
153 Put(RegionPutRequest),
154 Delete(RegionDeleteRequest),
155 Create(RegionCreateRequest),
156 Drop(RegionDropRequest),
157 Open(RegionOpenRequest),
158 CleanUp(RegionCleanUpRequest),
159 Close(RegionCloseRequest),
160 Alter(RegionAlterRequest),
161 Flush(RegionFlushRequest),
162 Compact(RegionCompactRequest),
163 BuildIndex(RegionBuildIndexRequest),
164 Truncate(RegionTruncateRequest),
165 Catchup(RegionCatchupRequest),
166 BulkInserts(RegionBulkInsertsRequest),
167 EnterStaging(EnterStagingRequest),
168 ApplyStagingManifest(ApplyStagingManifestRequest),
169}
170
171impl RegionRequest {
172 pub fn try_from_request_body(body: region_request::Body) -> Result<Vec<(RegionId, Self)>> {
175 match body {
176 region_request::Body::Inserts(inserts) => make_region_puts(inserts),
177 region_request::Body::Deletes(deletes) => make_region_deletes(deletes),
178 region_request::Body::Create(create) => make_region_create(create),
179 region_request::Body::Drop(drop) => make_region_drop(drop),
180 region_request::Body::Open(open) => make_region_open(open),
181 region_request::Body::CleanUp(clean_up) => make_region_clean_up(clean_up),
182 region_request::Body::Close(close) => make_region_close(close),
183 region_request::Body::Alter(alter) => make_region_alter(alter),
184 region_request::Body::Flush(flush) => make_region_flush(flush),
185 region_request::Body::Compact(compact) => make_region_compact(compact),
186 region_request::Body::BuildIndex(index) => make_region_build_index(index),
187 region_request::Body::Truncate(truncate) => make_region_truncate(truncate),
188 region_request::Body::Creates(creates) => make_region_creates(creates),
189 region_request::Body::Drops(drops) => make_region_drops(drops),
190 region_request::Body::Alters(alters) => make_region_alters(alters),
191 region_request::Body::BulkInsert(bulk) => make_region_bulk_inserts(bulk),
192 region_request::Body::Sync(_) => UnexpectedSnafu {
193 reason: "Sync request should be handled separately by RegionServer",
194 }
195 .fail(),
196 region_request::Body::ListMetadata(_) => UnexpectedSnafu {
197 reason: "ListMetadata request should be handled separately by RegionServer",
198 }
199 .fail(),
200 region_request::Body::RemoteDynFilter(_) => UnexpectedSnafu {
201 reason: "RemoteDynFilter request should be handled separately by RegionServer",
202 }
203 .fail(),
204 region_request::Body::ApplyStagingManifest(apply) => {
205 make_region_apply_staging_manifest(apply)
206 }
207 }
208 }
209
210 pub fn request_type(&self) -> &'static str {
212 self.into()
213 }
214}
215
216pub fn make_region_put(
218 request: api::v1::region::InsertRequest,
219) -> Option<(RegionId, RegionPutRequest)> {
220 request.rows.map(|rows| {
221 (
222 request.region_id.into(),
223 RegionPutRequest {
224 rows,
225 hint: None,
226 skip_wal: request.skip_wal,
227 partition_expr_version: request.partition_expr_version.map(|version| version.value),
228 },
229 )
230 })
231}
232
233fn make_region_puts(inserts: InsertRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
234 Ok(inserts
235 .requests
236 .into_iter()
237 .filter_map(make_region_put)
238 .map(|(region_id, request)| (region_id, RegionRequest::Put(request)))
239 .collect())
240}
241
242pub fn make_region_delete(
244 request: api::v1::region::DeleteRequest,
245) -> Option<(RegionId, RegionDeleteRequest)> {
246 request.rows.map(|rows| {
247 (
248 request.region_id.into(),
249 RegionDeleteRequest {
250 rows,
251 hint: None,
252 partition_expr_version: request.partition_expr_version.map(|version| version.value),
253 },
254 )
255 })
256}
257
258fn make_region_deletes(deletes: DeleteRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
259 Ok(deletes
260 .requests
261 .into_iter()
262 .filter_map(make_region_delete)
263 .map(|(region_id, request)| (region_id, RegionRequest::Delete(request)))
264 .collect())
265}
266
267fn parse_region_create(create: CreateRequest) -> Result<(RegionId, RegionCreateRequest)> {
268 let column_metadatas = create
269 .column_defs
270 .into_iter()
271 .map(ColumnMetadata::try_from_column_def)
272 .collect::<Result<Vec<_>>>()?;
273 let region_id = RegionId::from(create.region_id);
274 let table_dir = table_dir(&create.path, region_id.table_id());
275 let partition_expr_json = create.partition.as_ref().map(|p| p.expression.clone());
276 Ok((
277 region_id,
278 RegionCreateRequest {
279 engine: create.engine,
280 column_metadatas,
281 primary_key: create.primary_key,
282 options: create.options,
283 table_dir,
284 path_type: PathType::Bare,
285 partition_expr_json,
286 requirements: create
287 .requirements
288 .map(RegionRequirements::from)
289 .unwrap_or_default(),
290 },
291 ))
292}
293
294fn make_region_create(create: CreateRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
295 let (region_id, request) = parse_region_create(create)?;
296 Ok(vec![(region_id, RegionRequest::Create(request))])
297}
298
299fn make_region_creates(creates: CreateRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
300 let mut requests = Vec::with_capacity(creates.requests.len());
301 for create in creates.requests {
302 requests.extend(make_region_create(create)?);
303 }
304 Ok(requests)
305}
306
307fn parse_region_drop(drop: DropRequest) -> Result<(RegionId, RegionDropRequest)> {
308 let region_id = drop.region_id.into();
309 Ok((
310 region_id,
311 RegionDropRequest {
312 fast_path: drop.fast_path,
313 force: drop.force,
314 partial_drop: drop.partial_drop,
315 },
316 ))
317}
318
319fn make_region_drop(drop: DropRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
320 let (region_id, request) = parse_region_drop(drop)?;
321 Ok(vec![(region_id, RegionRequest::Drop(request))])
322}
323
324fn make_region_drops(drops: DropRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
325 let mut requests = Vec::with_capacity(drops.requests.len());
326 for drop in drops.requests {
327 requests.extend(make_region_drop(drop)?);
328 }
329 Ok(requests)
330}
331
332fn make_region_open(open: OpenRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
333 let region_id = RegionId::from(open.region_id);
334 let table_dir = table_dir(&open.path, region_id.table_id());
335 Ok(vec![(
336 region_id,
337 RegionRequest::Open(RegionOpenRequest {
338 engine: open.engine,
339 table_dir,
340 path_type: PathType::Bare,
341 options: open.options,
342 skip_wal_replay: false,
343 checkpoint: None,
344 requirements: Default::default(),
345 }),
346 )])
347}
348
349fn make_region_clean_up(clean_up: PbCleanUpRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
350 let region_id = RegionId::from(clean_up.region_id);
351 let table_dir = table_dir(&clean_up.path, region_id.table_id());
352 Ok(vec![(
353 region_id,
354 RegionRequest::CleanUp(RegionCleanUpRequest {
355 engine: clean_up.engine,
356 table_dir,
357 path_type: PathType::Bare,
358 options: clean_up.options,
359 }),
360 )])
361}
362
363fn make_region_close(close: CloseRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
364 let region_id = close.region_id.into();
365 Ok(vec![(
366 region_id,
367 RegionRequest::Close(RegionCloseRequest {
368 flush_on_close: close.flush_on_close,
369 }),
370 )])
371}
372
373fn parse_region_alter(alter: AlterRequest) -> Result<(RegionId, RegionAlterRequest)> {
374 let region_id = alter.region_id.into();
375 let request = RegionAlterRequest::try_from(alter)?;
376 Ok((region_id, request))
377}
378
379fn make_region_alter(alter: AlterRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
380 let (region_id, request) = parse_region_alter(alter)?;
381 Ok(vec![(region_id, RegionRequest::Alter(request))])
382}
383
384fn make_region_alters(alters: AlterRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
385 let mut requests = Vec::with_capacity(alters.requests.len());
386 for alter in alters.requests {
387 requests.extend(make_region_alter(alter)?);
388 }
389 Ok(requests)
390}
391
392fn make_region_flush(flush: FlushRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
393 let region_id = flush.region_id.into();
394 Ok(vec![(
395 region_id,
396 RegionRequest::Flush(RegionFlushRequest::default()),
397 )])
398}
399
400fn make_region_compact(compact: CompactRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
401 let region_id = compact.region_id.into();
402 let options = compact
403 .options
404 .unwrap_or(compact_request::Options::Regular(Default::default()));
405 let parallelism = if compact.parallelism == 0 {
407 None
408 } else {
409 Some(compact.parallelism)
410 };
411 let time_range = compact
412 .time_range
413 .map(|range| {
414 let time_unit = v1::TimeUnit::try_from(range.time_unit).map_err(|_| {
415 InvalidRegionRequestSnafu {
416 region_id,
417 err: format!("invalid compaction time unit: {}", range.time_unit),
418 }
419 .build()
420 })?;
421 let time_unit = from_pb_time_unit(time_unit);
422 let start = Timestamp::new(range.start, time_unit);
423 let end = Timestamp::new(range.end, time_unit);
424 ensure!(
425 start < end,
426 InvalidRegionRequestSnafu {
427 region_id,
428 err: format!(
429 "compaction start time must be earlier than end time: {} >= {}",
430 range.start, range.end
431 ),
432 }
433 );
434 TimestampRange::new(start, end).with_context(|| InvalidRegionRequestSnafu {
435 region_id,
436 err: "invalid compaction time range".to_string(),
437 })
438 })
439 .transpose()?;
440 Ok(vec![(
441 region_id,
442 RegionRequest::Compact(RegionCompactRequest {
443 options,
444 parallelism,
445 time_range,
446 }),
447 )])
448}
449
450fn make_region_build_index(index: BuildIndexRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
451 let region_id = index.region_id.into();
452 Ok(vec![(
453 region_id,
454 RegionRequest::BuildIndex(RegionBuildIndexRequest {
455 options: index.options,
456 }),
457 )])
458}
459
460fn make_region_truncate(truncate: TruncateRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
461 let region_id = truncate.region_id.into();
462 match truncate.kind {
463 None => InvalidRawRegionRequestSnafu {
464 err: "missing kind in TruncateRequest".to_string(),
465 }
466 .fail(),
467 Some(truncate_request::Kind::All(_)) => Ok(vec![(
468 region_id,
469 RegionRequest::Truncate(RegionTruncateRequest::All),
470 )]),
471 Some(truncate_request::Kind::TimeRanges(time_ranges)) => {
472 let time_ranges = from_pb_time_ranges(time_ranges).context(ConvertTimeRangesSnafu)?;
473
474 Ok(vec![(
475 region_id,
476 RegionRequest::Truncate(RegionTruncateRequest::ByTimeRanges { time_ranges }),
477 )])
478 }
479 Some(truncate_request::Kind::Unflushed(_)) => Ok(vec![(
480 region_id,
481 RegionRequest::Truncate(RegionTruncateRequest::Unflushed),
482 )]),
483 }
484}
485
486fn make_region_bulk_inserts(request: BulkInsertRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
488 let region_id = request.region_id.into();
489 let skip_wal = request.skip_wal;
490 let partition_expr_version = request.partition_expr_version.map(|v| v.value);
491 let aligned_schema_version = request.aligned_schema_version.map(|v| v.schema_version);
492 let Some(Body::ArrowIpc(request)) = request.body else {
493 return Ok(vec![]);
494 };
495
496 let decoder_timer = metrics::CONVERT_REGION_BULK_REQUEST
497 .with_label_values(&["decode"])
498 .start_timer();
499 let mut decoder =
500 FlightDecoder::try_from_schema_bytes(&request.schema).context(FlightCodecSnafu)?;
501 let payload = decoder
502 .try_decode_record_batch(&request.data_header, &request.payload)
503 .context(FlightCodecSnafu)?;
504 decoder_timer.observe_duration();
505 Ok(vec![(
506 region_id,
507 RegionRequest::BulkInserts(RegionBulkInsertsRequest {
508 region_id,
509 payload,
510 raw_data: request,
511 skip_wal,
512 partition_expr_version,
513 aligned_schema_version,
514 }),
515 )])
516}
517
518fn make_region_apply_staging_manifest(
519 api::v1::region::ApplyStagingManifestRequest {
520 region_id,
521 partition_expr,
522 central_region_id,
523 manifest_path,
524 }: api::v1::region::ApplyStagingManifestRequest,
525) -> Result<Vec<(RegionId, RegionRequest)>> {
526 let region_id = region_id.into();
527 Ok(vec![(
528 region_id,
529 RegionRequest::ApplyStagingManifest(ApplyStagingManifestRequest {
530 partition_expr,
531 central_region_id: central_region_id.into(),
532 manifest_path,
533 }),
534 )])
535}
536
537#[derive(Debug)]
539pub struct RegionPutRequest {
540 pub rows: Rows,
542 pub hint: Option<WriteHint>,
544 pub skip_wal: bool,
547 pub partition_expr_version: Option<u64>,
549}
550
551#[derive(Debug)]
553pub struct RegionDeleteRequest {
554 pub rows: Rows,
558 pub hint: Option<WriteHint>,
560 pub partition_expr_version: Option<u64>,
562}
563
564#[derive(Debug, Clone)]
565pub struct RegionCreateRequest {
566 pub engine: String,
568 pub column_metadatas: Vec<ColumnMetadata>,
570 pub primary_key: Vec<ColumnId>,
572 pub options: HashMap<String, String>,
574 pub table_dir: String,
576 pub path_type: PathType,
578 pub partition_expr_json: Option<String>,
581 pub requirements: RegionRequirements,
583}
584
585impl RegionCreateRequest {
586 pub fn validate(&self) -> Result<()> {
588 ensure!(
590 self.column_metadatas
591 .iter()
592 .any(|x| x.semantic_type == SemanticType::Timestamp),
593 InvalidRegionRequestSnafu {
594 region_id: RegionId::new(0, 0),
595 err: "missing timestamp column in create region request".to_string(),
596 }
597 );
598
599 let mut column_id_to_indices = HashMap::with_capacity(self.column_metadatas.len());
601 for (i, c) in self.column_metadatas.iter().enumerate() {
602 if let Some(previous) = column_id_to_indices.insert(c.column_id, i) {
603 return InvalidRegionRequestSnafu {
604 region_id: RegionId::new(0, 0),
605 err: format!(
606 "duplicate column id {} (at position {} and {}) in create region request",
607 c.column_id, previous, i
608 ),
609 }
610 .fail();
611 }
612 }
613
614 for column_id in &self.primary_key {
616 ensure!(
617 column_id_to_indices.contains_key(column_id),
618 InvalidRegionRequestSnafu {
619 region_id: RegionId::new(0, 0),
620 err: format!(
621 "missing primary key column {} in create region request",
622 column_id
623 ),
624 }
625 );
626 }
627
628 Ok(())
629 }
630
631 pub fn is_physical_table(&self) -> bool {
633 self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
634 }
635}
636
637#[derive(Debug, Clone)]
638pub struct RegionDropRequest {
639 pub fast_path: bool,
642
643 pub force: bool,
646
647 pub partial_drop: bool,
650}
651
652#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
654#[serde(default)]
655pub struct RegionRequirements {
656 pub object_storage: bool,
658}
659
660impl RegionRequirements {
661 pub fn empty() -> Self {
663 Self::default()
664 }
665
666 pub fn object_storage() -> Self {
668 Self {
669 object_storage: true,
670 }
671 }
672}
673
674impl From<api::v1::region::RegionRequirements> for RegionRequirements {
675 fn from(value: api::v1::region::RegionRequirements) -> Self {
676 Self {
677 object_storage: value.object_storage,
678 }
679 }
680}
681
682impl From<RegionRequirements> for api::v1::region::RegionRequirements {
683 fn from(value: RegionRequirements) -> Self {
684 Self {
685 object_storage: value.object_storage,
686 }
687 }
688}
689
690#[derive(Debug, Clone)]
692pub struct RegionOpenRequest {
693 pub engine: String,
695 pub table_dir: String,
697 pub path_type: PathType,
699 pub options: HashMap<String, String>,
701 pub skip_wal_replay: bool,
703 pub checkpoint: Option<ReplayCheckpoint>,
705 pub requirements: RegionRequirements,
707}
708
709#[derive(Debug, Clone, Copy, PartialEq, Eq)]
710pub struct ReplayCheckpoint {
711 pub entry_id: u64,
712 pub metadata_entry_id: Option<u64>,
713}
714
715impl RegionOpenRequest {
716 pub fn is_physical_table(&self) -> bool {
718 self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
719 }
720}
721
722#[derive(Debug, Clone)]
724pub struct RegionCleanUpRequest {
725 pub engine: String,
727 pub table_dir: String,
729 pub path_type: PathType,
731 pub options: HashMap<String, String>,
733}
734
735impl RegionCleanUpRequest {
736 pub fn is_physical_table(&self) -> bool {
738 self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
739 }
740}
741
742#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
744pub struct RegionCloseRequest {
745 pub flush_on_close: bool,
747}
748
749#[derive(Debug, PartialEq, Eq, Clone)]
751pub struct RegionAlterRequest {
752 pub kind: AlterKind,
754}
755
756impl RegionAlterRequest {
757 pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
759 self.kind.validate(metadata)?;
760
761 Ok(())
762 }
763
764 pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
768 debug_assert!(self.validate(metadata).is_ok());
769 self.kind.need_alter(metadata)
770 }
771}
772
773impl TryFrom<AlterRequest> for RegionAlterRequest {
774 type Error = MetadataError;
775
776 fn try_from(value: AlterRequest) -> Result<Self> {
777 let kind = value.kind.context(InvalidRawRegionRequestSnafu {
778 err: "missing kind in AlterRequest",
779 })?;
780
781 let kind = AlterKind::try_from(kind)?;
782 Ok(RegionAlterRequest { kind })
783 }
784}
785
786#[derive(Debug, PartialEq, Eq, Clone, AsRefStr)]
788pub enum AlterKind {
789 AddColumns {
791 columns: Vec<AddColumn>,
793 },
794 DropColumns {
796 names: Vec<String>,
798 },
799 ModifyColumnTypes {
804 columns: Vec<ModifyColumnType>,
806 },
807 SetJsonSettings {
809 column_name: String,
811 settings: JsonSettings,
813 },
814 SetRegionOptions { options: Vec<SetRegionOption> },
816 UnsetRegionOptions { keys: Vec<UnsetRegionOption> },
818 SetIndexes { options: Vec<SetIndexOption> },
820 UnsetIndexes { options: Vec<UnsetIndexOption> },
822 DropDefaults {
824 names: Vec<String>,
826 },
827 SetDefaults {
829 columns: Vec<SetDefault>,
831 },
832 SyncColumns {
834 column_metadatas: Vec<ColumnMetadata>,
835 },
836}
837#[derive(Debug, PartialEq, Eq, Clone)]
838pub struct SetDefault {
839 pub name: String,
840 pub default_constraint: Vec<u8>,
841}
842
843#[derive(Debug, PartialEq, Eq, Clone)]
844pub enum SetIndexOption {
845 Fulltext {
846 column_name: String,
847 options: FulltextOptions,
848 },
849 Inverted {
850 column_name: String,
851 },
852 Skipping {
853 column_name: String,
854 options: SkippingIndexOptions,
855 },
856}
857
858impl SetIndexOption {
859 pub fn column_name(&self) -> &String {
861 match self {
862 SetIndexOption::Fulltext { column_name, .. } => column_name,
863 SetIndexOption::Inverted { column_name } => column_name,
864 SetIndexOption::Skipping { column_name, .. } => column_name,
865 }
866 }
867
868 pub fn is_fulltext(&self) -> bool {
870 match self {
871 SetIndexOption::Fulltext { .. } => true,
872 SetIndexOption::Inverted { .. } => false,
873 SetIndexOption::Skipping { .. } => false,
874 }
875 }
876}
877
878impl TryFrom<v1::SetIndex> for SetIndexOption {
879 type Error = MetadataError;
880
881 fn try_from(value: v1::SetIndex) -> Result<Self> {
882 let option = value.options.context(InvalidRawRegionRequestSnafu {
883 err: "missing options in SetIndex",
884 })?;
885
886 let opt = match option {
887 v1::set_index::Options::Fulltext(x) => SetIndexOption::Fulltext {
888 column_name: x.column_name.clone(),
889 options: FulltextOptions::new(
890 x.enable,
891 as_fulltext_option_analyzer(
892 Analyzer::try_from(x.analyzer).context(DecodeProtoSnafu)?,
893 ),
894 x.case_sensitive,
895 as_fulltext_option_backend(
896 PbFulltextBackend::try_from(x.backend).context(DecodeProtoSnafu)?,
897 ),
898 x.granularity as u32,
899 x.false_positive_rate,
900 )
901 .context(InvalidIndexOptionSnafu)?,
902 },
903 v1::set_index::Options::Inverted(i) => SetIndexOption::Inverted {
904 column_name: i.column_name,
905 },
906 v1::set_index::Options::Skipping(s) => SetIndexOption::Skipping {
907 column_name: s.column_name,
908 options: SkippingIndexOptions::new(
909 s.granularity as u32,
910 s.false_positive_rate,
911 as_skipping_index_type(
912 PbSkippingIndexType::try_from(s.skipping_index_type)
913 .context(DecodeProtoSnafu)?,
914 ),
915 )
916 .context(InvalidIndexOptionSnafu)?,
917 },
918 };
919
920 Ok(opt)
921 }
922}
923
924#[derive(Debug, PartialEq, Eq, Clone)]
925pub enum UnsetIndexOption {
926 Fulltext { column_name: String },
927 Inverted { column_name: String },
928 Skipping { column_name: String },
929}
930
931impl UnsetIndexOption {
932 pub fn column_name(&self) -> &String {
933 match self {
934 UnsetIndexOption::Fulltext { column_name } => column_name,
935 UnsetIndexOption::Inverted { column_name } => column_name,
936 UnsetIndexOption::Skipping { column_name } => column_name,
937 }
938 }
939
940 pub fn is_fulltext(&self) -> bool {
941 match self {
942 UnsetIndexOption::Fulltext { .. } => true,
943 UnsetIndexOption::Inverted { .. } => false,
944 UnsetIndexOption::Skipping { .. } => false,
945 }
946 }
947}
948
949impl TryFrom<v1::UnsetIndex> for UnsetIndexOption {
950 type Error = MetadataError;
951
952 fn try_from(value: v1::UnsetIndex) -> Result<Self> {
953 let option = value.options.context(InvalidRawRegionRequestSnafu {
954 err: "missing options in UnsetIndex",
955 })?;
956
957 let opt = match option {
958 v1::unset_index::Options::Fulltext(f) => UnsetIndexOption::Fulltext {
959 column_name: f.column_name,
960 },
961 v1::unset_index::Options::Inverted(i) => UnsetIndexOption::Inverted {
962 column_name: i.column_name,
963 },
964 v1::unset_index::Options::Skipping(s) => UnsetIndexOption::Skipping {
965 column_name: s.column_name,
966 },
967 };
968
969 Ok(opt)
970 }
971}
972
973impl AlterKind {
974 pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
978 match self {
979 AlterKind::AddColumns { columns } => {
980 for col_to_add in columns {
981 col_to_add.validate(metadata)?;
982 }
983 }
984 AlterKind::DropColumns { names } => {
985 for name in names {
986 Self::validate_column_to_drop(name, metadata)?;
987 }
988 }
989 AlterKind::ModifyColumnTypes { columns } => {
990 for col_to_change in columns {
991 col_to_change.validate(metadata)?;
992 }
993 }
994 AlterKind::SetJsonSettings { column_name, .. } => {
995 Self::validate_set_json_settings(column_name, metadata)?
996 }
997 AlterKind::SetRegionOptions { .. } => {}
998 AlterKind::UnsetRegionOptions { .. } => {}
999 AlterKind::SetIndexes { options } => {
1000 for option in options {
1001 Self::validate_column_alter_index_option(
1002 option.column_name(),
1003 metadata,
1004 option.is_fulltext(),
1005 )?;
1006 }
1007 }
1008 AlterKind::UnsetIndexes { options } => {
1009 for option in options {
1010 Self::validate_column_alter_index_option(
1011 option.column_name(),
1012 metadata,
1013 option.is_fulltext(),
1014 )?;
1015 }
1016 }
1017 AlterKind::DropDefaults { names } => {
1018 names
1019 .iter()
1020 .try_for_each(|name| Self::validate_column_existence(name, metadata))?;
1021 }
1022 AlterKind::SetDefaults { columns } => {
1023 columns
1024 .iter()
1025 .try_for_each(|col| Self::validate_column_existence(&col.name, metadata))?;
1026 }
1027 AlterKind::SyncColumns { column_metadatas } => {
1028 let new_primary_keys = column_metadatas
1029 .iter()
1030 .filter(|c| c.semantic_type == SemanticType::Tag)
1031 .map(|c| (c.column_schema.name.as_str(), c.column_id))
1032 .collect::<HashMap<_, _>>();
1033
1034 let old_primary_keys = metadata
1035 .column_metadatas
1036 .iter()
1037 .filter(|c| c.semantic_type == SemanticType::Tag)
1038 .map(|c| (c.column_schema.name.as_str(), c.column_id));
1039
1040 for (name, id) in old_primary_keys {
1041 let primary_key =
1042 new_primary_keys
1043 .get(name)
1044 .with_context(|| InvalidRegionRequestSnafu {
1045 region_id: metadata.region_id,
1046 err: format!("column {} is not a primary key", name),
1047 })?;
1048
1049 ensure!(
1050 *primary_key == id,
1051 InvalidRegionRequestSnafu {
1052 region_id: metadata.region_id,
1053 err: format!(
1054 "column with same name {} has different id, existing: {}, got: {}",
1055 name, id, primary_key
1056 ),
1057 }
1058 );
1059 }
1060
1061 let new_ts_column = column_metadatas
1062 .iter()
1063 .find(|c| c.semantic_type == SemanticType::Timestamp)
1064 .map(|c| (c.column_schema.name.as_str(), c.column_id))
1065 .context(InvalidRegionRequestSnafu {
1066 region_id: metadata.region_id,
1067 err: "timestamp column not found",
1068 })?;
1069
1070 let old_ts_column = metadata
1072 .column_metadatas
1073 .iter()
1074 .find(|c| c.semantic_type == SemanticType::Timestamp)
1075 .map(|c| (c.column_schema.name.as_str(), c.column_id))
1076 .unwrap();
1077
1078 ensure!(
1079 new_ts_column == old_ts_column,
1080 InvalidRegionRequestSnafu {
1081 region_id: metadata.region_id,
1082 err: format!(
1083 "timestamp column {} has different id, existing: {}, got: {}",
1084 old_ts_column.0, old_ts_column.1, new_ts_column.1
1085 ),
1086 }
1087 );
1088 }
1089 }
1090 Ok(())
1091 }
1092
1093 pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
1095 debug_assert!(self.validate(metadata).is_ok());
1096 match self {
1097 AlterKind::AddColumns { columns } => columns
1098 .iter()
1099 .any(|col_to_add| col_to_add.need_alter(metadata)),
1100 AlterKind::DropColumns { names } => names
1101 .iter()
1102 .any(|name| metadata.column_by_name(name).is_some()),
1103 AlterKind::ModifyColumnTypes { columns } => columns
1104 .iter()
1105 .any(|col_to_change| col_to_change.need_alter(metadata)),
1106 AlterKind::SetJsonSettings {
1107 column_name,
1108 settings,
1109 } => metadata.column_by_name(column_name).is_some_and(|col| {
1110 col.column_schema
1111 .extension_type::<Json2ExtensionType>()
1112 .ok()
1113 .flatten()
1114 .is_none_or(|extension| {
1115 !extension.metadata().json_settings().equivalent(settings)
1116 })
1117 }),
1118 AlterKind::SetRegionOptions { .. } => true,
1119 AlterKind::UnsetRegionOptions { .. } => true,
1120 AlterKind::SetIndexes { options, .. } => options
1121 .iter()
1122 .any(|option| metadata.column_by_name(option.column_name()).is_some()),
1123 AlterKind::UnsetIndexes { options } => options
1124 .iter()
1125 .any(|option| metadata.column_by_name(option.column_name()).is_some()),
1126 AlterKind::DropDefaults { names } => names
1127 .iter()
1128 .any(|name| metadata.column_by_name(name).is_some()),
1129
1130 AlterKind::SetDefaults { columns } => columns
1131 .iter()
1132 .any(|x| metadata.column_by_name(&x.name).is_some()),
1133 AlterKind::SyncColumns { column_metadatas } => {
1134 metadata.column_metadatas != *column_metadatas
1135 }
1136 }
1137 }
1138
1139 fn validate_column_to_drop(name: &str, metadata: &RegionMetadata) -> Result<()> {
1141 let Some(column) = metadata.column_by_name(name) else {
1142 return Ok(());
1143 };
1144 ensure!(
1145 column.semantic_type == SemanticType::Field,
1146 InvalidRegionRequestSnafu {
1147 region_id: metadata.region_id,
1148 err: format!("column {} is not a field and could not be dropped", name),
1149 }
1150 );
1151 Ok(())
1152 }
1153
1154 fn validate_column_alter_index_option(
1156 column_name: &String,
1157 metadata: &RegionMetadata,
1158 is_fulltext: bool,
1159 ) -> Result<()> {
1160 let column = metadata
1161 .column_by_name(column_name)
1162 .context(InvalidRegionRequestSnafu {
1163 region_id: metadata.region_id,
1164 err: format!("column {} not found", column_name),
1165 })?;
1166
1167 if is_fulltext {
1168 ensure!(
1169 column.column_schema.data_type.is_string(),
1170 InvalidRegionRequestSnafu {
1171 region_id: metadata.region_id,
1172 err: format!(
1173 "cannot change alter index options for non-string column {}",
1174 column_name
1175 ),
1176 }
1177 );
1178 }
1179
1180 Ok(())
1181 }
1182
1183 fn validate_column_existence(column_name: &String, metadata: &RegionMetadata) -> Result<()> {
1185 metadata
1186 .column_by_name(column_name)
1187 .context(InvalidRegionRequestSnafu {
1188 region_id: metadata.region_id,
1189 err: format!("column {} not found", column_name),
1190 })?;
1191
1192 Ok(())
1193 }
1194
1195 fn validate_set_json_settings(col_name: &String, metadata: &RegionMetadata) -> Result<()> {
1196 let region_id = metadata.region_id;
1197
1198 let col = metadata
1199 .column_by_name(col_name)
1200 .with_context(|| InvalidRegionRequestSnafu {
1201 region_id,
1202 err: format!("column {} not found", col_name),
1203 })?;
1204
1205 ensure!(
1206 col.semantic_type == SemanticType::Field,
1207 InvalidRegionRequestSnafu {
1208 region_id,
1209 err: format!("column {} is not a field column", col_name),
1210 }
1211 );
1212 ensure!(
1213 col.column_schema.data_type.is_json2(),
1214 InvalidRegionRequestSnafu {
1215 region_id,
1216 err: format!("column {} is not a JSON2 column", col_name),
1217 }
1218 );
1219
1220 Ok(())
1221 }
1222}
1223
1224impl TryFrom<alter_request::Kind> for AlterKind {
1225 type Error = MetadataError;
1226
1227 fn try_from(kind: alter_request::Kind) -> Result<Self> {
1228 let alter_kind = match kind {
1229 alter_request::Kind::AddColumns(x) => {
1230 let columns = x
1231 .add_columns
1232 .into_iter()
1233 .map(|x| x.try_into())
1234 .collect::<Result<Vec<_>>>()?;
1235 AlterKind::AddColumns { columns }
1236 }
1237 alter_request::Kind::ModifyColumnTypes(x) => {
1238 let columns = x
1239 .modify_column_types
1240 .into_iter()
1241 .map(|x| x.into())
1242 .collect::<Vec<_>>();
1243 AlterKind::ModifyColumnTypes { columns }
1244 }
1245 alter_request::Kind::SetJsonSettings(x) => {
1246 let settings = x.settings.context(InvalidRawRegionRequestSnafu {
1247 err: "missing settings in SetJsonSettings",
1248 })?;
1249 AlterKind::SetJsonSettings {
1250 column_name: x.column_name,
1251 settings: json_settings_from_proto(settings)?,
1252 }
1253 }
1254 alter_request::Kind::DropColumns(x) => {
1255 let names = x.drop_columns.into_iter().map(|x| x.name).collect();
1256 AlterKind::DropColumns { names }
1257 }
1258 alter_request::Kind::SetTableOptions(options) => AlterKind::SetRegionOptions {
1259 options: options
1260 .table_options
1261 .iter()
1262 .map(TryFrom::try_from)
1263 .collect::<Result<Vec<_>>>()?,
1264 },
1265 alter_request::Kind::UnsetTableOptions(options) => AlterKind::UnsetRegionOptions {
1266 keys: options
1267 .keys
1268 .iter()
1269 .map(|key| UnsetRegionOption::try_from(key.as_str()))
1270 .collect::<Result<Vec<_>>>()?,
1271 },
1272 alter_request::Kind::SetIndex(o) => AlterKind::SetIndexes {
1273 options: vec![SetIndexOption::try_from(o)?],
1274 },
1275 alter_request::Kind::UnsetIndex(o) => AlterKind::UnsetIndexes {
1276 options: vec![UnsetIndexOption::try_from(o)?],
1277 },
1278 alter_request::Kind::SetIndexes(o) => AlterKind::SetIndexes {
1279 options: o
1280 .set_indexes
1281 .into_iter()
1282 .map(SetIndexOption::try_from)
1283 .collect::<Result<Vec<_>>>()?,
1284 },
1285 alter_request::Kind::UnsetIndexes(o) => AlterKind::UnsetIndexes {
1286 options: o
1287 .unset_indexes
1288 .into_iter()
1289 .map(UnsetIndexOption::try_from)
1290 .collect::<Result<Vec<_>>>()?,
1291 },
1292 alter_request::Kind::DropDefaults(x) => AlterKind::DropDefaults {
1293 names: x.drop_defaults.into_iter().map(|x| x.column_name).collect(),
1294 },
1295 alter_request::Kind::SetDefaults(x) => AlterKind::SetDefaults {
1296 columns: x
1297 .set_defaults
1298 .into_iter()
1299 .map(|x| {
1300 Ok(SetDefault {
1301 name: x.column_name,
1302 default_constraint: x.default_constraint.clone(),
1303 })
1304 })
1305 .collect::<Result<Vec<_>>>()?,
1306 },
1307 alter_request::Kind::SyncColumns(x) => AlterKind::SyncColumns {
1308 column_metadatas: x
1309 .column_defs
1310 .into_iter()
1311 .map(ColumnMetadata::try_from_column_def)
1312 .collect::<Result<Vec<_>>>()?,
1313 },
1314 };
1315
1316 Ok(alter_kind)
1317 }
1318}
1319
1320#[derive(Debug, PartialEq, Eq, Clone)]
1322pub struct AddColumn {
1323 pub column_metadata: ColumnMetadata,
1325 pub location: Option<AddColumnLocation>,
1328}
1329
1330impl AddColumn {
1331 pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
1336 ensure!(
1337 self.column_metadata.column_schema.is_nullable()
1338 || self
1339 .column_metadata
1340 .column_schema
1341 .default_constraint()
1342 .is_some(),
1343 InvalidRegionRequestSnafu {
1344 region_id: metadata.region_id,
1345 err: format!(
1346 "no default value for column {}",
1347 self.column_metadata.column_schema.name
1348 ),
1349 }
1350 );
1351
1352 if let Some(existing_column) =
1353 metadata.column_by_name(&self.column_metadata.column_schema.name)
1354 {
1355 ensure!(
1357 *existing_column == self.column_metadata,
1358 InvalidRegionRequestSnafu {
1359 region_id: metadata.region_id,
1360 err: format!(
1361 "column {} already exists with different metadata, existing: {:?}, got: {:?}",
1362 self.column_metadata.column_schema.name,
1363 existing_column,
1364 self.column_metadata,
1365 ),
1366 }
1367 );
1368 ensure!(
1369 self.location.is_none(),
1370 InvalidRegionRequestSnafu {
1371 region_id: metadata.region_id,
1372 err: format!(
1373 "column {} already exists, but location is specified",
1374 self.column_metadata.column_schema.name
1375 ),
1376 }
1377 );
1378 }
1379
1380 if let Some(existing_column) = metadata.column_by_id(self.column_metadata.column_id) {
1381 ensure!(
1383 existing_column.column_schema.name == self.column_metadata.column_schema.name,
1384 InvalidRegionRequestSnafu {
1385 region_id: metadata.region_id,
1386 err: format!(
1387 "column id {} already exists with different name {}",
1388 self.column_metadata.column_id, existing_column.column_schema.name
1389 ),
1390 }
1391 );
1392 }
1393
1394 Ok(())
1395 }
1396
1397 pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
1399 debug_assert!(self.validate(metadata).is_ok());
1400 metadata
1401 .column_by_name(&self.column_metadata.column_schema.name)
1402 .is_none()
1403 }
1404}
1405
1406impl TryFrom<v1::region::AddColumn> for AddColumn {
1407 type Error = MetadataError;
1408
1409 fn try_from(add_column: v1::region::AddColumn) -> Result<Self> {
1410 let column_def = add_column
1411 .column_def
1412 .context(InvalidRawRegionRequestSnafu {
1413 err: "missing column_def in AddColumn",
1414 })?;
1415
1416 let column_metadata = ColumnMetadata::try_from_column_def(column_def)?;
1417 let location = add_column
1418 .location
1419 .map(AddColumnLocation::try_from)
1420 .transpose()?;
1421
1422 Ok(AddColumn {
1423 column_metadata,
1424 location,
1425 })
1426 }
1427}
1428
1429#[derive(Debug, PartialEq, Eq, Clone)]
1431pub enum AddColumnLocation {
1432 First,
1434 After {
1436 column_name: String,
1438 },
1439}
1440
1441impl TryFrom<v1::AddColumnLocation> for AddColumnLocation {
1442 type Error = MetadataError;
1443
1444 fn try_from(location: v1::AddColumnLocation) -> Result<Self> {
1445 let location_type = LocationType::try_from(location.location_type)
1446 .map_err(|e| InvalidRawRegionRequestSnafu { err: e.to_string() }.build())?;
1447 let add_column_location = match location_type {
1448 LocationType::First => AddColumnLocation::First,
1449 LocationType::After => AddColumnLocation::After {
1450 column_name: location.after_column_name,
1451 },
1452 };
1453
1454 Ok(add_column_location)
1455 }
1456}
1457
1458#[derive(Debug, PartialEq, Eq, Clone)]
1460pub struct ModifyColumnType {
1461 pub column_name: String,
1463 pub target_type: ConcreteDataType,
1465}
1466
1467impl ModifyColumnType {
1468 pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
1470 let column_meta = metadata
1471 .column_by_name(&self.column_name)
1472 .with_context(|| InvalidRegionRequestSnafu {
1473 region_id: metadata.region_id,
1474 err: format!("column {} not found", self.column_name),
1475 })?;
1476
1477 match column_meta.semantic_type {
1478 SemanticType::Field => {
1479 ensure!(
1480 column_meta
1481 .column_schema
1482 .data_type
1483 .can_arrow_type_cast_to(&self.target_type),
1484 InvalidRegionRequestSnafu {
1485 region_id: metadata.region_id,
1486 err: format!(
1487 "column '{}' cannot be cast automatically to type '{}'",
1488 self.column_name, self.target_type
1489 ),
1490 }
1491 );
1492 }
1493 SemanticType::Timestamp => {
1498 ensure!(
1499 column_meta.column_schema.data_type == self.target_type
1500 || column_meta
1501 .column_schema
1502 .data_type
1503 .is_timestamp_unit_widening_to(&self.target_type),
1504 InvalidRegionRequestSnafu {
1505 region_id: metadata.region_id,
1506 err: time_index_not_widening_error(
1507 &column_meta.column_schema.name,
1508 &column_meta.column_schema.data_type,
1509 &self.target_type,
1510 ),
1511 }
1512 );
1513 }
1514 SemanticType::Tag => {
1515 return InvalidRegionRequestSnafu {
1516 region_id: metadata.region_id,
1517 err: format!(
1518 "tag column '{}' cannot change type, it is part of the primary key",
1519 self.column_name
1520 ),
1521 }
1522 .fail();
1523 }
1524 }
1525
1526 Ok(())
1527 }
1528
1529 pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
1533 debug_assert!(self.validate(metadata).is_ok());
1534 metadata
1535 .column_by_name(&self.column_name)
1536 .is_some_and(|column| column.column_schema.data_type != self.target_type)
1537 }
1538}
1539
1540impl From<v1::ModifyColumnType> for ModifyColumnType {
1541 fn from(modify_column_type: v1::ModifyColumnType) -> Self {
1542 let target_type = ColumnDataTypeWrapper::new(
1543 modify_column_type.target_type(),
1544 modify_column_type.target_type_extension,
1545 )
1546 .into();
1547
1548 ModifyColumnType {
1549 column_name: modify_column_type.column_name,
1550 target_type,
1551 }
1552 }
1553}
1554
1555fn json_settings_from_proto(settings: v1::JsonSettings) -> Result<JsonSettings> {
1556 let type_hints = settings
1557 .type_hints
1558 .into_iter()
1559 .map(|hint| {
1560 let wrapper = ColumnDataTypeWrapper::try_new(hint.data_type, hint.datatype_extension)
1561 .map_err(|err| {
1562 InvalidRawRegionRequestSnafu {
1563 err: err.to_string(),
1564 }
1565 .build()
1566 })?;
1567 let data_type = ConcreteDataType::from(wrapper);
1568
1569 Ok(JsonTypeHint {
1570 path: hint.path,
1571 data_type,
1572 inverted_index: false,
1573 })
1574 })
1575 .collect::<Result<Vec<_>>>()?;
1576
1577 JsonSettings::try_new(type_hints, settings.max_auto_expanded_paths).map_err(|err| {
1578 InvalidRawRegionRequestSnafu {
1579 err: err.to_string(),
1580 }
1581 .build()
1582 })
1583}
1584
1585#[derive(Debug, Eq, PartialEq, Clone)]
1590pub enum SetRegionOption {
1591 WriteBufferSize(Option<ReadableSize>),
1592 Ttl(Option<TimeToLive>),
1593 Twsc(String, String),
1595 Format(String),
1597 AppendMode(bool),
1599 AutoFlushInterval(Option<Duration>),
1601 MaxRowGroupRowCount(Option<usize>),
1603 PreserveRowSequence(bool),
1604 SkipWal(bool),
1606}
1607
1608#[derive(Serialize, Deserialize)]
1609enum SetRegionOptionSerde {
1610 WriteBufferSize(Option<ReadableSize>),
1611 Ttl(Option<TimeToLive>),
1612 Twsc(String, String),
1613 Format(String),
1614 AppendMode(bool),
1615 AutoFlushInterval(Option<Duration>),
1616 MaxRowGroupRowCount(Option<usize>),
1617 PreserveRowSequence(bool),
1618 SkipWal(bool),
1619}
1620
1621impl Serialize for SetRegionOption {
1622 fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
1623 where
1624 S: Serializer,
1625 {
1626 if matches!(self, Self::SkipWal(true)) {
1628 return serializer.serialize_unit_variant("SetRegionOption", 8, "SkipWal");
1629 }
1630
1631 let option = match self {
1632 Self::WriteBufferSize(value) => SetRegionOptionSerde::WriteBufferSize(*value),
1633 Self::Ttl(value) => SetRegionOptionSerde::Ttl(*value),
1634 Self::Twsc(key, value) => SetRegionOptionSerde::Twsc(key.clone(), value.clone()),
1635 Self::Format(value) => SetRegionOptionSerde::Format(value.clone()),
1636 Self::AppendMode(value) => SetRegionOptionSerde::AppendMode(*value),
1637 Self::AutoFlushInterval(value) => SetRegionOptionSerde::AutoFlushInterval(*value),
1638 Self::MaxRowGroupRowCount(value) => SetRegionOptionSerde::MaxRowGroupRowCount(*value),
1639 Self::PreserveRowSequence(value) => SetRegionOptionSerde::PreserveRowSequence(*value),
1640 Self::SkipWal(value) => SetRegionOptionSerde::SkipWal(*value),
1641 };
1642 option.serialize(serializer)
1643 }
1644}
1645
1646#[derive(Deserialize)]
1647enum LegacySetRegionOption {
1648 SkipWal,
1649}
1650
1651#[derive(Deserialize)]
1652#[serde(untagged)]
1653enum BackwardCompatibleSetRegionOption {
1654 Current(SetRegionOptionSerde),
1655 Legacy(LegacySetRegionOption),
1656}
1657
1658impl From<SetRegionOptionSerde> for SetRegionOption {
1659 fn from(option: SetRegionOptionSerde) -> Self {
1660 match option {
1661 SetRegionOptionSerde::WriteBufferSize(value) => Self::WriteBufferSize(value),
1662 SetRegionOptionSerde::Ttl(value) => Self::Ttl(value),
1663 SetRegionOptionSerde::Twsc(key, value) => Self::Twsc(key, value),
1664 SetRegionOptionSerde::Format(value) => Self::Format(value),
1665 SetRegionOptionSerde::AppendMode(value) => Self::AppendMode(value),
1666 SetRegionOptionSerde::AutoFlushInterval(value) => Self::AutoFlushInterval(value),
1667 SetRegionOptionSerde::MaxRowGroupRowCount(value) => Self::MaxRowGroupRowCount(value),
1668 SetRegionOptionSerde::PreserveRowSequence(value) => Self::PreserveRowSequence(value),
1669 SetRegionOptionSerde::SkipWal(value) => Self::SkipWal(value),
1670 }
1671 }
1672}
1673
1674impl<'de> Deserialize<'de> for SetRegionOption {
1675 fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
1676 where
1677 D: Deserializer<'de>,
1678 {
1679 Ok(
1680 match BackwardCompatibleSetRegionOption::deserialize(deserializer)? {
1681 BackwardCompatibleSetRegionOption::Current(option) => option.into(),
1682 BackwardCompatibleSetRegionOption::Legacy(LegacySetRegionOption::SkipWal) => {
1683 Self::SkipWal(true)
1684 }
1685 },
1686 )
1687 }
1688}
1689
1690impl TryFrom<&PbOption> for SetRegionOption {
1691 type Error = MetadataError;
1692
1693 fn try_from(value: &PbOption) -> std::result::Result<Self, Self::Error> {
1694 let PbOption { key, value } = value;
1695 match key.as_str() {
1696 WRITE_BUFFER_SIZE_KEY => {
1697 let size = value
1698 .parse::<ReadableSize>()
1699 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1700 Ok(Self::WriteBufferSize(Some(size)))
1701 }
1702 TTL_KEY => {
1703 let ttl = TimeToLive::from_humantime_or_str(value)
1704 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1705
1706 Ok(Self::Ttl(Some(ttl)))
1707 }
1708 TWCS_TRIGGER_FILE_NUM
1709 | TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM
1710 | TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER
1711 | TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM
1712 | TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER
1713 | TWCS_MAX_OUTPUT_FILE_SIZE
1714 | TWCS_TIME_WINDOW => Ok(Self::Twsc(key.clone(), value.clone())),
1715 SST_FORMAT_KEY => Ok(Self::Format(value.clone())),
1716 APPEND_MODE_KEY => {
1717 let append_mode = value
1718 .parse::<bool>()
1719 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1720 Ok(Self::AppendMode(append_mode))
1721 }
1722 AUTO_FLUSH_INTERVAL_KEY => {
1723 if value.is_empty() {
1724 return Ok(Self::AutoFlushInterval(None));
1728 }
1729 let interval = humantime::parse_duration(value)
1730 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1731 if interval <= Duration::ZERO {
1732 return InvalidSetRegionOptionRequestSnafu { key, value }.fail();
1733 }
1734 Ok(Self::AutoFlushInterval(Some(interval)))
1735 }
1736 MAX_ROW_GROUP_ROW_COUNT => {
1737 if value.is_empty() {
1738 return Ok(Self::MaxRowGroupRowCount(None));
1739 }
1740 let row_count = value
1741 .parse::<usize>()
1742 .ok()
1743 .filter(|row_count| {
1744 *row_count > 0 && *row_count <= MAX_ROW_GROUP_ROW_COUNT_LIMIT
1745 })
1746 .ok_or_else(|| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1747 Ok(Self::MaxRowGroupRowCount(Some(row_count)))
1748 }
1749 PRESERVE_ROW_SEQUENCE => {
1750 let preserve = value
1751 .parse::<bool>()
1752 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1753 Ok(Self::PreserveRowSequence(preserve))
1754 }
1755 SKIP_WAL_KEY => {
1756 let skip_wal = value
1757 .parse::<bool>()
1758 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1759 Ok(Self::SkipWal(skip_wal))
1760 }
1761 _ => InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
1762 }
1763 }
1764}
1765
1766impl From<&UnsetRegionOption> for SetRegionOption {
1767 fn from(unset_option: &UnsetRegionOption) -> Self {
1768 match unset_option {
1769 UnsetRegionOption::TwcsTriggerFileNum => {
1770 SetRegionOption::Twsc(unset_option.to_string(), String::new())
1771 }
1772 UnsetRegionOption::TwcsActiveWindowTriggerFileNum => {
1773 SetRegionOption::Twsc(unset_option.to_string(), String::new())
1774 }
1775 UnsetRegionOption::TwcsActiveWindowL1MergeTrigger => {
1776 SetRegionOption::Twsc(unset_option.to_string(), String::new())
1777 }
1778 UnsetRegionOption::TwcsInactiveWindowTriggerFileNum => {
1779 SetRegionOption::Twsc(unset_option.to_string(), String::new())
1780 }
1781 UnsetRegionOption::TwcsInactiveWindowL1MergeTrigger => {
1782 SetRegionOption::Twsc(unset_option.to_string(), String::new())
1783 }
1784 UnsetRegionOption::TwcsMaxOutputFileSize => {
1785 SetRegionOption::Twsc(unset_option.to_string(), String::new())
1786 }
1787 UnsetRegionOption::TwcsTimeWindow => {
1788 SetRegionOption::Twsc(unset_option.to_string(), String::new())
1789 }
1790 UnsetRegionOption::Ttl => SetRegionOption::Ttl(Default::default()),
1791 UnsetRegionOption::MaxRowGroupRowCount => SetRegionOption::MaxRowGroupRowCount(None),
1792 UnsetRegionOption::WriteBufferSize => SetRegionOption::WriteBufferSize(None),
1793 UnsetRegionOption::PreserveRowSequence => SetRegionOption::PreserveRowSequence(false),
1794 }
1795 }
1796}
1797
1798impl TryFrom<&str> for UnsetRegionOption {
1799 type Error = MetadataError;
1800
1801 fn try_from(key: &str) -> Result<Self> {
1802 match key.to_ascii_lowercase().as_str() {
1803 TTL_KEY => Ok(Self::Ttl),
1804 WRITE_BUFFER_SIZE_KEY => Ok(Self::WriteBufferSize),
1805 TWCS_TRIGGER_FILE_NUM => Ok(Self::TwcsTriggerFileNum),
1806 TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsActiveWindowTriggerFileNum),
1807 TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsActiveWindowL1MergeTrigger),
1808 TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsInactiveWindowTriggerFileNum),
1809 TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsInactiveWindowL1MergeTrigger),
1810 TWCS_MAX_OUTPUT_FILE_SIZE => Ok(Self::TwcsMaxOutputFileSize),
1811 TWCS_TIME_WINDOW => Ok(Self::TwcsTimeWindow),
1812 MAX_ROW_GROUP_ROW_COUNT => Ok(Self::MaxRowGroupRowCount),
1813 PRESERVE_ROW_SEQUENCE => Ok(Self::PreserveRowSequence),
1814 _ => InvalidUnsetRegionOptionRequestSnafu { key }.fail(),
1815 }
1816 }
1817}
1818
1819#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
1820pub enum UnsetRegionOption {
1821 TwcsTriggerFileNum,
1822 TwcsActiveWindowTriggerFileNum,
1823 TwcsInactiveWindowTriggerFileNum,
1824 TwcsInactiveWindowL1MergeTrigger,
1825 TwcsMaxOutputFileSize,
1826 TwcsTimeWindow,
1827 Ttl,
1828 MaxRowGroupRowCount,
1829 WriteBufferSize,
1830 PreserveRowSequence,
1831 TwcsActiveWindowL1MergeTrigger,
1832}
1833
1834impl UnsetRegionOption {
1835 pub fn as_str(&self) -> &str {
1836 match self {
1837 Self::Ttl => TTL_KEY,
1838 Self::WriteBufferSize => WRITE_BUFFER_SIZE_KEY,
1839 Self::TwcsTriggerFileNum => TWCS_TRIGGER_FILE_NUM,
1840 Self::TwcsActiveWindowTriggerFileNum => TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
1841 Self::TwcsActiveWindowL1MergeTrigger => TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER,
1842 Self::TwcsInactiveWindowTriggerFileNum => TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM,
1843 Self::TwcsInactiveWindowL1MergeTrigger => TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER,
1844 Self::TwcsMaxOutputFileSize => TWCS_MAX_OUTPUT_FILE_SIZE,
1845 Self::TwcsTimeWindow => TWCS_TIME_WINDOW,
1846 Self::MaxRowGroupRowCount => MAX_ROW_GROUP_ROW_COUNT,
1847 Self::PreserveRowSequence => PRESERVE_ROW_SEQUENCE,
1848 }
1849 }
1850}
1851
1852impl Display for UnsetRegionOption {
1853 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1854 write!(f, "{}", self.as_str())
1855 }
1856}
1857
1858#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1859pub enum RegionFlushReason {
1860 RegionMigration,
1862 Repartition,
1864 RemoteWalPrune,
1866 Closing,
1868 Downgrading,
1870}
1871
1872#[derive(Debug, Clone, Default)]
1873pub struct RegionFlushRequest {
1874 pub row_group_size: Option<usize>,
1875 pub reason: Option<RegionFlushReason>,
1876}
1877
1878#[derive(Debug)]
1879pub struct RegionCompactRequest {
1880 pub options: compact_request::Options,
1881 pub parallelism: Option<u32>,
1882 pub time_range: Option<TimestampRange>,
1883}
1884
1885impl Default for RegionCompactRequest {
1886 fn default() -> Self {
1887 Self {
1888 options: compact_request::Options::Regular(Default::default()),
1890 parallelism: None,
1891 time_range: None,
1892 }
1893 }
1894}
1895
1896#[derive(Debug, Clone, Default)]
1897pub struct RegionBuildIndexRequest {
1898 pub options: Option<build_index_request::Options>,
1900}
1901
1902#[derive(Debug)]
1904pub enum RegionTruncateRequest {
1905 All,
1907 Unflushed,
1917 ByTimeRanges {
1918 time_ranges: Vec<(Timestamp, Timestamp)>,
1922 },
1923}
1924
1925#[derive(Debug, Clone, Copy, Default)]
1930pub struct RegionCatchupRequest {
1931 pub set_writable: bool,
1933 pub entry_id: Option<entry::Id>,
1936 pub metadata_entry_id: Option<entry::Id>,
1940 pub checkpoint: Option<ReplayCheckpoint>,
1942}
1943
1944#[derive(Debug, Clone)]
1945pub struct RegionBulkInsertsRequest {
1946 pub skip_wal: bool,
1948 pub region_id: RegionId,
1949 pub payload: DfRecordBatch,
1950 pub raw_data: ArrowIpc,
1951 pub partition_expr_version: Option<u64>,
1952 pub aligned_schema_version: Option<u64>,
1953}
1954
1955impl RegionBulkInsertsRequest {
1956 pub fn estimated_size(&self) -> usize {
1957 self.payload.get_array_memory_size()
1958 }
1959}
1960
1961#[derive(Debug, Clone, PartialEq, Eq)]
1967pub enum StagingPartitionDirective {
1968 UpdatePartitionExpr(String),
1969 RejectAllWrites,
1970}
1971
1972impl StagingPartitionDirective {
1973 pub fn partition_expr(&self) -> Option<&str> {
1975 match self {
1976 Self::UpdatePartitionExpr(expr) => Some(expr),
1977 Self::RejectAllWrites => None,
1978 }
1979 }
1980}
1981
1982#[derive(Debug, Clone)]
1983pub struct EnterStagingRequest {
1984 pub partition_directive: StagingPartitionDirective,
1986}
1987
1988impl EnterStagingRequest {
1989 pub fn with_partition_expr(partition_expr: String) -> Self {
1991 Self {
1992 partition_directive: StagingPartitionDirective::UpdatePartitionExpr(partition_expr),
1993 }
1994 }
1995}
1996
1997#[derive(Debug, Clone)]
2016pub struct ApplyStagingManifestRequest {
2017 pub partition_expr: String,
2019 pub central_region_id: RegionId,
2021 pub manifest_path: String,
2024}
2025
2026impl fmt::Display for RegionRequest {
2027 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2028 match self {
2029 RegionRequest::Put(_) => write!(f, "Put"),
2030 RegionRequest::Delete(_) => write!(f, "Delete"),
2031 RegionRequest::Create(_) => write!(f, "Create"),
2032 RegionRequest::Drop(_) => write!(f, "Drop"),
2033 RegionRequest::Open(_) => write!(f, "Open"),
2034 RegionRequest::CleanUp(_) => write!(f, "CleanUp"),
2035 RegionRequest::Close(_) => write!(f, "Close"),
2036 RegionRequest::Alter(_) => write!(f, "Alter"),
2037 RegionRequest::Flush(_) => write!(f, "Flush"),
2038 RegionRequest::Compact(_) => write!(f, "Compact"),
2039 RegionRequest::BuildIndex(_) => write!(f, "BuildIndex"),
2040 RegionRequest::Truncate(_) => write!(f, "Truncate"),
2041 RegionRequest::Catchup(_) => write!(f, "Catchup"),
2042 RegionRequest::BulkInserts(_) => write!(f, "BulkInserts"),
2043 RegionRequest::EnterStaging(_) => write!(f, "EnterStaging"),
2044 RegionRequest::ApplyStagingManifest(_) => write!(f, "ApplyStagingManifest"),
2045 }
2046 }
2047}
2048
2049#[cfg(test)]
2050mod tests {
2051
2052 use api::v1::region::RegionColumnDef;
2053 use api::v1::{ColumnDataType, ColumnDef};
2054 use common_time::range::TimestampRange;
2055 use datatypes::extension::json::{Json2ExtensionType, JsonMetadata};
2056 use datatypes::json::JsonSettings;
2057 use datatypes::prelude::ConcreteDataType;
2058 use datatypes::schema::{ColumnSchema, FulltextAnalyzer, FulltextBackend};
2059 use datatypes::types::JsonType;
2060
2061 use super::*;
2062 use crate::metadata::RegionMetadataBuilder;
2063
2064 #[test]
2065 fn test_build_index_options_round_trip() {
2066 use prost::Message;
2067
2068 for options in [
2069 None,
2070 Some(build_index_request::Options::SstIndex(Default::default())),
2071 Some(build_index_request::Options::SeriesIndex(Default::default())),
2072 ] {
2073 let request = BuildIndexRequest {
2074 region_id: 42,
2075 options,
2076 };
2077 let decoded = BuildIndexRequest::decode(request.encode_to_vec().as_slice()).unwrap();
2078 let requests = make_region_build_index(decoded).unwrap();
2079 assert_eq!(1, requests.len());
2080 assert_eq!(RegionId::from_u64(42), requests[0].0);
2081 let RegionRequest::BuildIndex(request) = &requests[0].1 else {
2082 panic!("expected build-index request");
2083 };
2084 assert_eq!(options, request.options);
2085 }
2086 let legacy = BuildIndexRequest::decode(&[0x08, 42][..]).unwrap();
2088 assert!(legacy.options.is_none());
2089 }
2090
2091 #[test]
2092 fn test_make_region_puts_preserves_skip_wal() {
2093 let region_id = RegionId::new(42, 3);
2094 let rows = Rows::default();
2095 let requests = make_region_puts(InsertRequests {
2096 requests: [false, true, false]
2097 .into_iter()
2098 .map(|skip_wal| api::v1::region::InsertRequest {
2099 region_id: region_id.as_u64(),
2100 rows: Some(rows.clone()),
2101 partition_expr_version: Some(api::v1::PartitionExprVersion { value: 7 }),
2102 skip_wal,
2103 })
2104 .collect(),
2105 })
2106 .unwrap();
2107
2108 assert_eq!(3, requests.len());
2109 for ((id, request), skip_wal) in requests.into_iter().zip([false, true, false]) {
2110 assert_eq!(region_id, id);
2111 let RegionRequest::Put(request) = request else {
2112 panic!("expected a put request");
2113 };
2114 assert_eq!(rows, request.rows);
2115 assert_eq!(skip_wal, request.skip_wal);
2116 assert_eq!(Some(7), request.partition_expr_version);
2117 assert!(request.hint.is_none());
2118 }
2119 }
2120
2121 #[test]
2122 fn test_make_region_compact_with_time_range() {
2123 let requests = make_region_compact(CompactRequest {
2124 region_id: 42,
2125 time_range: Some(api::v1::region::CompactionTimeRange {
2126 start: 1_000,
2127 end: 2_000,
2128 time_unit: api::v1::TimeUnit::Microsecond as i32,
2129 }),
2130 ..Default::default()
2131 })
2132 .unwrap();
2133
2134 let RegionRequest::Compact(request) = &requests[0].1 else {
2135 unreachable!();
2136 };
2137 assert_eq!(
2138 Some(
2139 TimestampRange::new(
2140 Timestamp::new_microsecond(1_000),
2141 Timestamp::new_microsecond(2_000),
2142 )
2143 .unwrap()
2144 ),
2145 request.time_range
2146 );
2147 }
2148
2149 #[test]
2150 fn test_make_region_truncate_unflushed() {
2151 let region_id = RegionId::new(42, 3);
2152 let requests =
2153 RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest {
2154 region_id: region_id.as_u64(),
2155 kind: Some(truncate_request::Kind::Unflushed(
2156 api::v1::region::Unflushed {},
2157 )),
2158 }))
2159 .unwrap();
2160
2161 assert_eq!(region_id, requests[0].0);
2162 assert!(matches!(
2163 requests[0].1,
2164 RegionRequest::Truncate(RegionTruncateRequest::Unflushed)
2165 ));
2166 }
2167
2168 #[test]
2169 fn test_make_region_truncate_requires_kind() {
2170 let error =
2171 RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest {
2172 region_id: RegionId::new(42, 3).as_u64(),
2173 kind: None,
2174 }))
2175 .unwrap_err();
2176
2177 assert!(
2178 error
2179 .to_string()
2180 .contains("missing kind in TruncateRequest")
2181 );
2182 }
2183
2184 #[test]
2185 fn test_from_proto_location() {
2186 let proto_location = v1::AddColumnLocation {
2187 location_type: LocationType::First as i32,
2188 after_column_name: String::default(),
2189 };
2190 let location = AddColumnLocation::try_from(proto_location).unwrap();
2191 assert_eq!(location, AddColumnLocation::First);
2192
2193 let proto_location = v1::AddColumnLocation {
2194 location_type: 10,
2195 after_column_name: String::default(),
2196 };
2197 AddColumnLocation::try_from(proto_location).unwrap_err();
2198
2199 let proto_location = v1::AddColumnLocation {
2200 location_type: LocationType::After as i32,
2201 after_column_name: "a".to_string(),
2202 };
2203 let location = AddColumnLocation::try_from(proto_location).unwrap();
2204 assert_eq!(
2205 location,
2206 AddColumnLocation::After {
2207 column_name: "a".to_string()
2208 }
2209 );
2210 }
2211
2212 #[test]
2213 fn test_from_none_proto_add_column() {
2214 AddColumn::try_from(v1::region::AddColumn {
2215 column_def: None,
2216 location: None,
2217 })
2218 .unwrap_err();
2219 }
2220
2221 #[test]
2222 fn test_set_region_option_auto_flush_interval_try_from() {
2223 use std::time::Duration;
2224
2225 let pb = PbOption {
2227 key: "auto_flush_interval".to_string(),
2228 value: "5m".to_string(),
2229 };
2230 let opt = SetRegionOption::try_from(&pb).unwrap();
2231 assert_eq!(
2232 opt,
2233 SetRegionOption::AutoFlushInterval(Some(Duration::from_secs(300)))
2234 );
2235
2236 let pb = PbOption {
2238 key: "auto_flush_interval".to_string(),
2239 value: String::new(),
2240 };
2241 let opt = SetRegionOption::try_from(&pb).unwrap();
2242 assert_eq!(opt, SetRegionOption::AutoFlushInterval(None));
2243
2244 let pb = PbOption {
2246 key: "auto_flush_interval".to_string(),
2247 value: "0s".to_string(),
2248 };
2249 assert!(SetRegionOption::try_from(&pb).is_err());
2250
2251 let pb = PbOption {
2253 key: "auto_flush_interval".to_string(),
2254 value: "not_a_duration".to_string(),
2255 };
2256 assert!(SetRegionOption::try_from(&pb).is_err());
2257 }
2258
2259 #[test]
2260 fn test_set_region_option_skip_wal_try_from() {
2261 let pb = PbOption {
2262 key: SKIP_WAL_KEY.to_string(),
2263 value: "true".to_string(),
2264 };
2265 assert_eq!(
2266 SetRegionOption::SkipWal(true),
2267 SetRegionOption::try_from(&pb).unwrap()
2268 );
2269
2270 let pb = PbOption {
2271 key: SKIP_WAL_KEY.to_string(),
2272 value: "false".to_string(),
2273 };
2274 assert_eq!(
2275 SetRegionOption::SkipWal(false),
2276 SetRegionOption::try_from(&pb).unwrap()
2277 );
2278
2279 for value in ["", "invalid"] {
2280 let pb = PbOption {
2281 key: SKIP_WAL_KEY.to_string(),
2282 value: value.to_string(),
2283 };
2284 assert!(SetRegionOption::try_from(&pb).is_err());
2285 }
2286
2287 assert!(UnsetRegionOption::try_from(SKIP_WAL_KEY).is_err());
2288 }
2289
2290 #[test]
2291 fn test_set_region_option_skip_wal_serde_compatibility() {
2292 let legacy = serde_json::from_str::<SetRegionOption>(r#""SkipWal""#).unwrap();
2293 assert_eq!(SetRegionOption::SkipWal(true), legacy);
2294
2295 check_set_region_option_skip_wal_serde_compatibility(false);
2296 check_set_region_option_skip_wal_serde_compatibility(true);
2297 }
2298
2299 fn check_set_region_option_skip_wal_serde_compatibility(skip_wal: bool) {
2300 let option = SetRegionOption::SkipWal(skip_wal);
2301 let serialized = serde_json::to_string(&option).unwrap();
2302 let expected = if skip_wal {
2303 r#""SkipWal""#
2304 } else {
2305 r#"{"SkipWal":false}"#
2306 };
2307 assert_eq!(expected, serialized);
2308 assert_eq!(
2309 option,
2310 serde_json::from_str::<SetRegionOption>(&serialized).unwrap()
2311 );
2312 if skip_wal {
2313 assert!(serde_json::from_str::<LegacySetRegionOption>(&serialized).is_ok());
2314 assert_eq!(
2315 option,
2316 serde_json::from_str::<SetRegionOption>(r#"{"SkipWal":true}"#).unwrap()
2317 );
2318 }
2319 }
2320
2321 #[test]
2322 fn test_set_region_option_max_row_group_row_count_try_from() {
2323 let pb = PbOption {
2324 key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
2325 value: "512".to_string(),
2326 };
2327 assert_eq!(
2328 SetRegionOption::MaxRowGroupRowCount(Some(512)),
2329 SetRegionOption::try_from(&pb).unwrap()
2330 );
2331
2332 let pb = PbOption {
2333 key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
2334 value: String::new(),
2335 };
2336 assert_eq!(
2337 SetRegionOption::MaxRowGroupRowCount(None),
2338 SetRegionOption::try_from(&pb).unwrap()
2339 );
2340
2341 for value in [
2342 "0".to_string(),
2343 (MAX_ROW_GROUP_ROW_COUNT_LIMIT + 1).to_string(),
2344 "invalid".to_string(),
2345 ] {
2346 let pb = PbOption {
2347 key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
2348 value,
2349 };
2350 assert!(SetRegionOption::try_from(&pb).is_err());
2351 }
2352
2353 assert_eq!(
2354 UnsetRegionOption::MaxRowGroupRowCount,
2355 UnsetRegionOption::try_from(MAX_ROW_GROUP_ROW_COUNT).unwrap()
2356 );
2357 }
2358
2359 #[test]
2360 fn test_set_region_option_preserve_row_sequence_try_from() {
2361 for (value, expected) in [("true", true), ("false", false)] {
2362 let pb = PbOption {
2363 key: PRESERVE_ROW_SEQUENCE.to_string(),
2364 value: value.to_string(),
2365 };
2366 assert_eq!(
2367 SetRegionOption::PreserveRowSequence(expected),
2368 SetRegionOption::try_from(&pb).unwrap()
2369 );
2370 }
2371
2372 for value in ["1", "invalid"] {
2373 let pb = PbOption {
2374 key: PRESERVE_ROW_SEQUENCE.to_string(),
2375 value: value.to_string(),
2376 };
2377 assert!(SetRegionOption::try_from(&pb).is_err());
2378 }
2379
2380 assert_eq!(
2381 UnsetRegionOption::PreserveRowSequence,
2382 UnsetRegionOption::try_from(PRESERVE_ROW_SEQUENCE).unwrap()
2383 );
2384 assert_eq!(
2385 SetRegionOption::PreserveRowSequence(false),
2386 (&UnsetRegionOption::PreserveRowSequence).into()
2387 );
2388 }
2389
2390 #[test]
2391 fn test_set_twcs_window_trigger_options_try_from() {
2392 for key in [
2393 "compaction.twcs.active_window.trigger_file_num",
2394 "compaction.twcs.active_window.l1_merge_trigger",
2395 "compaction.twcs.inactive_window.trigger_file_num",
2396 "compaction.twcs.inactive_window.l1_merge_trigger",
2397 ] {
2398 let option = PbOption {
2399 key: key.to_string(),
2400 value: "8".to_string(),
2401 };
2402 assert_eq!(
2403 SetRegionOption::Twsc(key.to_string(), "8".to_string()),
2404 SetRegionOption::try_from(&option).unwrap(),
2405 "{key}"
2406 );
2407 }
2408 }
2409
2410 #[test]
2411 fn test_unset_twcs_window_trigger_options_try_from() {
2412 for (key, expected) in [
2413 (
2414 "compaction.twcs.active_window.trigger_file_num",
2415 UnsetRegionOption::TwcsActiveWindowTriggerFileNum,
2416 ),
2417 (
2418 "compaction.twcs.active_window.l1_merge_trigger",
2419 UnsetRegionOption::TwcsActiveWindowL1MergeTrigger,
2420 ),
2421 (
2422 "compaction.twcs.inactive_window.trigger_file_num",
2423 UnsetRegionOption::TwcsInactiveWindowTriggerFileNum,
2424 ),
2425 ] {
2426 assert_eq!(expected, UnsetRegionOption::try_from(key).unwrap());
2427 assert_eq!(
2428 SetRegionOption::Twsc(key.to_string(), String::new()),
2429 SetRegionOption::from(&expected)
2430 );
2431 }
2432
2433 let key = "compaction.twcs.inactive_window.l1_merge_trigger";
2434 let option = UnsetRegionOption::try_from(key).unwrap();
2435 assert_eq!(key, option.to_string());
2436 assert_eq!(
2437 SetRegionOption::Twsc(key.to_string(), String::new()),
2438 SetRegionOption::from(&option)
2439 );
2440 }
2441
2442 #[test]
2443 fn test_from_proto_alter_request() {
2444 RegionAlterRequest::try_from(AlterRequest {
2445 region_id: 0,
2446 schema_version: 1,
2447 kind: None,
2448 })
2449 .unwrap_err();
2450
2451 let request = RegionAlterRequest::try_from(AlterRequest {
2452 region_id: 0,
2453 schema_version: 1,
2454 kind: Some(alter_request::Kind::AddColumns(v1::region::AddColumns {
2455 add_columns: vec![v1::region::AddColumn {
2456 column_def: Some(RegionColumnDef {
2457 column_def: Some(ColumnDef {
2458 name: "a".to_string(),
2459 data_type: ColumnDataType::String as i32,
2460 is_nullable: true,
2461 default_constraint: vec![],
2462 semantic_type: SemanticType::Field as i32,
2463 comment: String::new(),
2464 ..Default::default()
2465 }),
2466 column_id: 1,
2467 }),
2468 location: Some(v1::AddColumnLocation {
2469 location_type: LocationType::First as i32,
2470 after_column_name: String::default(),
2471 }),
2472 }],
2473 })),
2474 })
2475 .unwrap();
2476
2477 assert_eq!(
2478 request,
2479 RegionAlterRequest {
2480 kind: AlterKind::AddColumns {
2481 columns: vec![AddColumn {
2482 column_metadata: ColumnMetadata {
2483 column_schema: ColumnSchema::new(
2484 "a",
2485 ConcreteDataType::string_datatype(),
2486 true,
2487 ),
2488 semantic_type: SemanticType::Field,
2489 column_id: 1,
2490 },
2491 location: Some(AddColumnLocation::First),
2492 }]
2493 },
2494 }
2495 );
2496
2497 let request = RegionAlterRequest::try_from(AlterRequest {
2498 region_id: 0,
2499 schema_version: 1,
2500 kind: Some(alter_request::Kind::SetJsonSettings(v1::SetJsonSettings {
2501 column_name: "payload".to_string(),
2502 settings: Some(v1::JsonSettings {
2503 type_hints: vec![v1::JsonTypeHint {
2504 path: vec!["service".to_string()],
2505 data_type: ColumnDataType::String as i32,
2506 datatype_extension: None,
2507 }],
2508 max_auto_expanded_paths: Some(10),
2509 }),
2510 })),
2511 })
2512 .unwrap();
2513
2514 let AlterKind::SetJsonSettings {
2515 column_name,
2516 settings,
2517 } = request.kind
2518 else {
2519 unreachable!()
2520 };
2521 assert_eq!("payload", column_name);
2522 assert_eq!(Some(10), settings.max_auto_expanded_paths());
2523 assert_eq!(1, settings.type_hints().len());
2524 }
2525
2526 #[test]
2527 fn test_write_buffer_size_region_options() {
2528 let option = PbOption {
2529 key: WRITE_BUFFER_SIZE_KEY.to_string(),
2530 value: "128MiB".to_string(),
2531 };
2532 assert_eq!(
2533 SetRegionOption::WriteBufferSize(Some(ReadableSize::mb(128))),
2534 SetRegionOption::try_from(&option).unwrap()
2535 );
2536
2537 let option = PbOption {
2538 key: WRITE_BUFFER_SIZE_KEY.to_string(),
2539 value: "invalid".to_string(),
2540 };
2541 SetRegionOption::try_from(&option).unwrap_err();
2542
2543 let option = PbOption {
2546 key: WRITE_BUFFER_SIZE_KEY.to_string(),
2547 value: String::new(),
2548 };
2549 SetRegionOption::try_from(&option).unwrap_err();
2550
2551 assert_eq!(
2552 UnsetRegionOption::WriteBufferSize,
2553 UnsetRegionOption::try_from(WRITE_BUFFER_SIZE_KEY).unwrap()
2554 );
2555 assert_eq!(
2556 SetRegionOption::WriteBufferSize(None),
2557 SetRegionOption::from(&UnsetRegionOption::WriteBufferSize)
2558 );
2559 }
2560
2561 fn new_metadata() -> RegionMetadata {
2564 let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 1));
2565 builder
2566 .push_column_metadata(ColumnMetadata {
2567 column_schema: ColumnSchema::new(
2568 "ts",
2569 ConcreteDataType::timestamp_millisecond_datatype(),
2570 false,
2571 ),
2572 semantic_type: SemanticType::Timestamp,
2573 column_id: 1,
2574 })
2575 .push_column_metadata(ColumnMetadata {
2576 column_schema: ColumnSchema::new(
2577 "tag_0",
2578 ConcreteDataType::string_datatype(),
2579 true,
2580 ),
2581 semantic_type: SemanticType::Tag,
2582 column_id: 2,
2583 })
2584 .push_column_metadata(ColumnMetadata {
2585 column_schema: ColumnSchema::new(
2586 "field_0",
2587 ConcreteDataType::string_datatype(),
2588 true,
2589 ),
2590 semantic_type: SemanticType::Field,
2591 column_id: 3,
2592 })
2593 .push_column_metadata(ColumnMetadata {
2594 column_schema: ColumnSchema::new(
2595 "field_1",
2596 ConcreteDataType::boolean_datatype(),
2597 true,
2598 ),
2599 semantic_type: SemanticType::Field,
2600 column_id: 4,
2601 })
2602 .primary_key(vec![2]);
2603 builder.build().unwrap()
2604 }
2605
2606 fn json2_column_schema(name: &str, settings: JsonSettings) -> ColumnSchema {
2607 let mut column_schema =
2608 ColumnSchema::new(name, ConcreteDataType::Json(JsonType::null()), true);
2609 column_schema.with_extension_type(&Json2ExtensionType::new(std::sync::Arc::new(
2610 JsonMetadata::new(settings),
2611 )));
2612 column_schema
2613 }
2614
2615 #[test]
2616 fn test_add_column_validate() {
2617 let metadata = new_metadata();
2618 let add_column = AddColumn {
2619 column_metadata: ColumnMetadata {
2620 column_schema: ColumnSchema::new(
2621 "tag_1",
2622 ConcreteDataType::string_datatype(),
2623 true,
2624 ),
2625 semantic_type: SemanticType::Tag,
2626 column_id: 5,
2627 },
2628 location: None,
2629 };
2630 add_column.validate(&metadata).unwrap();
2631 assert!(add_column.need_alter(&metadata));
2632
2633 AddColumn {
2635 column_metadata: ColumnMetadata {
2636 column_schema: ColumnSchema::new(
2637 "tag_1",
2638 ConcreteDataType::string_datatype(),
2639 false,
2640 ),
2641 semantic_type: SemanticType::Tag,
2642 column_id: 5,
2643 },
2644 location: None,
2645 }
2646 .validate(&metadata)
2647 .unwrap_err();
2648
2649 let add_column = AddColumn {
2651 column_metadata: ColumnMetadata {
2652 column_schema: ColumnSchema::new(
2653 "tag_0",
2654 ConcreteDataType::string_datatype(),
2655 true,
2656 ),
2657 semantic_type: SemanticType::Tag,
2658 column_id: 2,
2659 },
2660 location: None,
2661 };
2662 add_column.validate(&metadata).unwrap();
2663 assert!(!add_column.need_alter(&metadata));
2664 }
2665
2666 #[test]
2667 fn test_add_duplicate_columns() {
2668 let kind = AlterKind::AddColumns {
2669 columns: vec![
2670 AddColumn {
2671 column_metadata: ColumnMetadata {
2672 column_schema: ColumnSchema::new(
2673 "tag_1",
2674 ConcreteDataType::string_datatype(),
2675 true,
2676 ),
2677 semantic_type: SemanticType::Tag,
2678 column_id: 5,
2679 },
2680 location: None,
2681 },
2682 AddColumn {
2683 column_metadata: ColumnMetadata {
2684 column_schema: ColumnSchema::new(
2685 "tag_1",
2686 ConcreteDataType::string_datatype(),
2687 true,
2688 ),
2689 semantic_type: SemanticType::Field,
2690 column_id: 6,
2691 },
2692 location: None,
2693 },
2694 ],
2695 };
2696 let metadata = new_metadata();
2697 kind.validate(&metadata).unwrap();
2698 assert!(kind.need_alter(&metadata));
2699 }
2700
2701 #[test]
2702 fn test_add_existing_column_different_metadata() {
2703 let metadata = new_metadata();
2704
2705 let kind = AlterKind::AddColumns {
2707 columns: vec![AddColumn {
2708 column_metadata: ColumnMetadata {
2709 column_schema: ColumnSchema::new(
2710 "tag_0",
2711 ConcreteDataType::string_datatype(),
2712 true,
2713 ),
2714 semantic_type: SemanticType::Tag,
2715 column_id: 4,
2716 },
2717 location: None,
2718 }],
2719 };
2720 kind.validate(&metadata).unwrap_err();
2721
2722 let kind = AlterKind::AddColumns {
2724 columns: vec![AddColumn {
2725 column_metadata: ColumnMetadata {
2726 column_schema: ColumnSchema::new(
2727 "tag_0",
2728 ConcreteDataType::int64_datatype(),
2729 true,
2730 ),
2731 semantic_type: SemanticType::Tag,
2732 column_id: 2,
2733 },
2734 location: None,
2735 }],
2736 };
2737 kind.validate(&metadata).unwrap_err();
2738
2739 let kind = AlterKind::AddColumns {
2741 columns: vec![AddColumn {
2742 column_metadata: ColumnMetadata {
2743 column_schema: ColumnSchema::new(
2744 "tag_1",
2745 ConcreteDataType::string_datatype(),
2746 true,
2747 ),
2748 semantic_type: SemanticType::Tag,
2749 column_id: 2,
2750 },
2751 location: None,
2752 }],
2753 };
2754 kind.validate(&metadata).unwrap_err();
2755 }
2756
2757 #[test]
2758 fn test_add_existing_column_with_location() {
2759 let metadata = new_metadata();
2760 let kind = AlterKind::AddColumns {
2761 columns: vec![AddColumn {
2762 column_metadata: ColumnMetadata {
2763 column_schema: ColumnSchema::new(
2764 "tag_0",
2765 ConcreteDataType::string_datatype(),
2766 true,
2767 ),
2768 semantic_type: SemanticType::Tag,
2769 column_id: 2,
2770 },
2771 location: Some(AddColumnLocation::First),
2772 }],
2773 };
2774 kind.validate(&metadata).unwrap_err();
2775 }
2776
2777 #[test]
2778 fn test_validate_drop_column() {
2779 let metadata = new_metadata();
2780 let kind = AlterKind::DropColumns {
2781 names: vec!["xxxx".to_string()],
2782 };
2783 kind.validate(&metadata).unwrap();
2784 assert!(!kind.need_alter(&metadata));
2785
2786 AlterKind::DropColumns {
2787 names: vec!["tag_0".to_string()],
2788 }
2789 .validate(&metadata)
2790 .unwrap_err();
2791
2792 let kind = AlterKind::DropColumns {
2793 names: vec!["field_0".to_string()],
2794 };
2795 kind.validate(&metadata).unwrap();
2796 assert!(kind.need_alter(&metadata));
2797 }
2798
2799 #[test]
2800 fn test_validate_modify_column_type() {
2801 let metadata = new_metadata();
2802 AlterKind::ModifyColumnTypes {
2803 columns: vec![ModifyColumnType {
2804 column_name: "xxxx".to_string(),
2805 target_type: ConcreteDataType::string_datatype(),
2806 }],
2807 }
2808 .validate(&metadata)
2809 .unwrap_err();
2810
2811 AlterKind::ModifyColumnTypes {
2812 columns: vec![ModifyColumnType {
2813 column_name: "field_1".to_string(),
2814 target_type: ConcreteDataType::date_datatype(),
2815 }],
2816 }
2817 .validate(&metadata)
2818 .unwrap_err();
2819
2820 AlterKind::ModifyColumnTypes {
2821 columns: vec![ModifyColumnType {
2822 column_name: "ts".to_string(),
2823 target_type: ConcreteDataType::date_datatype(),
2824 }],
2825 }
2826 .validate(&metadata)
2827 .unwrap_err();
2828
2829 let kind = AlterKind::ModifyColumnTypes {
2831 columns: vec![ModifyColumnType {
2832 column_name: "ts".to_string(),
2833 target_type: ConcreteDataType::timestamp_microsecond_datatype(),
2834 }],
2835 };
2836 kind.validate(&metadata).unwrap();
2837 assert!(kind.need_alter(&metadata));
2838
2839 let metadata_nano = {
2841 let mut metadata = new_metadata();
2842 for col in metadata.column_metadatas.iter_mut() {
2843 if col.column_schema.name == "ts" {
2844 col.column_schema.data_type = ConcreteDataType::timestamp_nanosecond_datatype();
2845 }
2846 }
2847 metadata
2848 };
2849 AlterKind::ModifyColumnTypes {
2850 columns: vec![ModifyColumnType {
2851 column_name: "ts".to_string(),
2852 target_type: ConcreteDataType::timestamp_millisecond_datatype(),
2853 }],
2854 }
2855 .validate(&metadata_nano)
2856 .unwrap_err();
2857
2858 let same_type = AlterKind::ModifyColumnTypes {
2862 columns: vec![ModifyColumnType {
2863 column_name: "ts".to_string(),
2864 target_type: ConcreteDataType::timestamp_millisecond_datatype(),
2865 }],
2866 };
2867 same_type.validate(&metadata).unwrap();
2868 assert!(!same_type.need_alter(&metadata));
2869
2870 AlterKind::ModifyColumnTypes {
2872 columns: vec![ModifyColumnType {
2873 column_name: "ts".to_string(),
2874 target_type: ConcreteDataType::string_datatype(),
2875 }],
2876 }
2877 .validate(&metadata)
2878 .unwrap_err();
2879
2880 AlterKind::ModifyColumnTypes {
2881 columns: vec![ModifyColumnType {
2882 column_name: "tag_0".to_string(),
2883 target_type: ConcreteDataType::date_datatype(),
2884 }],
2885 }
2886 .validate(&metadata)
2887 .unwrap_err();
2888
2889 let kind = AlterKind::ModifyColumnTypes {
2890 columns: vec![ModifyColumnType {
2891 column_name: "field_0".to_string(),
2892 target_type: ConcreteDataType::int32_datatype(),
2893 }],
2894 };
2895 kind.validate(&metadata).unwrap();
2896 assert!(kind.need_alter(&metadata));
2897 }
2898
2899 #[test]
2900 fn test_validate_set_json_settings() {
2901 let mut metadata = new_metadata();
2902 let current_schema = json2_column_schema("field_0", JsonSettings::new_v2());
2903 let target_settings = JsonSettings::try_new(vec![], Some(10)).unwrap();
2904 metadata
2905 .column_metadatas
2906 .iter_mut()
2907 .find(|column| column.column_schema.name == "field_0")
2908 .unwrap()
2909 .column_schema = current_schema.clone();
2910
2911 let kind = AlterKind::SetJsonSettings {
2912 column_name: "field_0".to_string(),
2913 settings: target_settings.clone(),
2914 };
2915 kind.validate(&metadata).unwrap();
2916 assert!(kind.need_alter(&metadata));
2917
2918 let no_op = AlterKind::SetJsonSettings {
2919 column_name: "field_0".to_string(),
2920 settings: JsonSettings::new_v2(),
2921 };
2922 no_op.validate(&metadata).unwrap();
2923 assert!(!no_op.need_alter(&metadata));
2924
2925 AlterKind::SetJsonSettings {
2926 column_name: "tag_0".to_string(),
2927 settings: target_settings,
2928 }
2929 .validate(&metadata)
2930 .unwrap_err();
2931 }
2932
2933 #[test]
2934 fn test_validate_add_columns() {
2935 let kind = AlterKind::AddColumns {
2936 columns: vec![
2937 AddColumn {
2938 column_metadata: ColumnMetadata {
2939 column_schema: ColumnSchema::new(
2940 "tag_1",
2941 ConcreteDataType::string_datatype(),
2942 true,
2943 ),
2944 semantic_type: SemanticType::Tag,
2945 column_id: 5,
2946 },
2947 location: None,
2948 },
2949 AddColumn {
2950 column_metadata: ColumnMetadata {
2951 column_schema: ColumnSchema::new(
2952 "field_2",
2953 ConcreteDataType::string_datatype(),
2954 true,
2955 ),
2956 semantic_type: SemanticType::Field,
2957 column_id: 6,
2958 },
2959 location: None,
2960 },
2961 ],
2962 };
2963 let request = RegionAlterRequest { kind };
2964 let mut metadata = new_metadata();
2965 metadata.schema_version = 1;
2966 request.validate(&metadata).unwrap();
2967 }
2968
2969 #[test]
2970 fn test_validate_create_region() {
2971 let column_metadatas = vec![
2972 ColumnMetadata {
2973 column_schema: ColumnSchema::new(
2974 "ts",
2975 ConcreteDataType::timestamp_millisecond_datatype(),
2976 false,
2977 ),
2978 semantic_type: SemanticType::Timestamp,
2979 column_id: 1,
2980 },
2981 ColumnMetadata {
2982 column_schema: ColumnSchema::new(
2983 "tag_0",
2984 ConcreteDataType::string_datatype(),
2985 true,
2986 ),
2987 semantic_type: SemanticType::Tag,
2988 column_id: 2,
2989 },
2990 ColumnMetadata {
2991 column_schema: ColumnSchema::new(
2992 "field_0",
2993 ConcreteDataType::string_datatype(),
2994 true,
2995 ),
2996 semantic_type: SemanticType::Field,
2997 column_id: 3,
2998 },
2999 ];
3000 let create = RegionCreateRequest {
3001 engine: "mito".to_string(),
3002 column_metadatas,
3003 primary_key: vec![3, 4],
3004 options: HashMap::new(),
3005 table_dir: "path".to_string(),
3006 path_type: PathType::Bare,
3007 partition_expr_json: Some("".to_string()),
3008 requirements: Default::default(),
3009 };
3010
3011 assert!(create.validate().is_err());
3012 }
3013
3014 #[test]
3015 fn test_parse_create_region_requirements_defaults_to_empty() {
3016 let create = CreateRequest {
3017 region_id: RegionId::new(42, 0).as_u64(),
3018 engine: "mito".to_string(),
3019 column_defs: vec![],
3020 primary_key: vec![],
3021 path: "test".to_string(),
3022 options: HashMap::new(),
3023 partition: None,
3024 requirements: None,
3025 };
3026
3027 let requests =
3028 RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
3029 let RegionRequest::Create(request) = &requests[0].1 else {
3030 unreachable!()
3031 };
3032
3033 assert_eq!(request.requirements, RegionRequirements::empty());
3034 }
3035
3036 #[test]
3037 fn test_parse_create_region_requirements_from_proto() {
3038 let create = CreateRequest {
3039 region_id: RegionId::new(42, 0).as_u64(),
3040 engine: "mito".to_string(),
3041 column_defs: vec![],
3042 primary_key: vec![],
3043 path: "test".to_string(),
3044 options: HashMap::new(),
3045 partition: None,
3046 requirements: Some(api::v1::region::RegionRequirements {
3047 object_storage: true,
3048 }),
3049 };
3050
3051 let requests =
3052 RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
3053 let RegionRequest::Create(request) = &requests[0].1 else {
3054 unreachable!()
3055 };
3056
3057 assert_eq!(request.requirements, RegionRequirements::object_storage());
3058 }
3059
3060 #[test]
3061 fn test_parse_region_cleanup_from_proto() {
3062 let clean_up = api::v1::region::CleanUpRequest {
3063 region_id: RegionId::new(42, 3).as_u64(),
3064 engine: "mito".to_string(),
3065 path: "test".to_string(),
3066 options: HashMap::from([("k".to_string(), "v".to_string())]),
3067 };
3068
3069 let requests =
3070 RegionRequest::try_from_request_body(region_request::Body::CleanUp(clean_up)).unwrap();
3071 let RegionRequest::CleanUp(request) = &requests[0].1 else {
3072 unreachable!()
3073 };
3074
3075 assert_eq!(requests[0].0, RegionId::new(42, 3));
3076 assert_eq!(request.engine, "mito");
3077 assert_eq!(request.table_dir, "data/test/42/");
3078 assert_eq!(request.path_type, PathType::Bare);
3079 assert_eq!(request.options.get("k"), Some(&"v".to_string()));
3080 }
3081
3082 #[test]
3083 fn test_validate_modify_column_fulltext_options() {
3084 let kind = AlterKind::SetIndexes {
3085 options: vec![SetIndexOption::Fulltext {
3086 column_name: "tag_0".to_string(),
3087 options: FulltextOptions::new_unchecked(
3088 true,
3089 FulltextAnalyzer::Chinese,
3090 false,
3091 FulltextBackend::Bloom,
3092 1000,
3093 0.01,
3094 ),
3095 }],
3096 };
3097 let request = RegionAlterRequest { kind };
3098 let mut metadata = new_metadata();
3099 metadata.schema_version = 1;
3100 request.validate(&metadata).unwrap();
3101
3102 let kind = AlterKind::UnsetIndexes {
3103 options: vec![UnsetIndexOption::Fulltext {
3104 column_name: "tag_0".to_string(),
3105 }],
3106 };
3107 let request = RegionAlterRequest { kind };
3108 let mut metadata = new_metadata();
3109 metadata.schema_version = 1;
3110 request.validate(&metadata).unwrap();
3111 }
3112
3113 #[test]
3114 fn test_validate_sync_columns() {
3115 let metadata = new_metadata();
3116 let kind = AlterKind::SyncColumns {
3117 column_metadatas: vec![
3118 ColumnMetadata {
3119 column_schema: ColumnSchema::new(
3120 "tag_1",
3121 ConcreteDataType::string_datatype(),
3122 true,
3123 ),
3124 semantic_type: SemanticType::Tag,
3125 column_id: 5,
3126 },
3127 ColumnMetadata {
3128 column_schema: ColumnSchema::new(
3129 "field_2",
3130 ConcreteDataType::string_datatype(),
3131 true,
3132 ),
3133 semantic_type: SemanticType::Field,
3134 column_id: 6,
3135 },
3136 ],
3137 };
3138 let err = kind.validate(&metadata).unwrap_err();
3139 assert!(err.to_string().contains("not a primary key"));
3140
3141 let mut column_metadatas_with_different_ts_column = metadata.column_metadatas.clone();
3143 let ts_column = column_metadatas_with_different_ts_column
3144 .iter_mut()
3145 .find(|c| c.semantic_type == SemanticType::Timestamp)
3146 .unwrap();
3147 ts_column.column_schema.name = "ts1".to_string();
3148
3149 let kind = AlterKind::SyncColumns {
3150 column_metadatas: column_metadatas_with_different_ts_column,
3151 };
3152 let err = kind.validate(&metadata).unwrap_err();
3153 assert!(
3154 err.to_string()
3155 .contains("timestamp column ts has different id")
3156 );
3157
3158 let mut column_metadatas_with_different_pk_column = metadata.column_metadatas.clone();
3160 let pk_column = column_metadatas_with_different_pk_column
3161 .iter_mut()
3162 .find(|c| c.column_schema.name == "tag_0")
3163 .unwrap();
3164 pk_column.column_id = 100;
3165 let kind = AlterKind::SyncColumns {
3166 column_metadatas: column_metadatas_with_different_pk_column,
3167 };
3168 let err = kind.validate(&metadata).unwrap_err();
3169 assert!(
3170 err.to_string()
3171 .contains("column with same name tag_0 has different id")
3172 );
3173
3174 let mut column_metadatas_with_new_field_column = metadata.column_metadatas.clone();
3176 column_metadatas_with_new_field_column.push(ColumnMetadata {
3177 column_schema: ColumnSchema::new("field_2", ConcreteDataType::string_datatype(), true),
3178 semantic_type: SemanticType::Field,
3179 column_id: 4,
3180 });
3181 let kind = AlterKind::SyncColumns {
3182 column_metadatas: column_metadatas_with_new_field_column,
3183 };
3184 kind.validate(&metadata).unwrap();
3185 }
3186
3187 #[test]
3188 fn test_cast_path_type_to_primitive() {
3189 assert_eq!(PathType::Bare as u8, 0);
3190 assert_eq!(PathType::Data as u8, 1);
3191 assert_eq!(PathType::Metadata as u8, 2);
3192 assert_eq!(
3193 PathType::try_from(PathType::Bare as u8).unwrap(),
3194 PathType::Bare
3195 );
3196 assert_eq!(
3197 PathType::try_from(PathType::Data as u8).unwrap(),
3198 PathType::Data
3199 );
3200 assert_eq!(
3201 PathType::try_from(PathType::Metadata as u8).unwrap(),
3202 PathType::Metadata
3203 );
3204 }
3205}