datatypes/schema/
column_schema.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::collections::HashMap;
16use std::str::FromStr;
17use std::{fmt, mem};
18
19use arrow::datatypes::Field;
20use arrow_schema::extension::{
21    EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY, ExtensionType,
22};
23use common_base::readable_size::ReadableSize;
24use serde::{Deserialize, Serialize};
25use snafu::{ResultExt, ensure};
26use sqlparser_derive::{Visit, VisitMut};
27
28use crate::data_type::{ConcreteDataType, DataType};
29use crate::error::{
30    self, ArrowMetadataSnafu, Error, InvalidFulltextOptionSnafu, ParseExtendedTypeSnafu, Result,
31};
32use crate::extension::json::Json2ExtensionType;
33use crate::schema::TYPE_KEY;
34use crate::schema::constraint::ColumnDefaultConstraint;
35use crate::value::Value;
36use crate::vectors::json::builder::JsonVectorBuilder;
37use crate::vectors::{MutableVector, VectorRef};
38
39pub type Metadata = HashMap<String, String>;
40
41/// Key used to store whether the column is time index in arrow field's metadata.
42pub const TIME_INDEX_KEY: &str = "greptime:time_index";
43pub const COMMENT_KEY: &str = "greptime:storage:comment";
44/// Key used to store default constraint in arrow field's metadata.
45const DEFAULT_CONSTRAINT_KEY: &str = "greptime:default_constraint";
46/// Key used to store fulltext options in arrow field's metadata.
47pub const FULLTEXT_KEY: &str = "greptime:fulltext";
48/// Key used to store whether the column has inverted index in arrow field's metadata.
49pub const INVERTED_INDEX_KEY: &str = "greptime:inverted_index";
50/// Key used to store skip options in arrow field's metadata.
51pub const SKIPPING_INDEX_KEY: &str = "greptime:skipping_index";
52/// Key used to store SST compression options in arrow field's metadata.
53pub const COMPRESSION_KEY: &str = "greptime:compression";
54
55/// Keys used in fulltext options
56pub const COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE: &str = "enable";
57pub const COLUMN_FULLTEXT_OPT_KEY_ANALYZER: &str = "analyzer";
58pub const COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE: &str = "case_sensitive";
59pub const COLUMN_FULLTEXT_OPT_KEY_BACKEND: &str = "backend";
60pub const COLUMN_FULLTEXT_OPT_KEY_GRANULARITY: &str = "granularity";
61pub const COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE: &str = "false_positive_rate";
62
63/// Keys used in SKIPPING index options
64pub const COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY: &str = "granularity";
65pub const COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE: &str = "false_positive_rate";
66pub const COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE: &str = "type";
67
68/// Keys used in compression options
69pub const COLUMN_COMPRESSION_OPT_KEY_TYPE: &str = "type";
70pub const COLUMN_COMPRESSION_OPT_KEY_LEVEL: &str = "level";
71pub const COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE: &str = "page_size";
72pub const COLUMN_COMPRESSION_OPT_KEY_ENCODING: &str = "encoding";
73
74pub const DEFAULT_GRANULARITY: u32 = 10240;
75
76pub const DEFAULT_FALSE_POSITIVE_RATE: f64 = 0.01;
77
78/// Schema of a column, used as an immutable struct.
79#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
80pub struct ColumnSchema {
81    pub name: String,
82    pub data_type: ConcreteDataType,
83    is_nullable: bool,
84    is_time_index: bool,
85    default_constraint: Option<ColumnDefaultConstraint>,
86    metadata: Metadata,
87}
88
89impl fmt::Debug for ColumnSchema {
90    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
91        write!(
92            f,
93            "{} {} {}",
94            self.name,
95            self.data_type,
96            if self.is_nullable { "null" } else { "not null" },
97        )?;
98
99        if self.is_time_index {
100            write!(f, " time_index")?;
101        }
102
103        // Add default constraint if present
104        if let Some(default_constraint) = &self.default_constraint {
105            write!(f, " default={:?}", default_constraint)?;
106        }
107
108        // Add metadata if present
109        if !self.metadata.is_empty() {
110            write!(f, " metadata={:?}", self.metadata)?;
111        }
112
113        Ok(())
114    }
115}
116
117impl ColumnSchema {
118    pub fn new<T: Into<String>>(
119        name: T,
120        data_type: ConcreteDataType,
121        is_nullable: bool,
122    ) -> ColumnSchema {
123        ColumnSchema {
124            name: name.into(),
125            data_type,
126            is_nullable,
127            is_time_index: false,
128            default_constraint: None,
129            metadata: Metadata::new(),
130        }
131    }
132
133    /// Creates a mutable vector using this column's extension metadata.
134    pub fn create_mutable_vector(&self, capacity: usize) -> Box<dyn MutableVector> {
135        if self.data_type.is_json2()
136            && let Some(extension) = self.extension_type::<Json2ExtensionType>().ok().flatten()
137            && extension.metadata().is_version_2()
138        {
139            Box::new(JsonVectorBuilder::with_settings(
140                extension.metadata().json_settings(),
141                capacity,
142            ))
143        } else {
144            self.data_type.create_mutable_vector(capacity)
145        }
146    }
147
148    #[inline]
149    pub fn is_time_index(&self) -> bool {
150        self.is_time_index
151    }
152
153    #[inline]
154    pub fn is_nullable(&self) -> bool {
155        self.is_nullable
156    }
157
158    #[inline]
159    pub fn default_constraint(&self) -> Option<&ColumnDefaultConstraint> {
160        self.default_constraint.as_ref()
161    }
162
163    /// Check if the default constraint is a impure function.
164    pub fn is_default_impure(&self) -> bool {
165        self.default_constraint
166            .as_ref()
167            .map(|c| c.is_function())
168            .unwrap_or(false)
169    }
170
171    #[inline]
172    pub fn metadata(&self) -> &Metadata {
173        &self.metadata
174    }
175
176    #[inline]
177    pub fn mut_metadata(&mut self) -> &mut Metadata {
178        &mut self.metadata
179    }
180
181    /// Retrieve the column comment
182    pub fn column_comment(&self) -> Option<&String> {
183        self.metadata.get(COMMENT_KEY)
184    }
185
186    pub fn with_time_index(mut self, is_time_index: bool) -> Self {
187        self.is_time_index = is_time_index;
188        if is_time_index {
189            let _ = self
190                .metadata
191                .insert(TIME_INDEX_KEY.to_string(), "true".to_string());
192        } else {
193            let _ = self.metadata.remove(TIME_INDEX_KEY);
194        }
195        self
196    }
197
198    /// Returns the estimated memory footprint of this schema.
199    pub fn estimated_size(&self) -> usize {
200        mem::size_of_val(self) - mem::size_of_val(&self.data_type)
201            + self.data_type.as_arrow_type().size()
202            + self.name.capacity()
203            + self
204                .default_constraint
205                .as_ref()
206                .map(column_default_constraint_size)
207                .unwrap_or_default()
208            + metadata_size(&self.metadata)
209    }
210
211    /// Set the inverted index for the column.
212    /// Similar to [with_inverted_index] but don't take the ownership.
213    ///
214    /// [with_inverted_index]: Self::with_inverted_index
215    pub fn set_inverted_index(&mut self, value: bool) {
216        match value {
217            true => {
218                self.metadata
219                    .insert(INVERTED_INDEX_KEY.to_string(), value.to_string());
220            }
221            false => {
222                self.metadata.remove(INVERTED_INDEX_KEY);
223            }
224        }
225    }
226
227    /// Set the inverted index for the column.
228    /// Similar to [set_inverted_index] but take the ownership and return a owned value.
229    ///
230    /// [set_inverted_index]: Self::set_inverted_index
231    pub fn with_inverted_index(mut self, value: bool) -> Self {
232        self.set_inverted_index(value);
233        self
234    }
235
236    pub fn is_inverted_indexed(&self) -> bool {
237        self.metadata
238            .get(INVERTED_INDEX_KEY)
239            .map(|v| v.eq_ignore_ascii_case("true"))
240            .unwrap_or(false)
241    }
242
243    pub fn is_fulltext_indexed(&self) -> bool {
244        self.fulltext_options()
245            .unwrap_or_default()
246            .map(|option| option.enable)
247            .unwrap_or_default()
248    }
249
250    pub fn is_skipping_indexed(&self) -> bool {
251        self.skipping_index_options().unwrap_or_default().is_some()
252    }
253
254    pub fn has_inverted_index_key(&self) -> bool {
255        self.metadata.contains_key(INVERTED_INDEX_KEY)
256    }
257
258    /// Set default constraint.
259    ///
260    /// If a default constraint exists for the column, this method will
261    /// validate it against the column's data type and nullability.
262    pub fn with_default_constraint(
263        mut self,
264        default_constraint: Option<ColumnDefaultConstraint>,
265    ) -> Result<Self> {
266        if let Some(constraint) = &default_constraint {
267            constraint.validate(&self.data_type, self.is_nullable)?;
268        }
269
270        self.default_constraint = default_constraint;
271        Ok(self)
272    }
273
274    /// Set the nullablity to `true` of the column.
275    /// Similar to [set_nullable] but take the ownership and return a owned value.
276    ///
277    /// [set_nullable]: Self::set_nullable
278    pub fn with_nullable_set(mut self) -> Self {
279        self.is_nullable = true;
280        self
281    }
282
283    /// Set the nullability to `true` of the column.
284    /// Similar to [with_nullable_set] but don't take the ownership
285    ///
286    /// [with_nullable_set]: Self::with_nullable_set
287    pub fn set_nullable(&mut self) {
288        self.is_nullable = true;
289    }
290
291    /// Set the `is_time_index` to `true` of the column.
292    /// Similar to [with_time_index] but don't take the ownership.
293    ///
294    /// [with_time_index]: Self::with_time_index
295    pub fn set_time_index(&mut self) {
296        self.is_time_index = true;
297    }
298
299    /// Creates a new [`ColumnSchema`] with given metadata.
300    pub fn with_metadata(mut self, metadata: Metadata) -> Self {
301        self.metadata = metadata;
302        self
303    }
304
305    /// Creates a vector with default value for this column.
306    ///
307    /// If the column is `NOT NULL` but doesn't has `DEFAULT` value supplied, returns `Ok(None)`.
308    pub fn create_default_vector(&self, num_rows: usize) -> Result<Option<VectorRef>> {
309        match &self.default_constraint {
310            Some(c) => c
311                .create_default_vector(&self.data_type, self.is_nullable, num_rows)
312                .map(Some),
313            None => {
314                if self.is_nullable {
315                    // No default constraint, use null as default value.
316                    // TODO(yingwen): Use NullVector once it supports setting logical type.
317                    ColumnDefaultConstraint::null_value()
318                        .create_default_vector(&self.data_type, self.is_nullable, num_rows)
319                        .map(Some)
320                } else {
321                    Ok(None)
322                }
323            }
324        }
325    }
326
327    /// Creates a vector for padding.
328    ///
329    /// This method always returns a vector since it uses [DataType::default_value]
330    /// to fill the vector. Callers should only use the created vector for padding
331    /// and never read its content.
332    pub fn create_default_vector_for_padding(&self, num_rows: usize) -> VectorRef {
333        let padding_value = if self.is_nullable {
334            Value::Null
335        } else {
336            // If the column is not null, use the data type's default value as it is
337            // more efficient to acquire.
338            self.data_type.default_value()
339        };
340        let value_ref = padding_value.as_value_ref();
341        let mut mutable_vector = self.data_type.create_mutable_vector(num_rows);
342        for _ in 0..num_rows {
343            mutable_vector.push_value_ref(&value_ref);
344        }
345        mutable_vector.to_vector()
346    }
347
348    /// Creates a default value for this column.
349    ///
350    /// If the column is `NOT NULL` but doesn't has `DEFAULT` value supplied, returns `Ok(None)`.
351    pub fn create_default(&self) -> Result<Option<Value>> {
352        match &self.default_constraint {
353            Some(c) => c
354                .create_default(&self.data_type, self.is_nullable)
355                .map(Some),
356            None => {
357                if self.is_nullable {
358                    // No default constraint, use null as default value.
359                    ColumnDefaultConstraint::null_value()
360                        .create_default(&self.data_type, self.is_nullable)
361                        .map(Some)
362                } else {
363                    Ok(None)
364                }
365            }
366        }
367    }
368
369    /// Creates an impure default value for this column, only if it have a impure default constraint.
370    /// Otherwise, returns `Ok(None)`.
371    pub fn create_impure_default(&self) -> Result<Option<Value>> {
372        match &self.default_constraint {
373            Some(c) => c.create_impure_default(&self.data_type),
374            None => Ok(None),
375        }
376    }
377
378    /// Retrieves the fulltext options for the column.
379    pub fn fulltext_options(&self) -> Result<Option<FulltextOptions>> {
380        match self.metadata.get(FULLTEXT_KEY) {
381            None => Ok(None),
382            Some(json) => {
383                let options =
384                    serde_json::from_str(json).context(error::DeserializeSnafu { json })?;
385                Ok(Some(options))
386            }
387        }
388    }
389
390    pub fn with_fulltext_options(mut self, options: FulltextOptions) -> Result<Self> {
391        self.metadata.insert(
392            FULLTEXT_KEY.to_string(),
393            serde_json::to_string(&options).context(error::SerializeSnafu)?,
394        );
395        Ok(self)
396    }
397
398    pub fn set_fulltext_options(&mut self, options: &FulltextOptions) -> Result<()> {
399        self.metadata.insert(
400            FULLTEXT_KEY.to_string(),
401            serde_json::to_string(options).context(error::SerializeSnafu)?,
402        );
403        Ok(())
404    }
405
406    /// Retrieves the skipping index options for the column.
407    pub fn skipping_index_options(&self) -> Result<Option<SkippingIndexOptions>> {
408        match self.metadata.get(SKIPPING_INDEX_KEY) {
409            None => Ok(None),
410            Some(json) => {
411                let options =
412                    serde_json::from_str(json).context(error::DeserializeSnafu { json })?;
413                Ok(Some(options))
414            }
415        }
416    }
417
418    pub fn with_skipping_options(mut self, options: SkippingIndexOptions) -> Result<Self> {
419        self.metadata.insert(
420            SKIPPING_INDEX_KEY.to_string(),
421            serde_json::to_string(&options).context(error::SerializeSnafu)?,
422        );
423        Ok(self)
424    }
425
426    pub fn set_skipping_options(&mut self, options: &SkippingIndexOptions) -> Result<()> {
427        self.metadata.insert(
428            SKIPPING_INDEX_KEY.to_string(),
429            serde_json::to_string(options).context(error::SerializeSnafu)?,
430        );
431        Ok(())
432    }
433
434    pub fn unset_skipping_options(&mut self) -> Result<()> {
435        self.metadata.remove(SKIPPING_INDEX_KEY);
436        Ok(())
437    }
438
439    /// Retrieves the SST compression options for the column.
440    pub fn compression_options(&self) -> Result<Option<CompressionOptions>> {
441        match self.metadata.get(COMPRESSION_KEY) {
442            None => Ok(None),
443            Some(json) => {
444                let options = serde_json::from_str(json).map_err(|e| {
445                    error::InvalidCompressionOptionSnafu {
446                        msg: format!("invalid compression metadata for column {}: {e}", self.name),
447                    }
448                    .build()
449                })?;
450                self.validate_compression_options(&options)?;
451                Ok(Some(options))
452            }
453        }
454    }
455
456    /// Sets the SST compression options. Nested types are rejected: their values are
457    /// stored in child columns that the options would not reach.
458    pub fn with_compression_options(mut self, options: CompressionOptions) -> Result<Self> {
459        self.validate_compression_options(&options)?;
460        self.metadata.insert(
461            COMPRESSION_KEY.to_string(),
462            serde_json::to_string(&options).context(error::SerializeSnafu)?,
463        );
464        Ok(self)
465    }
466
467    fn validate_compression_options(&self, options: &CompressionOptions) -> Result<()> {
468        options.validate()?;
469        ensure!(
470            !self.data_type.as_arrow_type().is_nested(),
471            error::InvalidCompressionOptionSnafu {
472                msg: format!(
473                    "compression options are not supported on column {} of type {}",
474                    self.name, self.data_type
475                ),
476            }
477        );
478        if options.encoding == Some(ColumnEncoding::DeltaBinaryPacked) {
479            ensure!(
480                self.data_type.is_timestamp() || self.data_type.as_arrow_type().is_integer(),
481                error::InvalidCompressionOptionSnafu {
482                    msg: format!(
483                        "encoding delta_binary_packed requires an integer or timestamp column, but column {} is {}",
484                        self.name, self.data_type
485                    ),
486                }
487            );
488        }
489        Ok(())
490    }
491
492    pub fn extension_type<E>(&self) -> Result<Option<E>>
493    where
494        E: ExtensionType,
495    {
496        let extension_type_name = self.metadata.get(EXTENSION_TYPE_NAME_KEY);
497
498        if extension_type_name.map(|s| s.as_str()) == Some(E::NAME) {
499            let extension_metadata = self.metadata.get(EXTENSION_TYPE_METADATA_KEY);
500            let extension_metadata =
501                E::deserialize_metadata(extension_metadata.map(|s| s.as_str()))
502                    .context(ArrowMetadataSnafu)?;
503
504            let extension = E::try_new(&self.data_type.as_arrow_type(), extension_metadata)
505                .context(ArrowMetadataSnafu)?;
506            Ok(Some(extension))
507        } else {
508            Ok(None)
509        }
510    }
511
512    /// Sets the Arrow extension type metadata for this column.
513    pub fn with_extension_type<E>(&mut self, extension_type: &E)
514    where
515        E: ExtensionType,
516    {
517        self.metadata
518            .insert(EXTENSION_TYPE_NAME_KEY.to_string(), E::NAME.to_string());
519
520        if let Some(extension_metadata) = extension_type.serialize_metadata() {
521            self.metadata
522                .insert(EXTENSION_TYPE_METADATA_KEY.to_string(), extension_metadata);
523        } else {
524            // Replacing an extension must not retain metadata owned by the previous type.
525            self.metadata.remove(EXTENSION_TYPE_METADATA_KEY);
526        }
527    }
528
529    pub fn is_indexed(&self) -> bool {
530        self.is_inverted_indexed() || self.is_fulltext_indexed() || self.is_skipping_indexed()
531    }
532}
533
534fn metadata_size(metadata: &Metadata) -> usize {
535    mem::size_of::<(String, String)>() * metadata.capacity()
536        + metadata
537            .iter()
538            .map(|(key, value)| key.capacity() + value.capacity())
539            .sum::<usize>()
540}
541
542fn column_default_constraint_size(default_constraint: &ColumnDefaultConstraint) -> usize {
543    match default_constraint {
544        ColumnDefaultConstraint::Function(expr) => expr.capacity(),
545        ColumnDefaultConstraint::Value(value) => value.as_value_ref().data_size(),
546    }
547}
548
549/// Column extended type set in column schema's metadata.
550#[derive(Debug, Clone, PartialEq, Eq)]
551pub enum ColumnExtType {
552    /// Json type.
553    Json,
554
555    /// Vector type with dimension.
556    Vector(u32),
557}
558
559impl fmt::Display for ColumnExtType {
560    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
561        match self {
562            ColumnExtType::Json => write!(f, "Json"),
563            ColumnExtType::Vector(dim) => write!(f, "Vector({})", dim),
564        }
565    }
566}
567
568impl FromStr for ColumnExtType {
569    type Err = String;
570
571    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
572        match s {
573            "Json" => Ok(ColumnExtType::Json),
574            _ if s.starts_with("Vector(") && s.ends_with(')') => s[7..s.len() - 1]
575                .parse::<u32>()
576                .map(ColumnExtType::Vector)
577                .map_err(|_| "Invalid dimension for Vector".to_string()),
578            _ => Err("Unknown variant".to_string()),
579        }
580    }
581}
582
583impl TryFrom<&Field> for ColumnSchema {
584    type Error = Error;
585
586    fn try_from(field: &Field) -> Result<ColumnSchema> {
587        let mut data_type = ConcreteDataType::try_from(field.data_type())?;
588        // Override the data type if it is specified in the metadata.
589        if let Some(s) = field.metadata().get(TYPE_KEY) {
590            let extype = ColumnExtType::from_str(s)
591                .map_err(|_| ParseExtendedTypeSnafu { value: s }.build())?;
592            match extype {
593                ColumnExtType::Json => {
594                    data_type = ConcreteDataType::json_datatype();
595                }
596                ColumnExtType::Vector(dim) => {
597                    data_type = ConcreteDataType::vector_datatype(dim);
598                }
599            }
600        }
601        let mut metadata = field.metadata().clone();
602        let default_constraint = match metadata.remove(DEFAULT_CONSTRAINT_KEY) {
603            Some(json) => {
604                Some(serde_json::from_str(&json).context(error::DeserializeSnafu { json })?)
605            }
606            None => None,
607        };
608        let mut is_time_index = metadata.contains_key(TIME_INDEX_KEY);
609        if is_time_index && !data_type.is_timestamp() {
610            // If the column is time index but the data type is not timestamp, it is invalid.
611            // We set the time index to false and remove the metadata.
612            // This is possible if we cast the time index column to another type. DataFusion will
613            // keep the metadata:
614            // https://github.com/apache/datafusion/pull/12951
615            is_time_index = false;
616            metadata.remove(TIME_INDEX_KEY);
617            common_telemetry::debug!(
618                "Column {} is not timestamp ({:?}) but has time index metadata",
619                data_type,
620                field.name(),
621            );
622        }
623
624        Ok(ColumnSchema {
625            name: field.name().clone(),
626            data_type,
627            is_nullable: field.is_nullable(),
628            is_time_index,
629            default_constraint,
630            metadata,
631        })
632    }
633}
634
635impl TryFrom<&ColumnSchema> for Field {
636    type Error = Error;
637
638    fn try_from(column_schema: &ColumnSchema) -> Result<Field> {
639        let mut metadata = column_schema.metadata.clone();
640        if let Some(value) = &column_schema.default_constraint {
641            // Adds an additional metadata to store the default constraint.
642            let old = metadata.insert(
643                DEFAULT_CONSTRAINT_KEY.to_string(),
644                serde_json::to_string(&value).context(error::SerializeSnafu)?,
645            );
646
647            ensure!(
648                old.is_none(),
649                error::DuplicateMetaSnafu {
650                    key: DEFAULT_CONSTRAINT_KEY,
651                }
652            );
653        }
654
655        Ok(Field::new(
656            &column_schema.name,
657            column_schema.data_type.as_arrow_type(),
658            column_schema.is_nullable(),
659        )
660        .with_metadata(metadata))
661    }
662}
663
664/// Fulltext options for a column.
665#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Visit, VisitMut)]
666#[serde(rename_all = "kebab-case")]
667pub struct FulltextOptions {
668    /// Whether the fulltext index is enabled.
669    pub enable: bool,
670    /// The fulltext analyzer to use.
671    #[serde(default)]
672    pub analyzer: FulltextAnalyzer,
673    /// Whether the fulltext index is case-sensitive.
674    #[serde(default)]
675    pub case_sensitive: bool,
676    /// The fulltext backend to use.
677    #[serde(default)]
678    pub backend: FulltextBackend,
679    /// The granularity of the fulltext index (for bloom backend only)
680    #[serde(default = "fulltext_options_default_granularity")]
681    pub granularity: u32,
682    /// The false positive rate of the fulltext index (for bloom backend only)
683    #[serde(default = "index_options_default_false_positive_rate_in_10000")]
684    pub false_positive_rate_in_10000: u32,
685}
686
687fn fulltext_options_default_granularity() -> u32 {
688    DEFAULT_GRANULARITY
689}
690
691fn index_options_default_false_positive_rate_in_10000() -> u32 {
692    (DEFAULT_FALSE_POSITIVE_RATE * 10000.0) as u32
693}
694
695impl FulltextOptions {
696    /// Creates a new fulltext options.
697    pub fn new(
698        enable: bool,
699        analyzer: FulltextAnalyzer,
700        case_sensitive: bool,
701        backend: FulltextBackend,
702        granularity: u32,
703        false_positive_rate: f64,
704    ) -> Result<Self> {
705        ensure!(
706            0.0 < false_positive_rate && false_positive_rate <= 1.0,
707            error::InvalidFulltextOptionSnafu {
708                msg: format!(
709                    "Invalid false positive rate: {false_positive_rate}, expected: 0.0 < rate <= 1.0"
710                ),
711            }
712        );
713        ensure!(
714            granularity > 0,
715            error::InvalidFulltextOptionSnafu {
716                msg: format!("Invalid granularity: {granularity}, expected: positive integer"),
717            }
718        );
719        Ok(Self::new_unchecked(
720            enable,
721            analyzer,
722            case_sensitive,
723            backend,
724            granularity,
725            false_positive_rate,
726        ))
727    }
728
729    /// Creates a new fulltext options without checking `false_positive_rate` and `granularity`.
730    pub fn new_unchecked(
731        enable: bool,
732        analyzer: FulltextAnalyzer,
733        case_sensitive: bool,
734        backend: FulltextBackend,
735        granularity: u32,
736        false_positive_rate: f64,
737    ) -> Self {
738        Self {
739            enable,
740            analyzer,
741            case_sensitive,
742            backend,
743            granularity,
744            false_positive_rate_in_10000: (false_positive_rate * 10000.0) as u32,
745        }
746    }
747
748    /// Gets the false positive rate.
749    pub fn false_positive_rate(&self) -> f64 {
750        self.false_positive_rate_in_10000 as f64 / 10000.0
751    }
752}
753
754impl Default for FulltextOptions {
755    fn default() -> Self {
756        Self::new_unchecked(
757            false,
758            FulltextAnalyzer::default(),
759            false,
760            FulltextBackend::default(),
761            DEFAULT_GRANULARITY,
762            DEFAULT_FALSE_POSITIVE_RATE,
763        )
764    }
765}
766
767impl fmt::Display for FulltextOptions {
768    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
769        write!(f, "enable={}", self.enable)?;
770        if self.enable {
771            write!(f, ", analyzer={}", self.analyzer)?;
772            write!(f, ", case_sensitive={}", self.case_sensitive)?;
773            write!(f, ", backend={}", self.backend)?;
774            if self.backend == FulltextBackend::Bloom {
775                write!(f, ", granularity={}", self.granularity)?;
776                write!(f, ", false_positive_rate={}", self.false_positive_rate())?;
777            }
778        }
779        Ok(())
780    }
781}
782
783/// The backend of the fulltext index.
784#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default, Visit, VisitMut)]
785#[serde(rename_all = "kebab-case")]
786pub enum FulltextBackend {
787    #[default]
788    Bloom,
789    Tantivy,
790}
791
792impl fmt::Display for FulltextBackend {
793    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
794        match self {
795            FulltextBackend::Tantivy => write!(f, "tantivy"),
796            FulltextBackend::Bloom => write!(f, "bloom"),
797        }
798    }
799}
800
801impl TryFrom<HashMap<String, String>> for FulltextOptions {
802    type Error = Error;
803
804    fn try_from(options: HashMap<String, String>) -> Result<Self> {
805        let mut fulltext_options = FulltextOptions {
806            enable: true,
807            ..Default::default()
808        };
809
810        if let Some(enable) = options.get(COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE) {
811            match enable.to_ascii_lowercase().as_str() {
812                "true" => fulltext_options.enable = true,
813                "false" => fulltext_options.enable = false,
814                _ => {
815                    return InvalidFulltextOptionSnafu {
816                        msg: format!("{enable}, expected: 'true' | 'false'"),
817                    }
818                    .fail();
819                }
820            }
821        };
822
823        if let Some(analyzer) = options.get(COLUMN_FULLTEXT_OPT_KEY_ANALYZER) {
824            match analyzer.to_ascii_lowercase().as_str() {
825                "english" => fulltext_options.analyzer = FulltextAnalyzer::English,
826                "chinese" => fulltext_options.analyzer = FulltextAnalyzer::Chinese,
827                _ => {
828                    return InvalidFulltextOptionSnafu {
829                        msg: format!("{analyzer}, expected: 'English' | 'Chinese'"),
830                    }
831                    .fail();
832                }
833            }
834        };
835
836        if let Some(case_sensitive) = options.get(COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE) {
837            match case_sensitive.to_ascii_lowercase().as_str() {
838                "true" => fulltext_options.case_sensitive = true,
839                "false" => fulltext_options.case_sensitive = false,
840                _ => {
841                    return InvalidFulltextOptionSnafu {
842                        msg: format!("{case_sensitive}, expected: 'true' | 'false'"),
843                    }
844                    .fail();
845                }
846            }
847        }
848
849        if let Some(backend) = options.get(COLUMN_FULLTEXT_OPT_KEY_BACKEND) {
850            match backend.to_ascii_lowercase().as_str() {
851                "bloom" => fulltext_options.backend = FulltextBackend::Bloom,
852                "tantivy" => fulltext_options.backend = FulltextBackend::Tantivy,
853                _ => {
854                    return InvalidFulltextOptionSnafu {
855                        msg: format!("{backend}, expected: 'bloom' | 'tantivy'"),
856                    }
857                    .fail();
858                }
859            }
860        }
861
862        if fulltext_options.backend == FulltextBackend::Bloom {
863            // Parse granularity with default value 10240
864            let granularity = match options.get(COLUMN_FULLTEXT_OPT_KEY_GRANULARITY) {
865                Some(value) => value
866                    .parse::<u32>()
867                    .ok()
868                    .filter(|&v| v > 0)
869                    .ok_or_else(|| {
870                        error::InvalidFulltextOptionSnafu {
871                            msg: format!(
872                                "Invalid granularity: {value}, expected: positive integer"
873                            ),
874                        }
875                        .build()
876                    })?,
877                None => DEFAULT_GRANULARITY,
878            };
879            fulltext_options.granularity = granularity;
880
881            // Parse false positive rate with default value 0.01
882            let false_positive_rate = match options.get(COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE)
883            {
884                Some(value) => value
885                    .parse::<f64>()
886                    .ok()
887                    .filter(|&v| v > 0.0 && v <= 1.0)
888                    .ok_or_else(|| {
889                        error::InvalidFulltextOptionSnafu {
890                            msg: format!(
891                                "Invalid false positive rate: {value}, expected: 0.0 < rate <= 1.0"
892                            ),
893                        }
894                        .build()
895                    })?,
896                None => DEFAULT_FALSE_POSITIVE_RATE,
897            };
898            fulltext_options.false_positive_rate_in_10000 = (false_positive_rate * 10000.0) as u32;
899        }
900
901        Ok(fulltext_options)
902    }
903}
904
905/// Fulltext analyzer.
906#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default, Visit, VisitMut)]
907pub enum FulltextAnalyzer {
908    #[default]
909    English,
910    Chinese,
911}
912
913impl fmt::Display for FulltextAnalyzer {
914    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
915        match self {
916            FulltextAnalyzer::English => write!(f, "English"),
917            FulltextAnalyzer::Chinese => write!(f, "Chinese"),
918        }
919    }
920}
921
922/// Skipping options for a column.
923#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Visit, VisitMut)]
924#[serde(rename_all = "kebab-case")]
925pub struct SkippingIndexOptions {
926    /// The granularity of the skip index.
927    pub granularity: u32,
928    /// The false positive rate of the skip index (in ten-thousandths, e.g., 100 = 1%).
929    #[serde(default = "index_options_default_false_positive_rate_in_10000")]
930    pub false_positive_rate_in_10000: u32,
931    /// The type of the skip index.
932    #[serde(default)]
933    pub index_type: SkippingIndexType,
934}
935
936impl SkippingIndexOptions {
937    /// Creates a new skipping index options without checking `false_positive_rate` and `granularity`.
938    pub fn new_unchecked(
939        granularity: u32,
940        false_positive_rate: f64,
941        index_type: SkippingIndexType,
942    ) -> Self {
943        Self {
944            granularity,
945            false_positive_rate_in_10000: (false_positive_rate * 10000.0) as u32,
946            index_type,
947        }
948    }
949
950    /// Creates a new skipping index options.
951    pub fn new(
952        granularity: u32,
953        false_positive_rate: f64,
954        index_type: SkippingIndexType,
955    ) -> Result<Self> {
956        ensure!(
957            0.0 < false_positive_rate && false_positive_rate <= 1.0,
958            error::InvalidSkippingIndexOptionSnafu {
959                msg: format!(
960                    "Invalid false positive rate: {false_positive_rate}, expected: 0.0 < rate <= 1.0"
961                ),
962            }
963        );
964        ensure!(
965            granularity > 0,
966            error::InvalidSkippingIndexOptionSnafu {
967                msg: format!("Invalid granularity: {granularity}, expected: positive integer"),
968            }
969        );
970        Ok(Self::new_unchecked(
971            granularity,
972            false_positive_rate,
973            index_type,
974        ))
975    }
976
977    /// Gets the false positive rate.
978    pub fn false_positive_rate(&self) -> f64 {
979        self.false_positive_rate_in_10000 as f64 / 10000.0
980    }
981}
982
983impl Default for SkippingIndexOptions {
984    fn default() -> Self {
985        Self::new_unchecked(
986            DEFAULT_GRANULARITY,
987            DEFAULT_FALSE_POSITIVE_RATE,
988            SkippingIndexType::default(),
989        )
990    }
991}
992
993impl fmt::Display for SkippingIndexOptions {
994    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
995        write!(f, "granularity={}", self.granularity)?;
996        write!(f, ", false_positive_rate={}", self.false_positive_rate())?;
997        write!(f, ", index_type={}", self.index_type)?;
998        Ok(())
999    }
1000}
1001
1002/// Skip index types.
1003#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize, Visit, VisitMut)]
1004pub enum SkippingIndexType {
1005    #[default]
1006    BloomFilter,
1007}
1008
1009impl fmt::Display for SkippingIndexType {
1010    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1011        match self {
1012            SkippingIndexType::BloomFilter => write!(f, "BLOOM"),
1013        }
1014    }
1015}
1016
1017/// Compression algorithms for SST columns.
1018#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1019#[serde(rename_all = "snake_case")]
1020pub enum CompressionType {
1021    #[default]
1022    Zstd,
1023}
1024
1025impl fmt::Display for CompressionType {
1026    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1027        match self {
1028            CompressionType::Zstd => write!(f, "zstd"),
1029        }
1030    }
1031}
1032
1033/// Parquet value encodings for SST columns. Both replace the default dictionary encoding.
1034#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1035#[serde(rename_all = "snake_case")]
1036pub enum ColumnEncoding {
1037    Plain,
1038    /// Only valid for integer and timestamp columns.
1039    DeltaBinaryPacked,
1040}
1041
1042impl fmt::Display for ColumnEncoding {
1043    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1044        match self {
1045            ColumnEncoding::Plain => write!(f, "plain"),
1046            ColumnEncoding::DeltaBinaryPacked => write!(f, "delta_binary_packed"),
1047        }
1048    }
1049}
1050
1051/// Parquet compression settings for a column in SSTs. Unset fields keep the SST defaults.
1052#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1053pub struct CompressionOptions {
1054    #[serde(rename = "type", default)]
1055    pub compression_type: CompressionType,
1056    #[serde(default, skip_serializing_if = "Option::is_none")]
1057    pub level: Option<i32>,
1058    #[serde(default, skip_serializing_if = "Option::is_none")]
1059    pub page_size: Option<ReadableSize>,
1060    #[serde(default, skip_serializing_if = "Option::is_none")]
1061    pub encoding: Option<ColumnEncoding>,
1062}
1063
1064impl CompressionOptions {
1065    fn validate(&self) -> Result<()> {
1066        if let Some(level) = self.level {
1067            ensure!(
1068                ZSTD_LEVELS.contains(&level),
1069                error::InvalidCompressionOptionSnafu {
1070                    msg: format!("invalid level: {level}, expected: an integer in 1..=22"),
1071                }
1072            );
1073        }
1074        if let Some(page_size) = self.page_size {
1075            ensure!(
1076                page_size.as_bytes() > 0,
1077                error::InvalidCompressionOptionSnafu {
1078                    msg: "page_size must be positive",
1079                }
1080            );
1081        }
1082        Ok(())
1083    }
1084
1085    /// Returns the options as `COMPRESSION WITH (...)` key-value pairs.
1086    pub fn to_options_map(&self) -> HashMap<String, String> {
1087        let mut map = HashMap::from([(
1088            COLUMN_COMPRESSION_OPT_KEY_TYPE.to_string(),
1089            self.compression_type.to_string(),
1090        )]);
1091        if let Some(level) = self.level {
1092            map.insert(
1093                COLUMN_COMPRESSION_OPT_KEY_LEVEL.to_string(),
1094                level.to_string(),
1095            );
1096        }
1097        if let Some(page_size) = self.page_size {
1098            // `Display` rounds to one decimal; the serde form is exact and parses back.
1099            let page_size = match serde_json::to_value(page_size) {
1100                Ok(serde_json::Value::String(size)) => size,
1101                _ => page_size.as_bytes().to_string(),
1102            };
1103            map.insert(COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE.to_string(), page_size);
1104        }
1105        if let Some(encoding) = self.encoding {
1106            map.insert(
1107                COLUMN_COMPRESSION_OPT_KEY_ENCODING.to_string(),
1108                encoding.to_string(),
1109            );
1110        }
1111        map
1112    }
1113}
1114
1115/// Valid zstd levels, as accepted by the parquet writer.
1116const ZSTD_LEVELS: std::ops::RangeInclusive<i32> = 1..=22;
1117
1118impl TryFrom<HashMap<String, String>> for CompressionOptions {
1119    type Error = Error;
1120
1121    fn try_from(options: HashMap<String, String>) -> Result<Self> {
1122        let compression_type = match options.get(COLUMN_COMPRESSION_OPT_KEY_TYPE) {
1123            Some(value) if value.eq_ignore_ascii_case("zstd") => CompressionType::Zstd,
1124            Some(value) => {
1125                return error::InvalidCompressionOptionSnafu {
1126                    msg: format!("invalid type: {value}, expected: 'zstd'"),
1127                }
1128                .fail();
1129            }
1130            None => CompressionType::default(),
1131        };
1132
1133        let level = options
1134            .get(COLUMN_COMPRESSION_OPT_KEY_LEVEL)
1135            .map(|value| {
1136                value
1137                    .parse::<i32>()
1138                    .ok()
1139                    .filter(|level| ZSTD_LEVELS.contains(level))
1140                    .ok_or_else(|| {
1141                        error::InvalidCompressionOptionSnafu {
1142                            msg: format!(
1143                                "invalid level: {value}, expected: an integer in {}..={}",
1144                                ZSTD_LEVELS.start(),
1145                                ZSTD_LEVELS.end()
1146                            ),
1147                        }
1148                        .build()
1149                    })
1150            })
1151            .transpose()?;
1152
1153        let page_size = options
1154            .get(COLUMN_COMPRESSION_OPT_KEY_PAGE_SIZE)
1155            .map(|value| {
1156                ReadableSize::from_str(value)
1157                    .ok()
1158                    .filter(|size| size.as_bytes() > 0)
1159                    .ok_or_else(|| {
1160                        error::InvalidCompressionOptionSnafu {
1161                            msg: format!(
1162                                "invalid page_size: {value}, expected: a positive size such as '8MiB'"
1163                            ),
1164                        }
1165                        .build()
1166                    })
1167            })
1168            .transpose()?;
1169
1170        let encoding = options
1171            .get(COLUMN_COMPRESSION_OPT_KEY_ENCODING)
1172            .map(|value| match value.to_ascii_lowercase().as_str() {
1173                "plain" => Ok(ColumnEncoding::Plain),
1174                "delta_binary_packed" => Ok(ColumnEncoding::DeltaBinaryPacked),
1175                _ => error::InvalidCompressionOptionSnafu {
1176                    msg: format!(
1177                        "invalid encoding: {value}, expected: 'plain' or 'delta_binary_packed'"
1178                    ),
1179                }
1180                .fail(),
1181            })
1182            .transpose()?;
1183
1184        Ok(CompressionOptions {
1185            compression_type,
1186            level,
1187            page_size,
1188            encoding,
1189        })
1190    }
1191}
1192
1193impl TryFrom<HashMap<String, String>> for SkippingIndexOptions {
1194    type Error = Error;
1195
1196    fn try_from(options: HashMap<String, String>) -> Result<Self> {
1197        // Parse granularity with default value 1
1198        let granularity = match options.get(COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY) {
1199            Some(value) => value
1200                .parse::<u32>()
1201                .ok()
1202                .filter(|&v| v > 0)
1203                .ok_or_else(|| {
1204                    error::InvalidSkippingIndexOptionSnafu {
1205                        msg: format!("Invalid granularity: {value}, expected: positive integer"),
1206                    }
1207                    .build()
1208                })?,
1209            None => DEFAULT_GRANULARITY,
1210        };
1211
1212        // Parse false positive rate with default value 100
1213        let false_positive_rate =
1214            match options.get(COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE) {
1215                Some(value) => value
1216                    .parse::<f64>()
1217                    .ok()
1218                    .filter(|&v| v > 0.0 && v <= 1.0)
1219                    .ok_or_else(|| {
1220                        error::InvalidSkippingIndexOptionSnafu {
1221                            msg: format!(
1222                                "Invalid false positive rate: {value}, expected: 0.0 < rate <= 1.0"
1223                            ),
1224                        }
1225                        .build()
1226                    })?,
1227                None => DEFAULT_FALSE_POSITIVE_RATE,
1228            };
1229
1230        // Parse index type with default value BloomFilter
1231        let index_type = match options.get(COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE) {
1232            Some(typ) => match typ.to_ascii_uppercase().as_str() {
1233                "BLOOM" => SkippingIndexType::BloomFilter,
1234                _ => {
1235                    return error::InvalidSkippingIndexOptionSnafu {
1236                        msg: format!("Invalid index type: {typ}, expected: 'BLOOM'"),
1237                    }
1238                    .fail();
1239                }
1240            },
1241            None => SkippingIndexType::default(),
1242        };
1243
1244        Ok(SkippingIndexOptions::new_unchecked(
1245            granularity,
1246            false_positive_rate,
1247            index_type,
1248        ))
1249    }
1250}
1251
1252#[cfg(test)]
1253mod tests {
1254    use std::sync::Arc;
1255
1256    use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
1257
1258    use super::*;
1259    use crate::extension::json::{Json2ExtensionType, JsonExtensionType};
1260    use crate::types::{StructField, StructType};
1261    use crate::value::Value;
1262    use crate::vectors::Int32Vector;
1263
1264    #[test]
1265    fn test_column_schema() {
1266        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true);
1267        let field = Field::try_from(&column_schema).unwrap();
1268        assert_eq!("test", field.name());
1269        assert_eq!(ArrowDataType::Int32, *field.data_type());
1270        assert!(field.is_nullable());
1271
1272        let new_column_schema = ColumnSchema::try_from(&field).unwrap();
1273        assert_eq!(column_schema, new_column_schema);
1274    }
1275
1276    #[test]
1277    fn test_with_extension_type_replaces_metadata() {
1278        let mut schema = ColumnSchema::new("j", ConcreteDataType::json_datatype(), true);
1279
1280        schema.with_extension_type(&Json2ExtensionType::default());
1281        assert_eq!(
1282            Some(Json2ExtensionType::NAME),
1283            schema
1284                .metadata()
1285                .get(EXTENSION_TYPE_NAME_KEY)
1286                .map(String::as_str)
1287        );
1288        assert!(schema.metadata().contains_key(EXTENSION_TYPE_METADATA_KEY));
1289
1290        schema.with_extension_type(&JsonExtensionType);
1291        assert_eq!(
1292            Some(JsonExtensionType::NAME),
1293            schema
1294                .metadata()
1295                .get(EXTENSION_TYPE_NAME_KEY)
1296                .map(String::as_str)
1297        );
1298        assert!(!schema.metadata().contains_key(EXTENSION_TYPE_METADATA_KEY));
1299    }
1300
1301    #[test]
1302    fn test_column_schema_with_default_constraint() {
1303        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
1304            .with_default_constraint(Some(ColumnDefaultConstraint::Value(Value::from(99))))
1305            .unwrap();
1306        assert!(
1307            column_schema
1308                .metadata()
1309                .get(DEFAULT_CONSTRAINT_KEY)
1310                .is_none()
1311        );
1312
1313        let field = Field::try_from(&column_schema).unwrap();
1314        assert_eq!("test", field.name());
1315        assert_eq!(ArrowDataType::Int32, *field.data_type());
1316        assert!(field.is_nullable());
1317        assert_eq!(
1318            "{\"Value\":{\"Int32\":99}}",
1319            field.metadata().get(DEFAULT_CONSTRAINT_KEY).unwrap()
1320        );
1321
1322        let new_column_schema = ColumnSchema::try_from(&field).unwrap();
1323        assert_eq!(column_schema, new_column_schema);
1324    }
1325
1326    #[test]
1327    fn test_column_schema_with_metadata() {
1328        let metadata = Metadata::from([
1329            ("k1".to_string(), "v1".to_string()),
1330            (COMMENT_KEY.to_string(), "test comment".to_string()),
1331        ]);
1332        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
1333            .with_metadata(metadata)
1334            .with_default_constraint(Some(ColumnDefaultConstraint::null_value()))
1335            .unwrap();
1336        assert_eq!("v1", column_schema.metadata().get("k1").unwrap());
1337        assert_eq!("test comment", column_schema.column_comment().unwrap());
1338        assert!(
1339            column_schema
1340                .metadata()
1341                .get(DEFAULT_CONSTRAINT_KEY)
1342                .is_none()
1343        );
1344
1345        let field = Field::try_from(&column_schema).unwrap();
1346        assert_eq!("v1", field.metadata().get("k1").unwrap());
1347        let _ = field.metadata().get(DEFAULT_CONSTRAINT_KEY).unwrap();
1348
1349        let new_column_schema = ColumnSchema::try_from(&field).unwrap();
1350        assert_eq!(column_schema, new_column_schema);
1351    }
1352
1353    #[test]
1354    fn test_column_schema_with_duplicate_metadata() {
1355        let metadata = Metadata::from([(DEFAULT_CONSTRAINT_KEY.to_string(), "v1".to_string())]);
1356        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
1357            .with_metadata(metadata)
1358            .with_default_constraint(Some(ColumnDefaultConstraint::null_value()))
1359            .unwrap();
1360        assert!(Field::try_from(&column_schema).is_err());
1361    }
1362
1363    #[test]
1364    fn test_column_schema_invalid_default_constraint() {
1365        assert!(
1366            ColumnSchema::new("test", ConcreteDataType::int32_datatype(), false)
1367                .with_default_constraint(Some(ColumnDefaultConstraint::null_value()))
1368                .is_err()
1369        );
1370    }
1371
1372    #[test]
1373    fn test_column_default_constraint_try_into_from() {
1374        let default_constraint = ColumnDefaultConstraint::Value(Value::from(42i64));
1375
1376        let bytes: Vec<u8> = default_constraint.clone().try_into().unwrap();
1377        let from_value = ColumnDefaultConstraint::try_from(&bytes[..]).unwrap();
1378
1379        assert_eq!(default_constraint, from_value);
1380    }
1381
1382    #[test]
1383    fn test_column_schema_create_default_null() {
1384        // Implicit default null.
1385        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true);
1386        let v = column_schema.create_default_vector(5).unwrap().unwrap();
1387        assert_eq!(5, v.len());
1388        assert!(v.only_null());
1389
1390        // Explicit default null.
1391        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
1392            .with_default_constraint(Some(ColumnDefaultConstraint::null_value()))
1393            .unwrap();
1394        let v = column_schema.create_default_vector(5).unwrap().unwrap();
1395        assert_eq!(5, v.len());
1396        assert!(v.only_null());
1397    }
1398
1399    #[test]
1400    fn test_column_schema_create_default_null_struct_with_list() {
1401        let list_type =
1402            ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int32_datatype()));
1403        let struct_type =
1404            ConcreteDataType::struct_datatype(StructType::new(Arc::new(vec![StructField::new(
1405                "values".to_string(),
1406                list_type,
1407                true,
1408            )])));
1409        let column_schema = ColumnSchema::new("test", struct_type, true);
1410
1411        let v = column_schema.create_default_vector(5).unwrap().unwrap();
1412
1413        assert_eq!(5, v.len());
1414        assert!(v.only_null());
1415    }
1416
1417    #[test]
1418    fn test_column_schema_no_default() {
1419        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), false);
1420        assert!(column_schema.create_default_vector(5).unwrap().is_none());
1421    }
1422
1423    #[test]
1424    fn test_create_default_vector_for_padding() {
1425        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true);
1426        let vector = column_schema.create_default_vector_for_padding(4);
1427        assert!(vector.only_null());
1428        assert_eq!(4, vector.len());
1429
1430        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), false);
1431        let vector = column_schema.create_default_vector_for_padding(4);
1432        assert_eq!(4, vector.len());
1433        let expect: VectorRef = Arc::new(Int32Vector::from_slice([0, 0, 0, 0]));
1434        assert_eq!(expect, vector);
1435    }
1436
1437    #[test]
1438    fn test_column_schema_single_create_default_null() {
1439        // Implicit default null.
1440        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true);
1441        let v = column_schema.create_default().unwrap().unwrap();
1442        assert!(v.is_null());
1443
1444        // Explicit default null.
1445        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
1446            .with_default_constraint(Some(ColumnDefaultConstraint::null_value()))
1447            .unwrap();
1448        let v = column_schema.create_default().unwrap().unwrap();
1449        assert!(v.is_null());
1450    }
1451
1452    #[test]
1453    fn test_column_schema_single_create_default_not_null() {
1454        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
1455            .with_default_constraint(Some(ColumnDefaultConstraint::Value(Value::Int32(6))))
1456            .unwrap();
1457        let v = column_schema.create_default().unwrap().unwrap();
1458        assert_eq!(v, Value::Int32(6));
1459    }
1460
1461    #[test]
1462    fn test_column_schema_single_no_default() {
1463        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), false);
1464        assert!(column_schema.create_default().unwrap().is_none());
1465    }
1466
1467    #[test]
1468    fn test_debug_for_column_schema() {
1469        let column_schema_int8 =
1470            ColumnSchema::new("test_column_1", ConcreteDataType::int8_datatype(), true);
1471
1472        let column_schema_int32 =
1473            ColumnSchema::new("test_column_2", ConcreteDataType::int32_datatype(), false);
1474
1475        let formatted_int8 = format!("{:?}", column_schema_int8);
1476        let formatted_int32 = format!("{:?}", column_schema_int32);
1477        assert_eq!(formatted_int8, "test_column_1 Int8 null");
1478        assert_eq!(formatted_int32, "test_column_2 Int32 not null");
1479    }
1480
1481    #[test]
1482    fn test_from_field_to_column_schema() {
1483        let field = Field::new("test", ArrowDataType::Int32, true);
1484        let column_schema = ColumnSchema::try_from(&field).unwrap();
1485        assert_eq!("test", column_schema.name);
1486        assert_eq!(ConcreteDataType::int32_datatype(), column_schema.data_type);
1487        assert!(column_schema.is_nullable);
1488        assert!(!column_schema.is_time_index);
1489        assert!(column_schema.default_constraint.is_none());
1490        assert!(column_schema.metadata.is_empty());
1491
1492        let field = Field::new("test", ArrowDataType::Binary, true);
1493        let field = field.with_metadata(Metadata::from([(
1494            TYPE_KEY.to_string(),
1495            ConcreteDataType::json_datatype().name(),
1496        )]));
1497        let column_schema = ColumnSchema::try_from(&field).unwrap();
1498        assert_eq!("test", column_schema.name);
1499        assert_eq!(ConcreteDataType::json_datatype(), column_schema.data_type);
1500        assert!(column_schema.is_nullable);
1501        assert!(!column_schema.is_time_index);
1502        assert!(column_schema.default_constraint.is_none());
1503        assert_eq!(
1504            column_schema.metadata.get(TYPE_KEY).unwrap(),
1505            &ConcreteDataType::json_datatype().name()
1506        );
1507
1508        let field = Field::new("test", ArrowDataType::Binary, true);
1509        let field = field.with_metadata(Metadata::from([(
1510            TYPE_KEY.to_string(),
1511            ConcreteDataType::vector_datatype(3).name(),
1512        )]));
1513        let column_schema = ColumnSchema::try_from(&field).unwrap();
1514        assert_eq!("test", column_schema.name);
1515        assert_eq!(
1516            ConcreteDataType::vector_datatype(3),
1517            column_schema.data_type
1518        );
1519        assert!(column_schema.is_nullable);
1520        assert!(!column_schema.is_time_index);
1521        assert!(column_schema.default_constraint.is_none());
1522        assert_eq!(
1523            column_schema.metadata.get(TYPE_KEY).unwrap(),
1524            &ConcreteDataType::vector_datatype(3).name()
1525        );
1526    }
1527
1528    #[test]
1529    fn test_column_schema_fix_time_index() {
1530        let field = Field::new(
1531            "test",
1532            ArrowDataType::Timestamp(TimeUnit::Second, None),
1533            false,
1534        );
1535        let field = field.with_metadata(Metadata::from([(
1536            TIME_INDEX_KEY.to_string(),
1537            "true".to_string(),
1538        )]));
1539        let column_schema = ColumnSchema::try_from(&field).unwrap();
1540        assert_eq!("test", column_schema.name);
1541        assert_eq!(
1542            ConcreteDataType::timestamp_second_datatype(),
1543            column_schema.data_type
1544        );
1545        assert!(!column_schema.is_nullable);
1546        assert!(column_schema.is_time_index);
1547        assert!(column_schema.default_constraint.is_none());
1548        assert_eq!(1, column_schema.metadata().len());
1549
1550        let field = Field::new("test", ArrowDataType::Int32, false);
1551        let field = field.with_metadata(Metadata::from([(
1552            TIME_INDEX_KEY.to_string(),
1553            "true".to_string(),
1554        )]));
1555        let column_schema = ColumnSchema::try_from(&field).unwrap();
1556        assert_eq!("test", column_schema.name);
1557        assert_eq!(ConcreteDataType::int32_datatype(), column_schema.data_type);
1558        assert!(!column_schema.is_nullable);
1559        assert!(!column_schema.is_time_index);
1560        assert!(column_schema.default_constraint.is_none());
1561        assert!(column_schema.metadata.is_empty());
1562    }
1563
1564    #[test]
1565    fn test_compression_options_from_map() {
1566        let parse = |pairs: &[(&str, &str)]| {
1567            CompressionOptions::try_from(
1568                pairs
1569                    .iter()
1570                    .map(|(k, v)| (k.to_string(), v.to_string()))
1571                    .collect::<HashMap<_, _>>(),
1572            )
1573        };
1574
1575        let options = parse(&[("type", "ZSTD"), ("level", "3"), ("page_size", "8MiB")]).unwrap();
1576        assert_eq!(
1577            CompressionOptions {
1578                compression_type: CompressionType::Zstd,
1579                level: Some(3),
1580                page_size: Some(ReadableSize::mb(8)),
1581                encoding: None,
1582            },
1583            options
1584        );
1585        // Stored in column metadata; this format must stay readable.
1586        assert_eq!(
1587            r#"{"type":"zstd","level":3,"page_size":"8MiB"}"#,
1588            serde_json::to_string(&options).unwrap()
1589        );
1590        assert_eq!(
1591            options,
1592            parse(
1593                &options
1594                    .to_options_map()
1595                    .iter()
1596                    .map(|(k, v)| (k.as_str(), v.as_str()))
1597                    .collect::<Vec<_>>()
1598            )
1599            .unwrap()
1600        );
1601
1602        let odd_size = parse(&[("page_size", "1.25MiB")]).unwrap();
1603        assert_eq!(
1604            odd_size,
1605            parse(&[("page_size", odd_size.to_options_map()["page_size"].as_str())]).unwrap()
1606        );
1607
1608        let page_only = parse(&[("page_size", "1MiB")]).unwrap();
1609        assert_eq!(None, page_only.level);
1610        assert_eq!(
1611            r#"{"type":"zstd","page_size":"1MiB"}"#,
1612            serde_json::to_string(&page_only).unwrap()
1613        );
1614
1615        let delta = parse(&[("encoding", "DELTA_BINARY_PACKED")]).unwrap();
1616        assert_eq!(Some(ColumnEncoding::DeltaBinaryPacked), delta.encoding);
1617        assert_eq!(
1618            r#"{"type":"zstd","encoding":"delta_binary_packed"}"#,
1619            serde_json::to_string(&delta).unwrap()
1620        );
1621        assert_eq!(
1622            delta,
1623            parse(&[("encoding", delta.to_options_map()["encoding"].as_str())]).unwrap()
1624        );
1625
1626        for invalid in [
1627            &[("type", "lz4")][..],
1628            &[("level", "0")],
1629            &[("level", "23")],
1630            &[("level", "high")],
1631            &[("page_size", "0")],
1632            &[("page_size", "large")],
1633            &[("encoding", "dictionary")],
1634        ] {
1635            assert!(parse(invalid).is_err(), "{invalid:?}");
1636        }
1637    }
1638
1639    #[test]
1640    fn test_compression_options_reject_nested_types() {
1641        let options = CompressionOptions {
1642            level: Some(3),
1643            ..Default::default()
1644        };
1645        let string_column = ColumnSchema::new("s", ConcreteDataType::string_datatype(), true)
1646            .with_compression_options(options)
1647            .unwrap();
1648        assert_eq!(Some(options), string_column.compression_options().unwrap());
1649
1650        let struct_type =
1651            ConcreteDataType::struct_datatype(StructType::new(Arc::new(vec![StructField::new(
1652                "a".to_string(),
1653                ConcreteDataType::string_datatype(),
1654                true,
1655            )])));
1656        let list_type =
1657            ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::string_datatype()));
1658        for data_type in [struct_type, list_type] {
1659            assert!(
1660                ColumnSchema::new("n", data_type, true)
1661                    .with_compression_options(options)
1662                    .is_err()
1663            );
1664        }
1665    }
1666
1667    #[test]
1668    fn test_delta_encoding_requires_integer_or_timestamp() {
1669        let options = CompressionOptions {
1670            encoding: Some(ColumnEncoding::DeltaBinaryPacked),
1671            ..Default::default()
1672        };
1673        for data_type in [
1674            ConcreteDataType::int32_datatype(),
1675            ConcreteDataType::uint64_datatype(),
1676            ConcreteDataType::timestamp_nanosecond_datatype(),
1677        ] {
1678            assert!(
1679                ColumnSchema::new("c", data_type, true)
1680                    .with_compression_options(options)
1681                    .is_ok()
1682            );
1683        }
1684        for data_type in [
1685            ConcreteDataType::string_datatype(),
1686            ConcreteDataType::float64_datatype(),
1687        ] {
1688            assert!(
1689                ColumnSchema::new("c", data_type, true)
1690                    .with_compression_options(options)
1691                    .is_err()
1692            );
1693        }
1694    }
1695
1696    #[test]
1697    fn test_skipping_index_options_deserialization() {
1698        let original_options = "{\"granularity\":1024,\"false-positive-rate-in-10000\":10,\"index-type\":\"BloomFilter\"}";
1699        let options = serde_json::from_str::<SkippingIndexOptions>(original_options).unwrap();
1700        assert_eq!(1024, options.granularity);
1701        assert_eq!(SkippingIndexType::BloomFilter, options.index_type);
1702        assert_eq!(0.001, options.false_positive_rate());
1703
1704        let options_str = serde_json::to_string(&options).unwrap();
1705        assert_eq!(options_str, original_options);
1706    }
1707
1708    #[test]
1709    fn test_skipping_index_options_deserialization_v0_14_to_v0_15() {
1710        let options = "{\"granularity\":10240,\"index-type\":\"BloomFilter\"}";
1711        let options = serde_json::from_str::<SkippingIndexOptions>(options).unwrap();
1712        assert_eq!(10240, options.granularity);
1713        assert_eq!(SkippingIndexType::BloomFilter, options.index_type);
1714        assert_eq!(DEFAULT_FALSE_POSITIVE_RATE, options.false_positive_rate());
1715
1716        let options_str = serde_json::to_string(&options).unwrap();
1717        assert_eq!(
1718            options_str,
1719            "{\"granularity\":10240,\"false-positive-rate-in-10000\":100,\"index-type\":\"BloomFilter\"}"
1720        );
1721    }
1722
1723    #[test]
1724    fn test_fulltext_options_deserialization() {
1725        let original_options = "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\",\"granularity\":1024,\"false-positive-rate-in-10000\":10}";
1726        let options = serde_json::from_str::<FulltextOptions>(original_options).unwrap();
1727        assert!(!options.case_sensitive);
1728        assert!(options.enable);
1729        assert_eq!(FulltextBackend::Bloom, options.backend);
1730        assert_eq!(FulltextAnalyzer::default(), options.analyzer);
1731        assert_eq!(1024, options.granularity);
1732        assert_eq!(0.001, options.false_positive_rate());
1733
1734        let options_str = serde_json::to_string(&options).unwrap();
1735        assert_eq!(options_str, original_options);
1736    }
1737
1738    #[test]
1739    fn test_fulltext_options_deserialization_v0_14_to_v0_15() {
1740        // 0.14 to 0.15
1741        let options = "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\"}";
1742        let options = serde_json::from_str::<FulltextOptions>(options).unwrap();
1743        assert!(!options.case_sensitive);
1744        assert!(options.enable);
1745        assert_eq!(FulltextBackend::Bloom, options.backend);
1746        assert_eq!(FulltextAnalyzer::default(), options.analyzer);
1747        assert_eq!(DEFAULT_GRANULARITY, options.granularity);
1748        assert_eq!(DEFAULT_FALSE_POSITIVE_RATE, options.false_positive_rate());
1749
1750        let options_str = serde_json::to_string(&options).unwrap();
1751        assert_eq!(
1752            options_str,
1753            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\",\"granularity\":10240,\"false-positive-rate-in-10000\":100}"
1754        );
1755    }
1756}
datatypes/schema/column_schema.rs

datatypes/schema/
column_schema.rs
