1use crate::core::edge_type::{
5 MAX_SCHEMA_TYPE_ID, VIRTUAL_EDGE_TYPE_ID_SENTINEL, VIRTUAL_EDGE_TYPE_ID_START,
6 is_schemaless_edge_type, make_schemaless_id,
7};
8use crate::sync::{acquire_read, acquire_write};
9use anyhow::{Result, anyhow};
10use chrono::{DateTime, Utc};
11use object_store::ObjectStore;
12use object_store::ObjectStoreExt;
13use object_store::local::LocalFileSystem;
14use object_store::path::Path as ObjectStorePath;
15use serde::{Deserialize, Serialize};
16use std::collections::HashMap;
17use std::path::Path;
18use std::sync::{Arc, RwLock};
19
20#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
21#[non_exhaustive]
22pub enum SchemaElementState {
23 Active,
24 Hidden {
25 since: DateTime<Utc>,
26 last_active_snapshot: String, },
28 Tombstone {
29 since: DateTime<Utc>,
30 },
31}
32
33use arrow_schema::{DataType as ArrowDataType, Field, Fields, TimeUnit};
34
35pub fn datetime_struct_fields() -> Fields {
42 Fields::from(vec![
43 Field::new(
44 "nanos_since_epoch",
45 ArrowDataType::Timestamp(TimeUnit::Nanosecond, None),
46 true,
47 ),
48 Field::new("offset_seconds", ArrowDataType::Int32, true),
49 Field::new("timezone_name", ArrowDataType::Utf8, true),
50 ])
51}
52
53pub fn time_struct_fields() -> Fields {
59 Fields::from(vec![
60 Field::new(
61 "nanos_since_midnight",
62 ArrowDataType::Time64(TimeUnit::Nanosecond),
63 true,
64 ),
65 Field::new("offset_seconds", ArrowDataType::Int32, true),
66 ])
67}
68
69pub fn is_datetime_struct(arrow_dt: &ArrowDataType) -> bool {
71 matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == datetime_struct_fields())
72}
73
74pub fn is_time_struct(arrow_dt: &ArrowDataType) -> bool {
76 matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == time_struct_fields())
77}
78
79pub fn sparse_vector_struct_fields() -> Fields {
86 Fields::from(vec![
87 Field::new(
88 "indices",
89 ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::UInt32, true))),
90 false,
91 ),
92 Field::new(
93 "values",
94 ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::Float32, true))),
95 false,
96 ),
97 ])
98}
99
100pub fn is_sparse_vector_struct(arrow_dt: &ArrowDataType) -> bool {
102 matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == sparse_vector_struct_fields())
103}
104
105pub fn raw_bytes_field_metadata() -> HashMap<String, String> {
113 HashMap::from([("uni_raw_bytes".to_string(), "true".to_string())])
114}
115
116#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
117#[non_exhaustive]
118pub enum CrdtType {
119 GCounter,
120 GSet,
121 ORSet,
122 LWWRegister,
123 LWWMap,
124 Rga,
125 VectorClock,
126 VCRegister,
127}
128
129impl CrdtType {
130 #[must_use]
142 pub fn type_name(&self) -> &'static str {
143 match self {
144 CrdtType::GCounter => "GCounter",
145 CrdtType::GSet => "GSet",
146 CrdtType::ORSet => "ORSet",
147 CrdtType::LWWRegister => "LWWRegister",
148 CrdtType::LWWMap => "LWWMap",
149 CrdtType::Rga => "Rga",
150 CrdtType::VectorClock => "VectorClock",
151 CrdtType::VCRegister => "VCRegister",
152 }
153 }
154}
155
156#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq)]
157pub enum PointType {
158 Geographic, Cartesian2D, Cartesian3D, }
162
163#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
164#[non_exhaustive]
165pub enum DataType {
166 String,
167 Int32,
168 Int64,
169 Float32,
170 Float64,
171 Bool,
172 Timestamp,
173 Date,
174 Time,
175 DateTime,
176 Duration,
177 CypherValue,
178 Bytes,
179 Point(PointType),
180 Vector {
181 dimensions: usize,
182 },
183 SparseVector {
186 dimensions: usize,
187 },
188 BinaryVector {
193 dimensions: usize,
194 },
195 Btic,
196 Crdt(CrdtType),
197 List(Box<DataType>),
198 Map(Box<DataType>, Box<DataType>),
199}
200
201impl DataType {
202 #[allow(non_upper_case_globals)]
204 pub const Float: DataType = DataType::Float64;
205 #[allow(non_upper_case_globals)]
206 pub const Int: DataType = DataType::Int64;
207
208 pub fn to_arrow(&self) -> ArrowDataType {
209 match self {
210 DataType::String => ArrowDataType::Utf8,
211 DataType::Int32 => ArrowDataType::Int32,
212 DataType::Int64 => ArrowDataType::Int64,
213 DataType::Float32 => ArrowDataType::Float32,
214 DataType::Float64 => ArrowDataType::Float64,
215 DataType::Bool => ArrowDataType::Boolean,
216 DataType::Timestamp => {
217 ArrowDataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
218 }
219 DataType::Date => ArrowDataType::Date32,
220 DataType::Time => ArrowDataType::Struct(time_struct_fields()),
221 DataType::DateTime => ArrowDataType::Struct(datetime_struct_fields()),
222 DataType::Duration => ArrowDataType::LargeBinary, DataType::CypherValue => ArrowDataType::LargeBinary, DataType::Bytes => ArrowDataType::LargeBinary, DataType::Point(pt) => match pt {
226 PointType::Geographic => ArrowDataType::Struct(Fields::from(vec![
227 Field::new("latitude", ArrowDataType::Float64, false),
228 Field::new("longitude", ArrowDataType::Float64, false),
229 Field::new("crs", ArrowDataType::Utf8, false),
230 ])),
231 PointType::Cartesian2D => ArrowDataType::Struct(Fields::from(vec![
232 Field::new("x", ArrowDataType::Float64, false),
233 Field::new("y", ArrowDataType::Float64, false),
234 Field::new("crs", ArrowDataType::Utf8, false),
235 ])),
236 PointType::Cartesian3D => ArrowDataType::Struct(Fields::from(vec![
237 Field::new("x", ArrowDataType::Float64, false),
238 Field::new("y", ArrowDataType::Float64, false),
239 Field::new("z", ArrowDataType::Float64, false),
240 Field::new("crs", ArrowDataType::Utf8, false),
241 ])),
242 },
243 DataType::Vector { dimensions } => ArrowDataType::FixedSizeList(
244 Arc::new(Field::new("item", ArrowDataType::Float32, true)),
245 *dimensions as i32,
246 ),
247 DataType::SparseVector { .. } => ArrowDataType::Struct(sparse_vector_struct_fields()),
248 DataType::BinaryVector { dimensions } => ArrowDataType::FixedSizeList(
249 Arc::new(Field::new("item", ArrowDataType::UInt8, true)),
250 *dimensions as i32,
251 ),
252 DataType::Btic => ArrowDataType::FixedSizeBinary(24),
253 DataType::Crdt(_) => ArrowDataType::Binary, DataType::List(inner) => {
255 let item = Field::new("item", inner.to_arrow(), true);
259 let item = if matches!(**inner, DataType::Bytes) {
260 item.with_metadata(raw_bytes_field_metadata())
261 } else {
262 item
263 };
264 ArrowDataType::List(Arc::new(item))
265 }
266 DataType::Map(key, value) => {
267 let value_field = if value.map_value_is_typed() {
274 let f = Field::new("value", value.to_arrow(), true);
275 if matches!(**value, DataType::Bytes) {
276 f.with_metadata(raw_bytes_field_metadata())
277 } else {
278 f
279 }
280 } else {
281 Field::new("value", ArrowDataType::LargeBinary, true)
282 };
283 ArrowDataType::List(Arc::new(Field::new(
284 "item",
285 ArrowDataType::Struct(Fields::from(vec![
286 Field::new("key", key.to_arrow(), false),
287 value_field,
288 ])),
289 true,
290 )))
291 }
292 }
293 }
294
295 pub fn map_value_is_typed(&self) -> bool {
300 matches!(
301 self,
302 DataType::String
303 | DataType::Int64
304 | DataType::Int32
305 | DataType::Float64
306 | DataType::Float32
307 | DataType::Bool
308 | DataType::Bytes
309 )
310 }
311
312 pub fn accepts(&self, value: &crate::value::Value) -> bool {
338 use crate::value::{TemporalValue, Value};
339
340 if matches!(value, Value::Null) {
342 return true;
343 }
344
345 match self {
346 DataType::CypherValue | DataType::Crdt(_) | DataType::Point(_) => true,
348
349 DataType::String => matches!(value, Value::String(_)),
350 DataType::Int32 | DataType::Int64 => matches!(value, Value::Int(_)),
351 DataType::Float32 | DataType::Float64 => {
353 matches!(value, Value::Int(_) | Value::Float(_))
354 }
355 DataType::Bool => matches!(value, Value::Bool(_)),
356
357 DataType::Timestamp => matches!(
360 value,
361 Value::String(_)
362 | Value::Int(_)
363 | Value::Temporal(
364 TemporalValue::DateTime { .. } | TemporalValue::LocalDateTime { .. }
365 )
366 ),
367 DataType::DateTime => matches!(
368 value,
369 Value::Temporal(
370 TemporalValue::DateTime { .. } | TemporalValue::LocalDateTime { .. }
371 )
372 ),
373 DataType::Date => {
374 matches!(
375 value,
376 Value::Int(_) | Value::Temporal(TemporalValue::Date { .. })
377 )
378 }
379 DataType::Time => matches!(
380 value,
381 Value::Int(_)
382 | Value::Temporal(TemporalValue::Time { .. } | TemporalValue::LocalTime { .. })
383 ),
384 DataType::Duration => {
385 matches!(value, Value::Temporal(TemporalValue::Duration { .. }))
386 }
387 DataType::Bytes => matches!(value, Value::Bytes(_)),
388 DataType::Btic => matches!(
390 value,
391 Value::String(_) | Value::List(_) | Value::Temporal(TemporalValue::Btic { .. })
392 ),
393 DataType::Vector { .. } => matches!(value, Value::Vector(_) | Value::List(_)),
396 DataType::SparseVector { .. } => {
401 matches!(value, Value::SparseVector { .. } | Value::Map(_))
402 }
403 DataType::BinaryVector { .. } => {
407 matches!(value, Value::BinaryVector(_) | Value::List(_))
408 }
409 DataType::List(_) => matches!(value, Value::List(_)),
412 DataType::Map(_, _) => matches!(value, Value::Map(_)),
413 }
414 }
415
416 pub fn check_vector_dims(&self, value: &crate::value::Value) -> Result<(), VectorDimError> {
434 use crate::value::Value;
435
436 if matches!(value, Value::Null) {
437 return Ok(());
438 }
439
440 match self {
441 DataType::Vector { dimensions } => check_dense_vector_value(value, *dimensions),
442 DataType::BinaryVector { dimensions } => check_binary_vector_value(value, *dimensions),
443 DataType::List(inner) => {
444 let DataType::Vector { dimensions } = inner.as_ref() else {
445 return Ok(());
446 };
447 let Value::List(tokens) = value else {
448 return Err(VectorDimError::NotATokenList {
449 actual: value_variant_name(value),
450 });
451 };
452 for (token, token_value) in tokens.iter().enumerate() {
453 check_dense_vector_value(token_value, *dimensions)
454 .map_err(|e| e.for_token(token))?;
455 }
456 Ok(())
457 }
458 _ => Ok(()),
459 }
460 }
461}
462
463#[derive(Debug, Clone, PartialEq, Eq)]
469pub enum VectorDimError {
470 WrongLength {
472 expected: usize,
474 actual: usize,
476 },
477 NonNumericElement {
479 index: usize,
481 },
482 NotAVector {
484 actual: &'static str,
486 },
487 TokenWrongLength {
489 token: usize,
491 expected: usize,
493 actual: usize,
495 },
496 TokenNonNumericElement {
498 token: usize,
500 index: usize,
502 },
503 TokenNotAVector {
505 token: usize,
507 actual: &'static str,
509 },
510 NotATokenList {
512 actual: &'static str,
514 },
515}
516
517impl VectorDimError {
518 fn for_token(self, token: usize) -> Self {
520 match self {
521 Self::WrongLength { expected, actual } => Self::TokenWrongLength {
522 token,
523 expected,
524 actual,
525 },
526 Self::NonNumericElement { index } => Self::TokenNonNumericElement { token, index },
527 Self::NotAVector { actual } => Self::TokenNotAVector { token, actual },
528 other => other,
529 }
530 }
531}
532
533impl std::fmt::Display for VectorDimError {
534 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
535 match self {
536 Self::WrongLength { expected, actual } => write!(
537 f,
538 "got a vector of length {actual}, expected {expected} dimensions"
539 ),
540 Self::NonNumericElement { index } => {
541 write!(f, "element {index} is not numeric")
542 }
543 Self::NotAVector { actual } => {
544 write!(f, "got a non-vector value of type {actual}")
545 }
546 Self::TokenWrongLength {
547 token,
548 expected,
549 actual,
550 } => write!(
551 f,
552 "token {token} has {actual} dimensions, expected {expected}"
553 ),
554 Self::TokenNonNumericElement { token, index } => {
555 write!(f, "token {token} element {index} is not numeric")
556 }
557 Self::TokenNotAVector { token, actual } => {
558 write!(f, "token {token} is not a vector (got {actual})")
559 }
560 Self::NotATokenList { actual } => write!(
561 f,
562 "got a non-list value of type {actual} for a multi-vector column"
563 ),
564 }
565 }
566}
567
568impl std::error::Error for VectorDimError {}
569
570pub fn check_dense_vector_value(
584 value: &crate::value::Value,
585 dimensions: usize,
586) -> Result<(), VectorDimError> {
587 use crate::value::Value;
588
589 match value {
590 Value::Null => Ok(()),
591 Value::Vector(v) => {
592 if v.len() == dimensions {
593 Ok(())
594 } else {
595 Err(VectorDimError::WrongLength {
596 expected: dimensions,
597 actual: v.len(),
598 })
599 }
600 }
601 Value::List(items) => {
602 if items.len() != dimensions {
603 return Err(VectorDimError::WrongLength {
604 expected: dimensions,
605 actual: items.len(),
606 });
607 }
608 if let Some(index) = items.iter().position(|e| !e.is_number()) {
609 return Err(VectorDimError::NonNumericElement { index });
610 }
611 Ok(())
612 }
613 other => Err(VectorDimError::NotAVector {
614 actual: value_variant_name(other),
615 }),
616 }
617}
618
619pub fn check_binary_vector_value(
633 value: &crate::value::Value,
634 dimensions: usize,
635) -> Result<(), VectorDimError> {
636 use crate::value::Value;
637
638 match value {
639 Value::Null => Ok(()),
640 Value::BinaryVector(bytes) => {
641 if bytes.len() == dimensions {
642 Ok(())
643 } else {
644 Err(VectorDimError::WrongLength {
645 expected: dimensions,
646 actual: bytes.len(),
647 })
648 }
649 }
650 Value::List(items) => {
651 if items.len() != dimensions {
652 return Err(VectorDimError::WrongLength {
653 expected: dimensions,
654 actual: items.len(),
655 });
656 }
657 if let Some(index) = items
658 .iter()
659 .position(|e| !matches!(e.as_i64(), Some(0..=255)))
660 {
661 return Err(VectorDimError::NonNumericElement { index });
662 }
663 Ok(())
664 }
665 other => Err(VectorDimError::NotAVector {
666 actual: value_variant_name(other),
667 }),
668 }
669}
670
671fn value_variant_name(value: &crate::value::Value) -> &'static str {
673 use crate::value::Value;
674
675 match value {
676 Value::Null => "Null",
677 Value::Bool(_) => "Bool",
678 Value::Int(_) => "Int",
679 Value::Float(_) => "Float",
680 Value::String(_) => "String",
681 Value::Bytes(_) => "Bytes",
682 Value::List(_) => "List",
683 Value::Map(_) => "Map",
684 Value::Node(_) => "Node",
685 Value::Edge(_) => "Edge",
686 Value::Path(_) => "Path",
687 Value::Vector(_) => "Vector",
688 Value::SparseVector { .. } => "SparseVector",
689 Value::BinaryVector(_) => "BinaryVector",
690 Value::Temporal(_) => "Temporal",
691 }
692}
693
694fn default_created_at() -> DateTime<Utc> {
695 Utc::now()
696}
697
698fn default_state() -> SchemaElementState {
699 SchemaElementState::Active
700}
701
702fn default_version_1() -> u32 {
703 1
704}
705
706#[derive(Clone, Debug, Serialize, Deserialize)]
707pub struct PropertyMeta {
708 pub r#type: DataType,
709 pub nullable: bool,
710 #[serde(default = "default_version_1")]
711 pub added_in: u32, #[serde(default = "default_state")]
713 pub state: SchemaElementState,
714 #[serde(default)]
715 pub generation_expression: Option<String>,
716 #[serde(default, skip_serializing_if = "Option::is_none")]
717 pub description: Option<String>,
718}
719
720#[derive(Clone, Debug, Serialize, Deserialize)]
721pub struct LabelMeta {
722 pub id: u16, #[serde(default = "default_created_at")]
724 pub created_at: DateTime<Utc>,
725 #[serde(default = "default_state")]
726 pub state: SchemaElementState,
727 #[serde(default, skip_serializing_if = "Option::is_none")]
728 pub description: Option<String>,
729}
730
731#[derive(Clone, Debug, Serialize, Deserialize)]
732pub struct EdgeTypeMeta {
733 pub id: u32,
735 pub src_labels: Vec<String>,
736 pub dst_labels: Vec<String>,
737 #[serde(default = "default_state")]
738 pub state: SchemaElementState,
739 #[serde(default, skip_serializing_if = "Option::is_none")]
740 pub description: Option<String>,
741}
742
743#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
744#[non_exhaustive]
745pub enum ConstraintType {
746 Unique {
747 properties: Vec<String>,
748 },
749 Exists {
750 property: String,
751 },
752 Check {
753 expression: String,
754 },
755 NodeKey {
759 properties: Vec<String>,
760 },
761}
762
763impl ConstraintType {
764 #[must_use]
770 pub fn unique_properties(&self) -> Option<&[String]> {
771 match self {
772 ConstraintType::Unique { properties } | ConstraintType::NodeKey { properties } => {
773 Some(properties)
774 }
775 _ => None,
776 }
777 }
778}
779
780#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
781#[non_exhaustive]
782pub enum ConstraintTarget {
783 Label(String),
784 EdgeType(String),
785}
786
787#[derive(Clone, Debug, Serialize, Deserialize)]
788pub struct Constraint {
789 pub name: String,
790 pub constraint_type: ConstraintType,
791 pub target: ConstraintTarget,
792 pub enabled: bool,
793}
794
795#[derive(Clone, Debug, Serialize, Deserialize)]
801pub struct SchemalessEdgeTypeRegistry {
802 name_to_id: HashMap<String, u32>,
803 id_to_name: HashMap<u32, String>,
804 next_local_id: u32,
806}
807
808impl SchemalessEdgeTypeRegistry {
809 pub fn new() -> Self {
810 Self {
811 name_to_id: HashMap::new(),
812 id_to_name: HashMap::new(),
813 next_local_id: 1,
814 }
815 }
816
817 pub fn get_or_assign_id(&mut self, type_name: &str) -> u32 {
819 if let Some(&id) = self.name_to_id.get(type_name) {
820 return id;
821 }
822
823 let id = make_schemaless_id(self.next_local_id);
824 self.next_local_id += 1;
825
826 self.name_to_id.insert(type_name.to_string(), id);
827 self.id_to_name.insert(id, type_name.to_string());
828
829 id
830 }
831
832 pub fn type_name_by_id(&self, type_id: u32) -> Option<&str> {
834 self.id_to_name.get(&type_id).map(String::as_str)
835 }
836
837 pub fn contains(&self, type_name: &str) -> bool {
839 self.name_to_id.contains_key(type_name)
840 }
841
842 pub fn id_by_name(&self, type_name: &str) -> Option<u32> {
844 self.name_to_id.get(type_name).copied()
845 }
846
847 pub fn id_by_name_case_insensitive(&self, type_name: &str) -> Option<u32> {
849 self.name_to_id
850 .iter()
851 .find(|(k, _)| k.eq_ignore_ascii_case(type_name))
852 .map(|(_, &id)| id)
853 }
854
855 pub fn all_type_ids(&self) -> Vec<u32> {
857 self.id_to_name.keys().copied().collect()
858 }
859
860 pub fn is_empty(&self) -> bool {
862 self.name_to_id.is_empty()
863 }
864}
865
866impl Default for SchemalessEdgeTypeRegistry {
867 fn default() -> Self {
868 Self::new()
869 }
870}
871
872pub const VIRTUAL_LABEL_ID_START: u16 = 0xFF00;
878pub const VIRTUAL_LABEL_ID_SENTINEL: u16 = 0xFFFF;
880
881const MAX_SCHEMA_NAME_LEN: usize = 255;
886
887#[inline]
889pub fn is_virtual_label_id(id: u16) -> bool {
890 (VIRTUAL_LABEL_ID_START..VIRTUAL_LABEL_ID_SENTINEL).contains(&id)
891}
892
893#[derive(Clone, Debug, Serialize, Deserialize)]
894pub struct Schema {
895 pub schema_version: u32,
896 pub labels: HashMap<String, LabelMeta>,
897 pub edge_types: HashMap<String, EdgeTypeMeta>,
898 pub properties: HashMap<String, HashMap<String, PropertyMeta>>,
899 #[serde(default)]
900 pub indexes: Vec<IndexDefinition>,
901 #[serde(default)]
902 pub constraints: Vec<Constraint>,
903 #[serde(default)]
905 pub schemaless_registry: SchemalessEdgeTypeRegistry,
906}
907
908impl Default for Schema {
909 fn default() -> Self {
910 Self {
911 schema_version: 1,
912 labels: HashMap::new(),
913 edge_types: HashMap::new(),
914 properties: HashMap::new(),
915 indexes: Vec::new(),
916 constraints: Vec::new(),
917 schemaless_registry: SchemalessEdgeTypeRegistry::new(),
918 }
919 }
920}
921
922impl Schema {
923 fn bump_version(&mut self) {
932 self.schema_version = self.schema_version.wrapping_add(1);
933 }
934
935 pub fn label_name_by_id(&self, label_id: u16) -> Option<&str> {
940 self.labels
941 .iter()
942 .find(|(_, meta)| meta.id == label_id)
943 .map(|(name, _)| name.as_str())
944 }
945
946 pub fn label_id_by_name(&self, label_name: &str) -> Option<u16> {
948 self.labels.get(label_name).map(|meta| meta.id)
949 }
950
951 pub fn edge_type_name_by_id(&self, type_id: u32) -> Option<&str> {
956 self.edge_types
957 .iter()
958 .find(|(_, meta)| meta.id == type_id)
959 .map(|(name, _)| name.as_str())
960 }
961
962 pub fn edge_type_id_by_name(&self, type_name: &str) -> Option<u32> {
964 self.edge_types.get(type_name).map(|meta| meta.id)
965 }
966
967 pub fn vector_index_for_property(
972 &self,
973 label: &str,
974 property: &str,
975 ) -> Option<&VectorIndexConfig> {
976 self.indexes.iter().find_map(|idx| {
977 if let IndexDefinition::Vector(config) = idx
978 && config.label == label
979 && config.property == property
980 && config.metadata.status == IndexStatus::Online
981 {
982 return Some(config);
983 }
984 None
985 })
986 }
987
988 pub fn sparse_index_for_property(
990 &self,
991 label: &str,
992 property: &str,
993 ) -> Option<&SparseVectorIndexConfig> {
994 self.indexes.iter().find_map(|idx| {
995 if let IndexDefinition::Sparse(config) = idx
996 && config.label == label
997 && config.property == property
998 && config.metadata.status == IndexStatus::Online
999 {
1000 return Some(config);
1001 }
1002 None
1003 })
1004 }
1005
1006 pub fn fulltext_index_for_property(
1011 &self,
1012 label: &str,
1013 property: &str,
1014 ) -> Option<&FullTextIndexConfig> {
1015 self.indexes.iter().find_map(|idx| {
1016 if let IndexDefinition::FullText(config) = idx
1017 && config.label == label
1018 && config.properties.iter().any(|p| p == property)
1019 && config.metadata.status == IndexStatus::Online
1020 {
1021 return Some(config);
1022 }
1023 None
1024 })
1025 }
1026
1027 pub fn get_label_case_insensitive(&self, name: &str) -> Option<&LabelMeta> {
1032 self.labels
1033 .iter()
1034 .find(|(k, _)| k.eq_ignore_ascii_case(name))
1035 .map(|(_, v)| v)
1036 }
1037
1038 pub fn canonical_label_name(&self, name: &str) -> Option<String> {
1045 self.labels
1046 .iter()
1047 .find(|(k, _)| k.eq_ignore_ascii_case(name))
1048 .map(|(k, _)| k.clone())
1049 }
1050
1051 pub fn label_id_by_name_case_insensitive(&self, label_name: &str) -> Option<u16> {
1053 self.get_label_case_insensitive(label_name)
1054 .map(|meta| meta.id)
1055 }
1056
1057 pub fn get_edge_type_case_insensitive(&self, name: &str) -> Option<&EdgeTypeMeta> {
1062 self.edge_types
1063 .iter()
1064 .find(|(k, _)| k.eq_ignore_ascii_case(name))
1065 .map(|(_, v)| v)
1066 }
1067
1068 pub fn edge_type_id_by_name_case_insensitive(&self, type_name: &str) -> Option<u32> {
1070 self.get_edge_type_case_insensitive(type_name)
1071 .map(|meta| meta.id)
1072 }
1073
1074 pub fn edge_type_id_unified_case_insensitive(&self, type_name: &str) -> Option<u32> {
1077 self.edge_type_id_by_name_case_insensitive(type_name)
1078 .or_else(|| {
1079 self.schemaless_registry
1080 .id_by_name_case_insensitive(type_name)
1081 })
1082 }
1083
1084 pub fn get_or_assign_edge_type_id(&mut self, type_name: &str) -> u32 {
1090 if let Some(id) = self.edge_type_id_unified(type_name) {
1091 return id;
1092 }
1093 let id = self.schemaless_registry.get_or_assign_id(type_name);
1101 self.bump_version();
1102 id
1103 }
1104
1105 pub fn edge_type_id_unified(&self, type_name: &str) -> Option<u32> {
1112 self.edge_type_id_by_name(type_name)
1113 .or_else(|| self.schemaless_registry.id_by_name(type_name))
1114 }
1115
1116 pub fn edge_type_name_by_id_unified(&self, type_id: u32) -> Option<String> {
1120 if is_schemaless_edge_type(type_id) {
1121 self.schemaless_registry
1122 .type_name_by_id(type_id)
1123 .map(str::to_owned)
1124 } else {
1125 self.edge_type_name_by_id(type_id).map(str::to_owned)
1126 }
1127 }
1128
1129 pub fn all_edge_type_ids(&self) -> Vec<u32> {
1132 let mut ids: Vec<u32> = self.edge_types.values().map(|m| m.id).collect();
1133 ids.extend(self.schemaless_registry.all_type_ids());
1134 ids.sort_unstable();
1135 ids
1136 }
1137}
1138
1139#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
1141pub enum IndexStatus {
1142 #[default]
1144 Online,
1145 Building,
1147 Stale,
1149 Failed,
1151}
1152
1153#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
1155pub struct IndexMetadata {
1156 #[serde(default)]
1158 pub status: IndexStatus,
1159 #[serde(default)]
1161 pub last_built_at: Option<DateTime<Utc>>,
1162 #[serde(default)]
1164 pub row_count_at_build: Option<u64>,
1165}
1166
1167#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1168#[serde(tag = "type")]
1169#[non_exhaustive]
1170pub enum IndexDefinition {
1171 Vector(VectorIndexConfig),
1172 FullText(FullTextIndexConfig),
1173 Scalar(ScalarIndexConfig),
1174 Inverted(InvertedIndexConfig),
1175 JsonFullText(JsonFtsIndexConfig),
1176 Sparse(SparseVectorIndexConfig),
1178}
1179
1180impl IndexDefinition {
1181 pub fn name(&self) -> &str {
1183 match self {
1184 IndexDefinition::Vector(c) => &c.name,
1185 IndexDefinition::FullText(c) => &c.name,
1186 IndexDefinition::Scalar(c) => &c.name,
1187 IndexDefinition::Inverted(c) => &c.name,
1188 IndexDefinition::JsonFullText(c) => &c.name,
1189 IndexDefinition::Sparse(c) => &c.name,
1190 }
1191 }
1192
1193 pub fn label(&self) -> &str {
1195 match self {
1196 IndexDefinition::Vector(c) => &c.label,
1197 IndexDefinition::FullText(c) => &c.label,
1198 IndexDefinition::Scalar(c) => &c.label,
1199 IndexDefinition::Inverted(c) => &c.label,
1200 IndexDefinition::JsonFullText(c) => &c.label,
1201 IndexDefinition::Sparse(c) => &c.label,
1202 }
1203 }
1204
1205 pub fn metadata(&self) -> &IndexMetadata {
1207 match self {
1208 IndexDefinition::Vector(c) => &c.metadata,
1209 IndexDefinition::FullText(c) => &c.metadata,
1210 IndexDefinition::Scalar(c) => &c.metadata,
1211 IndexDefinition::Inverted(c) => &c.metadata,
1212 IndexDefinition::JsonFullText(c) => &c.metadata,
1213 IndexDefinition::Sparse(c) => &c.metadata,
1214 }
1215 }
1216
1217 pub fn metadata_mut(&mut self) -> &mut IndexMetadata {
1219 match self {
1220 IndexDefinition::Vector(c) => &mut c.metadata,
1221 IndexDefinition::FullText(c) => &mut c.metadata,
1222 IndexDefinition::Scalar(c) => &mut c.metadata,
1223 IndexDefinition::Inverted(c) => &mut c.metadata,
1224 IndexDefinition::JsonFullText(c) => &mut c.metadata,
1225 IndexDefinition::Sparse(c) => &mut c.metadata,
1226 }
1227 }
1228}
1229
1230#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1231pub struct InvertedIndexConfig {
1232 pub name: String,
1233 pub label: String,
1234 pub property: String,
1235 #[serde(default = "default_normalize")]
1236 pub normalize: bool,
1237 #[serde(default = "default_max_terms_per_doc")]
1238 pub max_terms_per_doc: usize,
1239 #[serde(default)]
1240 pub metadata: IndexMetadata,
1241}
1242
1243fn default_normalize() -> bool {
1244 true
1245}
1246
1247fn default_max_terms_per_doc() -> usize {
1248 10_000
1249}
1250
1251#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1258pub struct SparseVectorIndexConfig {
1259 pub name: String,
1260 pub label: String,
1261 pub property: String,
1262 pub dimensions: usize,
1264 #[serde(default = "default_sparse_quantize")]
1266 pub quantize: bool,
1267 #[serde(default)]
1271 pub embedding_config: Option<EmbeddingConfig>,
1272 #[serde(default)]
1273 pub metadata: IndexMetadata,
1274}
1275
1276fn default_sparse_quantize() -> bool {
1277 true
1278}
1279
1280#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1281pub struct VectorIndexConfig {
1282 pub name: String,
1283 pub label: String,
1284 pub property: String,
1285 pub index_type: VectorIndexType,
1286 pub metric: DistanceMetric,
1287 pub embedding_config: Option<EmbeddingConfig>,
1288 #[serde(default)]
1289 pub metadata: IndexMetadata,
1290}
1291
1292#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1293pub struct EmbeddingConfig {
1294 pub alias: String,
1296 pub source_properties: Vec<String>,
1297 pub batch_size: usize,
1298 #[serde(default)]
1301 pub document_prefix: Option<String>,
1302 #[serde(default)]
1305 pub query_prefix: Option<String>,
1306}
1307
1308#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1309#[non_exhaustive]
1310pub enum VectorIndexType {
1311 Flat,
1312 IvfFlat {
1313 num_partitions: u32,
1314 },
1315 IvfPq {
1316 num_partitions: u32,
1317 num_sub_vectors: u32,
1318 bits_per_subvector: u8,
1319 },
1320 IvfSq {
1321 num_partitions: u32,
1322 },
1323 IvfRq {
1324 num_partitions: u32,
1325 #[serde(default)]
1326 num_bits: Option<u8>,
1327 },
1328 HnswFlat {
1329 m: u32,
1330 ef_construction: u32,
1331 #[serde(default)]
1332 num_partitions: Option<u32>,
1333 },
1334 HnswSq {
1335 m: u32,
1336 ef_construction: u32,
1337 #[serde(default)]
1338 num_partitions: Option<u32>,
1339 },
1340 HnswPq {
1341 m: u32,
1342 ef_construction: u32,
1343 num_sub_vectors: u32,
1344 #[serde(default)]
1345 num_partitions: Option<u32>,
1346 },
1347 Muvera {
1354 k_sim: u32,
1356 reps: u32,
1358 d_proj: u32,
1360 seed: u64,
1362 inner: Box<VectorIndexType>,
1364 },
1365}
1366
1367#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1368#[non_exhaustive]
1369pub enum DistanceMetric {
1370 Cosine,
1371 L2,
1372 Dot,
1373 L1,
1376 Hamming,
1381 Jaccard,
1386}
1387
1388impl DistanceMetric {
1389 pub fn compute_distance(&self, a: &[f32], b: &[f32]) -> f32 {
1402 assert_eq!(a.len(), b.len(), "vector dimension mismatch");
1403 match self {
1404 DistanceMetric::L2 => a.iter().zip(b).map(|(x, y)| (x - y).powi(2)).sum(),
1405 DistanceMetric::L1 => a.iter().zip(b).map(|(x, y)| (x - y).abs()).sum(),
1406 DistanceMetric::Cosine => {
1407 let dot: f32 = a.iter().zip(b).map(|(x, y)| x * y).sum();
1408 let norm_a: f32 = a.iter().map(|x| x.powi(2)).sum::<f32>().sqrt();
1409 let norm_b: f32 = b.iter().map(|x| x.powi(2)).sum::<f32>().sqrt();
1410 let denom = norm_a * norm_b;
1411 if denom == 0.0 { 1.0 } else { 1.0 - dot / denom }
1412 }
1413 DistanceMetric::Dot => {
1414 let dot: f32 = a.iter().zip(b).map(|(x, y)| x * y).sum();
1415 -dot
1416 }
1417 DistanceMetric::Hamming | DistanceMetric::Jaccard => {
1421 panic!("{self:?} is a binary-vector metric; use compute_distance_binary")
1422 }
1423 }
1424 }
1425
1426 pub fn is_binary(&self) -> bool {
1434 matches!(self, DistanceMetric::Hamming | DistanceMetric::Jaccard)
1435 }
1436
1437 pub fn compute_distance_binary(&self, a: &[u8], b: &[u8]) -> f32 {
1450 assert_eq!(a.len(), b.len(), "binary vector dimension mismatch");
1451 match self {
1452 DistanceMetric::Hamming => a
1453 .iter()
1454 .zip(b)
1455 .map(|(x, y)| (x ^ y).count_ones())
1456 .sum::<u32>() as f32,
1457 DistanceMetric::Jaccard => {
1458 let mut inter: u32 = 0;
1459 let mut union: u32 = 0;
1460 for (x, y) in a.iter().zip(b) {
1461 inter += (x & y).count_ones();
1462 union += (x | y).count_ones();
1463 }
1464 if union == 0 {
1465 0.0
1466 } else {
1467 1.0 - (inter as f32) / (union as f32)
1468 }
1469 }
1470 other => panic!("{other:?} is a float-vector metric; use compute_distance"),
1471 }
1472 }
1473}
1474
1475#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1476pub struct FullTextIndexConfig {
1477 pub name: String,
1478 pub label: String,
1479 pub properties: Vec<String>,
1480 pub tokenizer: TokenizerConfig,
1481 pub with_positions: bool,
1482 #[serde(default)]
1483 pub metadata: IndexMetadata,
1484}
1485
1486#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1487#[non_exhaustive]
1488pub enum TokenizerConfig {
1489 Standard,
1490 Whitespace,
1491 Ngram {
1492 min: u8,
1493 max: u8,
1494 },
1495 Custom {
1496 name: String,
1497 },
1498 Analyzer(AnalyzerConfig),
1504}
1505
1506#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1515pub struct AnalyzerConfig {
1516 #[serde(default)]
1518 pub base: BaseTokenizer,
1519 #[serde(default)]
1521 pub language: FtsLanguage,
1522 #[serde(default = "default_true")]
1524 pub lower_case: bool,
1525 #[serde(default = "default_true")]
1527 pub stem: bool,
1528 #[serde(default = "default_true")]
1530 pub remove_stop_words: bool,
1531 #[serde(default)]
1533 pub custom_stop_words: Option<Vec<String>>,
1534 #[serde(default = "default_true")]
1536 pub ascii_folding: bool,
1537 #[serde(default)]
1539 pub max_token_length: Option<u32>,
1540}
1541
1542impl Default for AnalyzerConfig {
1543 fn default() -> Self {
1544 Self {
1545 base: BaseTokenizer::default(),
1546 language: FtsLanguage::default(),
1547 lower_case: true,
1548 stem: true,
1549 remove_stop_words: true,
1550 custom_stop_words: None,
1551 ascii_folding: true,
1552 max_token_length: None,
1553 }
1554 }
1555}
1556
1557#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
1562#[non_exhaustive]
1563pub enum BaseTokenizer {
1564 #[default]
1566 Simple,
1567 Whitespace,
1569 Raw,
1571 Ngram {
1573 min: u32,
1575 max: u32,
1577 },
1578 Custom(String),
1580}
1581
1582#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
1588#[non_exhaustive]
1589pub enum FtsLanguage {
1590 Arabic,
1592 Danish,
1594 Dutch,
1596 #[default]
1598 English,
1599 Finnish,
1601 French,
1603 German,
1605 Greek,
1607 Hungarian,
1609 Italian,
1611 Norwegian,
1613 Portuguese,
1615 Romanian,
1617 Russian,
1619 Spanish,
1621 Swedish,
1623 Tamil,
1625 Turkish,
1627}
1628
1629fn default_true() -> bool {
1631 true
1632}
1633
1634#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1635pub struct JsonFtsIndexConfig {
1636 pub name: String,
1637 pub label: String,
1638 pub column: String,
1639 #[serde(default)]
1640 pub paths: Vec<String>,
1641 #[serde(default)]
1642 pub with_positions: bool,
1643 #[serde(default)]
1644 pub metadata: IndexMetadata,
1645}
1646
1647#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1648pub struct ScalarIndexConfig {
1649 pub name: String,
1650 pub label: String,
1651 pub properties: Vec<String>,
1652 pub index_type: ScalarIndexType,
1653 pub where_clause: Option<String>,
1654 #[serde(default)]
1655 pub metadata: IndexMetadata,
1656}
1657
1658#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1659#[non_exhaustive]
1660pub enum ScalarIndexType {
1661 BTree,
1662 Hash,
1663 Bitmap,
1664 LabelList,
1665}
1666
1667pub struct SchemaManager {
1668 store: Arc<dyn ObjectStore>,
1669 path: ObjectStorePath,
1670 schema: RwLock<Arc<Schema>>,
1671}
1672
1673impl SchemaManager {
1674 pub async fn load(path: impl AsRef<Path>) -> Result<Self> {
1675 let path = path.as_ref();
1676 let parent = path
1677 .parent()
1678 .ok_or_else(|| anyhow!("Invalid schema path"))?;
1679 let filename = path
1680 .file_name()
1681 .ok_or_else(|| anyhow!("Invalid schema filename"))?
1682 .to_str()
1683 .ok_or_else(|| anyhow!("Invalid utf8 filename"))?;
1684
1685 let store = Arc::new(LocalFileSystem::new_with_prefix(parent)?);
1686 let obj_path = ObjectStorePath::from(filename);
1687
1688 Self::load_from_store(store, &obj_path).await
1689 }
1690
1691 pub async fn load_from_store(
1692 store: Arc<dyn ObjectStore>,
1693 path: &ObjectStorePath,
1694 ) -> Result<Self> {
1695 match store.get(path).await {
1696 Ok(result) => {
1697 let bytes = result.bytes().await?;
1698 let content = String::from_utf8(bytes.to_vec())?;
1699 let mut schema: Schema = serde_json::from_str(&content)?;
1700 let original_len = schema.indexes.len();
1708 if original_len > 0 {
1709 let mut seen: std::collections::HashSet<String> =
1710 std::collections::HashSet::with_capacity(original_len);
1711 let mut dedup: Vec<IndexDefinition> = schema
1712 .indexes
1713 .iter()
1714 .rev()
1715 .filter(|idx| seen.insert(idx.name().to_string()))
1716 .cloned()
1717 .collect();
1718 dedup.reverse();
1719 if dedup.len() != original_len {
1720 tracing::warn!(
1721 collapsed = original_len - dedup.len(),
1722 kept = dedup.len(),
1723 "schema.indexes: collapsed duplicate entries on load (issue #63)"
1724 );
1725 schema.indexes = dedup;
1726 }
1727 }
1728 Ok(Self {
1729 store,
1730 path: path.clone(),
1731 schema: RwLock::new(Arc::new(schema)),
1732 })
1733 }
1734 Err(object_store::Error::NotFound { .. }) => Ok(Self {
1735 store,
1736 path: path.clone(),
1737 schema: RwLock::new(Arc::new(Schema::default())),
1738 }),
1739 Err(e) => Err(anyhow::Error::from(e)),
1740 }
1741 }
1742
1743 pub async fn save(&self) -> Result<()> {
1744 let content = {
1745 let schema_guard = acquire_read(&self.schema, "schema")?;
1746 serde_json::to_string_pretty(&**schema_guard)?
1747 };
1748 self.store
1749 .put(&self.path, content.into())
1750 .await
1751 .map_err(anyhow::Error::from)?;
1752 Ok(())
1753 }
1754
1755 pub fn path(&self) -> &ObjectStorePath {
1756 &self.path
1757 }
1758
1759 pub fn schema(&self) -> Arc<Schema> {
1760 self.schema
1761 .read()
1762 .expect("Schema lock poisoned - a thread panicked while holding it")
1763 .clone()
1764 }
1765
1766 fn normalize_function_names(expr: &str) -> String {
1769 let mut result = String::with_capacity(expr.len());
1770 let mut chars = expr.chars().peekable();
1771
1772 while let Some(ch) = chars.next() {
1773 if ch.is_alphabetic() {
1774 let mut ident = String::new();
1776 ident.push(ch);
1777
1778 while let Some(&next) = chars.peek() {
1779 if next.is_alphanumeric() || next == '_' {
1780 ident.push(chars.next().unwrap());
1781 } else {
1782 break;
1783 }
1784 }
1785
1786 if chars.peek() == Some(&'(') {
1788 result.push_str(&ident.to_uppercase());
1789 } else {
1790 result.push_str(&ident); }
1792 } else {
1793 result.push(ch);
1794 }
1795 }
1796
1797 result
1798 }
1799
1800 pub fn generated_column_name(expr: &str) -> String {
1808 let normalized = Self::normalize_function_names(expr);
1810
1811 let sanitized = normalized
1812 .replace(|c: char| !c.is_alphanumeric(), "_")
1813 .trim_matches('_')
1814 .to_string();
1815
1816 const FNV_OFFSET_BASIS: u64 = 14695981039346656037;
1818 const FNV_PRIME: u64 = 1099511628211;
1819
1820 let mut hash = FNV_OFFSET_BASIS;
1821 for byte in normalized.as_bytes() {
1822 hash ^= *byte as u64;
1823 hash = hash.wrapping_mul(FNV_PRIME);
1824 }
1825
1826 format!("_gen_{}_{:x}", sanitized, hash)
1827 }
1828
1829 pub fn replace_schema(&self, new_schema: Schema) {
1830 let mut schema = self
1831 .schema
1832 .write()
1833 .expect("Schema lock poisoned - a thread panicked while holding it");
1834 *schema = Arc::new(new_schema);
1835 }
1836
1837 #[must_use]
1850 pub fn with_overlay(&self, overlay: &crate::core::fork::SchemaDelta) -> Arc<Self> {
1851 let primary = self.schema();
1852 let merged = if overlay.is_empty() {
1853 (*primary).clone()
1854 } else {
1855 let mut merged = (*primary).clone();
1856 for (name, label) in &overlay.added_labels {
1857 merged.labels.insert(name.clone(), label.clone());
1858 }
1859 for (name, edge_type) in &overlay.added_edge_types {
1860 merged.edge_types.insert(name.clone(), edge_type.clone());
1861 }
1862 for addition in &overlay.added_properties {
1863 let props = merged.properties.entry(addition.owner.clone()).or_default();
1864 props.insert(
1865 addition.property.clone(),
1866 PropertyMeta {
1867 r#type: addition.data_type.clone(),
1868 nullable: addition.nullable,
1869 added_in: merged.schema_version,
1870 state: SchemaElementState::Active,
1871 generation_expression: None,
1872 description: None,
1873 },
1874 );
1875 }
1876 merged
1877 };
1878
1879 Arc::new(Self {
1880 store: self.store.clone(),
1881 path: self.path.clone(),
1882 schema: RwLock::new(Arc::new(merged)),
1883 })
1884 }
1885
1886 pub fn next_label_id(&self) -> u16 {
1887 self.schema()
1888 .labels
1889 .values()
1890 .map(|l| l.id)
1891 .max()
1892 .unwrap_or(0)
1893 + 1
1894 }
1895
1896 pub fn next_type_id(&self) -> u32 {
1897 let max_schema_id = self
1898 .schema()
1899 .edge_types
1900 .values()
1901 .map(|t| t.id)
1902 .max()
1903 .unwrap_or(0);
1904
1905 if max_schema_id >= MAX_SCHEMA_TYPE_ID {
1907 panic!("Schema edge type ID exhaustion");
1908 }
1909
1910 max_schema_id + 1
1911 }
1912
1913 pub fn validate_schema_element_name(kind: &str, name: &str) -> Result<()> {
1931 if name.is_empty() || name.chars().all(char::is_whitespace) {
1932 return Err(anyhow!(
1933 "{kind} name must be non-empty and not all whitespace"
1934 ));
1935 }
1936 if name.len() > MAX_SCHEMA_NAME_LEN {
1937 return Err(anyhow!("{kind} name exceeds {MAX_SCHEMA_NAME_LEN} bytes"));
1938 }
1939 if let Some(c) = name
1940 .chars()
1941 .find(|c| c.is_control() || c.is_whitespace() || matches!(c, '/' | '\\'))
1942 {
1943 return Err(anyhow!(
1944 "{kind} name '{name}' contains an unsafe character ({c:?})"
1945 ));
1946 }
1947 Ok(())
1948 }
1949
1950 pub fn add_label(&self, name: &str) -> Result<u16> {
1951 self.add_label_with_desc(name, None)
1952 }
1953
1954 pub fn add_label_with_desc(&self, name: &str, description: Option<String>) -> Result<u16> {
1955 Self::validate_schema_element_name("Label", name)?;
1956 let mut guard = acquire_write(&self.schema, "schema")?;
1957 let schema = Arc::make_mut(&mut *guard);
1958 if schema.labels.contains_key(name) {
1959 return Err(anyhow!("Label '{}' already exists", name));
1960 }
1961
1962 let id = schema.labels.values().map(|l| l.id).max().unwrap_or(0) + 1;
1963 if id >= VIRTUAL_LABEL_ID_START {
1964 return Err(anyhow!(
1965 "Native label space exhausted (next id {id:#x} would enter the \
1966 virtual range {VIRTUAL_LABEL_ID_START:#x}..{VIRTUAL_LABEL_ID_SENTINEL:#x} \
1967 reserved for catalog-resolved labels)"
1968 ));
1969 }
1970 schema.labels.insert(
1971 name.to_string(),
1972 LabelMeta {
1973 id,
1974 created_at: Utc::now(),
1975 state: SchemaElementState::Active,
1976 description,
1977 },
1978 );
1979 schema.bump_version();
1980 Ok(id)
1981 }
1982
1983 pub fn add_edge_type(
1984 &self,
1985 name: &str,
1986 src_labels: Vec<String>,
1987 dst_labels: Vec<String>,
1988 ) -> Result<u32> {
1989 self.add_edge_type_with_desc(name, src_labels, dst_labels, None)
1990 }
1991
1992 pub fn add_edge_type_with_desc(
1993 &self,
1994 name: &str,
1995 src_labels: Vec<String>,
1996 dst_labels: Vec<String>,
1997 description: Option<String>,
1998 ) -> Result<u32> {
1999 Self::validate_schema_element_name("Edge type", name)?;
2000 let mut guard = acquire_write(&self.schema, "schema")?;
2001 let schema = Arc::make_mut(&mut *guard);
2002 if schema.edge_types.contains_key(name) {
2003 return Err(anyhow!("Edge type '{}' already exists", name));
2004 }
2005
2006 let id = schema.edge_types.values().map(|t| t.id).max().unwrap_or(0) + 1;
2007
2008 if id >= VIRTUAL_EDGE_TYPE_ID_START {
2013 return Err(anyhow!(
2014 "Native edge type space exhausted (next id {id:#x} would enter the \
2015 virtual range {VIRTUAL_EDGE_TYPE_ID_START:#x}..{VIRTUAL_EDGE_TYPE_ID_SENTINEL:#x} \
2016 reserved for catalog-resolved edge types)"
2017 ));
2018 }
2019
2020 schema.edge_types.insert(
2021 name.to_string(),
2022 EdgeTypeMeta {
2023 id,
2024 src_labels,
2025 dst_labels,
2026 state: SchemaElementState::Active,
2027 description,
2028 },
2029 );
2030 schema.bump_version();
2031 Ok(id)
2032 }
2033
2034 pub fn get_or_assign_edge_type_id(&self, type_name: &str) -> u32 {
2043 {
2044 let guard = acquire_read(&self.schema, "schema")
2045 .expect("Schema lock poisoned - a thread panicked while holding it");
2046 if let Some(id) = guard.edge_type_id_unified(type_name) {
2047 return id;
2048 }
2049 }
2050 let mut guard = acquire_write(&self.schema, "schema")
2051 .expect("Schema lock poisoned - a thread panicked while holding it");
2052 let schema = Arc::make_mut(&mut *guard);
2053 schema.get_or_assign_edge_type_id(type_name)
2054 }
2055
2056 pub fn edge_type_name_by_id_unified(&self, type_id: u32) -> Option<String> {
2058 let schema = acquire_read(&self.schema, "schema")
2059 .expect("Schema lock poisoned - a thread panicked while holding it");
2060 schema.edge_type_name_by_id_unified(type_id)
2061 }
2062
2063 pub fn add_property(
2064 &self,
2065 label_or_type: &str,
2066 prop_name: &str,
2067 data_type: DataType,
2068 nullable: bool,
2069 ) -> Result<()> {
2070 self.add_property_with_desc(label_or_type, prop_name, data_type, nullable, None)
2071 }
2072
2073 pub fn add_property_with_desc(
2074 &self,
2075 label_or_type: &str,
2076 prop_name: &str,
2077 data_type: DataType,
2078 nullable: bool,
2079 description: Option<String>,
2080 ) -> Result<()> {
2081 validate_property_name(prop_name)?;
2082 let mut guard = acquire_write(&self.schema, "schema")?;
2083 let schema = Arc::make_mut(&mut *guard);
2084 let version = schema.schema_version;
2085 let props = schema
2086 .properties
2087 .entry(label_or_type.to_string())
2088 .or_default();
2089
2090 if props.contains_key(prop_name) {
2091 return Err(anyhow!(
2092 "Property '{}' already exists for '{}'",
2093 prop_name,
2094 label_or_type
2095 ));
2096 }
2097
2098 props.insert(
2099 prop_name.to_string(),
2100 PropertyMeta {
2101 r#type: data_type,
2102 nullable,
2103 added_in: version,
2104 state: SchemaElementState::Active,
2105 generation_expression: None,
2106 description,
2107 },
2108 );
2109 schema.bump_version();
2111 Ok(())
2112 }
2113
2114 pub fn declare_property(
2132 &self,
2133 label_or_type: &str,
2134 prop_name: &str,
2135 data_type: DataType,
2136 nullable: bool,
2137 description: Option<String>,
2138 ) -> Result<bool> {
2139 validate_property_name(prop_name)?;
2140 let mut guard = acquire_write(&self.schema, "schema")?;
2141 let schema = Arc::make_mut(&mut *guard);
2142 let version = schema.schema_version;
2143 let props = schema
2144 .properties
2145 .entry(label_or_type.to_string())
2146 .or_default();
2147
2148 if let Some(existing) = props.get(prop_name) {
2149 if existing.r#type == data_type && existing.nullable == nullable {
2150 return Ok(false); }
2152 return Err(anyhow!(
2153 "Property '{}' on '{}' is declared as {:?} (nullable: {}); cannot re-declare \
2154 as {:?} (nullable: {}). Property types are immutable — use a new property \
2155 name or migrate the data",
2156 prop_name,
2157 label_or_type,
2158 existing.r#type,
2159 existing.nullable,
2160 data_type,
2161 nullable
2162 ));
2163 }
2164
2165 props.insert(
2166 prop_name.to_string(),
2167 PropertyMeta {
2168 r#type: data_type,
2169 nullable,
2170 added_in: version,
2171 state: SchemaElementState::Active,
2172 generation_expression: None,
2173 description,
2174 },
2175 );
2176 schema.bump_version();
2178 Ok(true)
2179 }
2180
2181 pub fn add_internal_property(
2192 &self,
2193 label_or_type: &str,
2194 prop_name: &str,
2195 data_type: DataType,
2196 nullable: bool,
2197 ) -> Result<bool> {
2198 validate_reserved_property_name(prop_name)?;
2199 let mut guard = acquire_write(&self.schema, "schema")?;
2200 let schema = Arc::make_mut(&mut *guard);
2201 let version = schema.schema_version;
2202 let props = schema
2203 .properties
2204 .entry(label_or_type.to_string())
2205 .or_default();
2206
2207 if let Some(existing) = props.get(prop_name) {
2208 if existing.r#type == data_type {
2209 return Ok(false); }
2211 return Err(anyhow!(
2212 "Internal property '{}' already exists for '{}' with a different type",
2213 prop_name,
2214 label_or_type
2215 ));
2216 }
2217
2218 props.insert(
2219 prop_name.to_string(),
2220 PropertyMeta {
2221 r#type: data_type,
2222 nullable,
2223 added_in: version,
2224 state: SchemaElementState::Active,
2225 generation_expression: None,
2226 description: None,
2227 },
2228 );
2229 schema.bump_version();
2230 Ok(true)
2231 }
2232
2233 pub fn add_generated_property(
2234 &self,
2235 label_or_type: &str,
2236 prop_name: &str,
2237 data_type: DataType,
2238 expr: String,
2239 ) -> Result<()> {
2240 validate_reserved_property_name(prop_name)?;
2243 let mut guard = acquire_write(&self.schema, "schema")?;
2244 let schema = Arc::make_mut(&mut *guard);
2245 let version = schema.schema_version;
2246 let props = schema
2247 .properties
2248 .entry(label_or_type.to_string())
2249 .or_default();
2250
2251 if props.contains_key(prop_name) {
2252 return Err(anyhow!("Property '{}' already exists", prop_name));
2253 }
2254
2255 props.insert(
2256 prop_name.to_string(),
2257 PropertyMeta {
2258 r#type: data_type,
2259 nullable: true,
2260 added_in: version,
2261 state: SchemaElementState::Active,
2262 generation_expression: Some(expr),
2263 description: None,
2264 },
2265 );
2266 schema.bump_version();
2268 Ok(())
2269 }
2270
2271 pub fn set_label_description(&self, name: &str, description: Option<String>) -> Result<()> {
2272 let mut guard = acquire_write(&self.schema, "schema")?;
2273 let schema = Arc::make_mut(&mut *guard);
2274 let meta = schema
2275 .labels
2276 .get_mut(name)
2277 .ok_or_else(|| anyhow!("Label '{}' does not exist", name))?;
2278 meta.description = description;
2279 Ok(())
2280 }
2281
2282 pub fn set_edge_type_description(&self, name: &str, description: Option<String>) -> Result<()> {
2283 let mut guard = acquire_write(&self.schema, "schema")?;
2284 let schema = Arc::make_mut(&mut *guard);
2285 let meta = schema
2286 .edge_types
2287 .get_mut(name)
2288 .ok_or_else(|| anyhow!("Edge type '{}' does not exist", name))?;
2289 meta.description = description;
2290 Ok(())
2291 }
2292
2293 pub fn set_property_description(
2294 &self,
2295 entity: &str,
2296 prop_name: &str,
2297 description: Option<String>,
2298 ) -> Result<()> {
2299 let mut guard = acquire_write(&self.schema, "schema")?;
2300 let schema = Arc::make_mut(&mut *guard);
2301 let props = schema
2302 .properties
2303 .get_mut(entity)
2304 .ok_or_else(|| anyhow!("Entity '{}' does not exist", entity))?;
2305 let meta = props
2306 .get_mut(prop_name)
2307 .ok_or_else(|| anyhow!("Property '{}' does not exist on '{}'", prop_name, entity))?;
2308 meta.description = description;
2309 Ok(())
2310 }
2311
2312 pub fn add_index(&self, index_def: IndexDefinition) -> Result<()> {
2321 let mut guard = acquire_write(&self.schema, "schema")?;
2322 let schema = Arc::make_mut(&mut *guard);
2323 if let Some(existing) = schema
2324 .indexes
2325 .iter_mut()
2326 .find(|i| i.name() == index_def.name())
2327 {
2328 *existing = index_def;
2329 } else {
2330 schema.indexes.push(index_def);
2331 }
2332 schema.bump_version();
2333 Ok(())
2334 }
2335
2336 pub fn get_index(&self, name: &str) -> Option<IndexDefinition> {
2337 let schema = self.schema.read().expect("Schema lock poisoned");
2338 schema.indexes.iter().find(|i| i.name() == name).cloned()
2339 }
2340
2341 pub fn update_index_metadata(
2346 &self,
2347 index_name: &str,
2348 f: impl FnOnce(&mut IndexMetadata),
2349 ) -> Result<()> {
2350 let mut guard = acquire_write(&self.schema, "schema")?;
2351 let schema = Arc::make_mut(&mut *guard);
2352 let idx = schema
2353 .indexes
2354 .iter_mut()
2355 .find(|i| i.name() == index_name)
2356 .ok_or_else(|| anyhow!("Index '{}' not found", index_name))?;
2357 f(idx.metadata_mut());
2358 Ok(())
2359 }
2360
2361 pub fn remove_index(&self, name: &str) -> Result<()> {
2362 let mut guard = acquire_write(&self.schema, "schema")?;
2363 let schema = Arc::make_mut(&mut *guard);
2364 if let Some(pos) = schema.indexes.iter().position(|i| i.name() == name) {
2365 schema.indexes.remove(pos);
2366 schema.bump_version();
2367 Ok(())
2368 } else {
2369 Err(anyhow!("Index '{}' not found", name))
2370 }
2371 }
2372
2373 pub fn add_constraint(&self, constraint: Constraint) -> Result<()> {
2374 let mut guard = acquire_write(&self.schema, "schema")?;
2375 let schema = Arc::make_mut(&mut *guard);
2376 if schema.constraints.iter().any(|c| c.name == constraint.name) {
2377 return Err(anyhow!("Constraint '{}' already exists", constraint.name));
2378 }
2379 schema.constraints.push(constraint);
2380 schema.bump_version();
2381 Ok(())
2382 }
2383
2384 pub fn drop_constraint(&self, name: &str, if_exists: bool) -> Result<()> {
2385 let mut guard = acquire_write(&self.schema, "schema")?;
2386 let schema = Arc::make_mut(&mut *guard);
2387 if let Some(pos) = schema.constraints.iter().position(|c| c.name == name) {
2388 schema.constraints.remove(pos);
2389 schema.bump_version();
2390 Ok(())
2391 } else if if_exists {
2392 Ok(())
2393 } else {
2394 Err(anyhow!("Constraint '{}' not found", name))
2395 }
2396 }
2397
2398 pub fn drop_property(&self, label_or_type: &str, prop_name: &str) -> Result<()> {
2399 let mut guard = acquire_write(&self.schema, "schema")?;
2400 let schema = Arc::make_mut(&mut *guard);
2401 let Some(props) = schema.properties.get_mut(label_or_type) else {
2402 return Err(anyhow!("Label or Edge Type '{}' not found", label_or_type));
2403 };
2404 if props.remove(prop_name).is_none() {
2405 return Err(anyhow!(
2406 "Property '{}' not found for '{}'",
2407 prop_name,
2408 label_or_type
2409 ));
2410 }
2411 schema.bump_version();
2412 Ok(())
2413 }
2414
2415 pub fn rename_property(
2416 &self,
2417 label_or_type: &str,
2418 old_name: &str,
2419 new_name: &str,
2420 ) -> Result<()> {
2421 validate_property_name(new_name)?;
2426 let mut guard = acquire_write(&self.schema, "schema")?;
2427 let schema = Arc::make_mut(&mut *guard);
2428 let Some(props) = schema.properties.get_mut(label_or_type) else {
2429 return Err(anyhow!("Label or Edge Type '{}' not found", label_or_type));
2430 };
2431 let Some(meta) = props.remove(old_name) else {
2432 return Err(anyhow!(
2433 "Property '{}' not found for '{}'",
2434 old_name,
2435 label_or_type
2436 ));
2437 };
2438 if props.contains_key(new_name) {
2439 props.insert(old_name.to_string(), meta); return Err(anyhow!("Property '{}' already exists", new_name));
2442 }
2443 props.insert(new_name.to_string(), meta);
2444 schema.bump_version();
2445 Ok(())
2446 }
2447
2448 pub fn drop_label(&self, name: &str, if_exists: bool) -> Result<()> {
2449 let mut guard = acquire_write(&self.schema, "schema")?;
2450 let schema = Arc::make_mut(&mut *guard);
2451 if let Some(label_meta) = schema.labels.get_mut(name) {
2452 label_meta.state = SchemaElementState::Tombstone { since: Utc::now() };
2453 schema.bump_version();
2455 Ok(())
2456 } else if if_exists {
2457 Ok(())
2458 } else {
2459 Err(anyhow!("Label '{}' not found", name))
2460 }
2461 }
2462
2463 pub fn drop_edge_type(&self, name: &str, if_exists: bool) -> Result<()> {
2464 let mut guard = acquire_write(&self.schema, "schema")?;
2465 let schema = Arc::make_mut(&mut *guard);
2466 if let Some(edge_meta) = schema.edge_types.get_mut(name) {
2467 edge_meta.state = SchemaElementState::Tombstone { since: Utc::now() };
2468 schema.bump_version();
2470 Ok(())
2471 } else if if_exists {
2472 Ok(())
2473 } else {
2474 Err(anyhow!("Edge Type '{}' not found", name))
2475 }
2476 }
2477}
2478
2479pub fn validate_identifier(name: &str) -> Result<()> {
2481 if name.is_empty() || name.len() > 64 {
2483 return Err(anyhow!("Identifier '{}' must be 1-64 characters", name));
2484 }
2485
2486 let first = name.chars().next().unwrap();
2488 if !first.is_alphabetic() && first != '_' {
2489 return Err(anyhow!(
2490 "Identifier '{}' must start with letter or underscore",
2491 name
2492 ));
2493 }
2494
2495 if !name.chars().all(|c| c.is_alphanumeric() || c == '_') {
2497 return Err(anyhow!(
2498 "Identifier '{}' must contain only alphanumeric and underscore",
2499 name
2500 ));
2501 }
2502
2503 const RESERVED: &[&str] = &[
2505 "MATCH", "CREATE", "DELETE", "SET", "RETURN", "WHERE", "MERGE", "CALL", "YIELD", "WITH",
2506 "UNION", "ORDER", "LIMIT",
2507 ];
2508 if RESERVED.contains(&name.to_uppercase().as_str()) {
2509 return Err(anyhow!("Identifier '{}' cannot be a reserved word", name));
2510 }
2511
2512 Ok(())
2513}
2514
2515pub fn validate_property_name(name: &str) -> Result<()> {
2522 if name.starts_with('_') {
2523 return Err(anyhow!(
2524 "Property name '{}' is reserved: names starting with '_' are reserved by the storage layer",
2525 name
2526 ));
2527 }
2528 validate_reserved_property_name(name)
2529}
2530
2531fn validate_reserved_property_name(name: &str) -> Result<()> {
2538 const RESERVED_PROPS: &[&str] = &[
2547 "ext_id",
2548 "overflow_json",
2549 "eid",
2550 "src_vid",
2551 "dst_vid",
2552 "op",
2553 "__set_struct__",
2561 ];
2562 if RESERVED_PROPS.contains(&name) {
2563 return Err(anyhow!(
2564 "Property name '{}' is reserved by the storage layer; please choose a different name",
2565 name
2566 ));
2567 }
2568 Ok(())
2569}
2570
2571#[cfg(test)]
2572mod tests {
2573 use super::*;
2574 use crate::value::{TemporalValue, Value};
2575 use object_store::local::LocalFileSystem;
2576 use tempfile::tempdir;
2577
2578 #[test]
2579 fn binary_vector_metrics_exact() {
2580 assert_eq!(
2583 DistanceMetric::Hamming.compute_distance_binary(&[0x00], &[0xFF]),
2584 8.0
2585 );
2586 assert_eq!(
2587 DistanceMetric::Hamming.compute_distance_binary(&[0xA5, 0x0F], &[0xA5, 0x00]),
2588 4.0
2589 );
2590 assert_eq!(
2591 DistanceMetric::Hamming.compute_distance_binary(&[0xA5], &[0xA5]),
2592 0.0
2593 );
2594
2595 let j = DistanceMetric::Jaccard.compute_distance_binary(&[0b1100], &[0b1010]);
2598 assert!((j - (2.0 / 3.0)).abs() < 1e-6, "got {j}");
2599 assert_eq!(
2601 DistanceMetric::Jaccard.compute_distance_binary(&[0xFF], &[0xFF]),
2602 0.0
2603 );
2604 assert_eq!(
2606 DistanceMetric::Jaccard.compute_distance_binary(&[0x00, 0x00], &[0x00, 0x00]),
2607 0.0
2608 );
2609 }
2610
2611 #[test]
2612 fn binary_metrics_are_binary_and_route_correctly() {
2613 assert!(DistanceMetric::Hamming.is_binary());
2614 assert!(DistanceMetric::Jaccard.is_binary());
2615 assert!(!DistanceMetric::L2.is_binary());
2616 assert!(!DistanceMetric::L1.is_binary());
2617 }
2618
2619 #[test]
2620 #[should_panic(expected = "binary-vector metric")]
2621 fn float_compute_distance_rejects_binary_metric() {
2622 DistanceMetric::Hamming.compute_distance(&[1.0], &[0.0]);
2623 }
2624
2625 #[test]
2626 fn check_binary_vector_value_guards() {
2627 let ty = DataType::BinaryVector { dimensions: 3 };
2628 assert!(
2629 ty.check_vector_dims(&Value::BinaryVector(vec![1, 2, 3]))
2630 .is_ok()
2631 );
2632 assert!(ty.check_vector_dims(&Value::Null).is_ok());
2633 assert!(
2635 ty.check_vector_dims(&Value::BinaryVector(vec![1, 2]))
2636 .is_err()
2637 );
2638 assert!(
2640 ty.check_vector_dims(&Value::List(vec![
2641 Value::Int(0),
2642 Value::Int(255),
2643 Value::Int(128)
2644 ]))
2645 .is_ok()
2646 );
2647 assert!(
2649 ty.check_vector_dims(&Value::List(vec![
2650 Value::Int(0),
2651 Value::Int(256),
2652 Value::Int(1)
2653 ]))
2654 .is_err()
2655 );
2656 }
2657
2658 #[test]
2659 fn test_datatype_accepts_matrix() {
2660 let dt = || TemporalValue::DateTime {
2661 nanos_since_epoch: 0,
2662 offset_seconds: 0,
2663 timezone_name: None,
2664 };
2665
2666 for ty in [
2668 DataType::String,
2669 DataType::Int64,
2670 DataType::Bool,
2671 DataType::DateTime,
2672 DataType::Float64,
2673 ] {
2674 assert!(ty.accepts(&Value::Null), "{ty:?} must accept Null");
2675 }
2676
2677 assert!(DataType::String.accepts(&Value::String("x".into())));
2679 assert!(DataType::Int64.accepts(&Value::Int(1)));
2680 assert!(DataType::Bool.accepts(&Value::Bool(true)));
2681 assert!(DataType::DateTime.accepts(&Value::Temporal(dt())));
2682
2683 assert!(
2685 DataType::Float64.accepts(&Value::Int(3)),
2686 "Int widens to Float"
2687 );
2688 assert!(DataType::Int32.accepts(&Value::Int(3)), "Int fits Int32");
2689 assert!(DataType::Timestamp.accepts(&Value::Temporal(dt())));
2690 assert!(
2691 DataType::Timestamp.accepts(&Value::String("2026-01-01T00:00:00Z".into())),
2692 "storage parses strings for non-struct Timestamp columns"
2693 );
2694
2695 assert!(
2697 !DataType::DateTime.accepts(&Value::String("2026-01-01T00:00:00Z".into())),
2698 "String into a DateTime struct column nulls silently — reject here"
2699 );
2700 assert!(!DataType::Bool.accepts(&Value::Int(1)));
2701 assert!(!DataType::Int64.accepts(&Value::Bool(true)));
2702 assert!(!DataType::Int64.accepts(&Value::Float(1.5)));
2703 assert!(
2704 !DataType::String.accepts(&Value::Int(10)),
2705 "no implicit stringification"
2706 );
2707 assert!(!DataType::Duration.accepts(&Value::String("P1D".into())));
2708
2709 assert!(DataType::CypherValue.accepts(&Value::Map(Default::default())));
2711 }
2712
2713 #[test]
2714 fn test_check_vector_dims_matrix() {
2715 let vec3 = DataType::Vector { dimensions: 3 };
2716 let multi2 = DataType::List(Box::new(DataType::Vector { dimensions: 2 }));
2717 let flist = |vals: &[f64]| Value::List(vals.iter().map(|f| Value::Float(*f)).collect());
2718
2719 assert!(vec3.check_vector_dims(&Value::Null).is_ok());
2721 assert!(multi2.check_vector_dims(&Value::Null).is_ok());
2722
2723 assert!(
2725 vec3.check_vector_dims(&Value::Vector(vec![1.0, 2.0, 3.0]))
2726 .is_ok()
2727 );
2728 assert!(vec3.check_vector_dims(&flist(&[1.0, 2.0, 3.0])).is_ok());
2729 assert!(
2730 vec3.check_vector_dims(&Value::List(vec![
2731 Value::Int(1),
2732 Value::Float(2.0),
2733 Value::Int(3)
2734 ]))
2735 .is_ok()
2736 );
2737
2738 assert_eq!(
2740 vec3.check_vector_dims(&Value::Vector(vec![1.0, 2.0])),
2741 Err(VectorDimError::WrongLength {
2742 expected: 3,
2743 actual: 2
2744 })
2745 );
2746 assert_eq!(
2747 vec3.check_vector_dims(&flist(&[1.0, 2.0, 3.0, 4.0, 5.0])),
2748 Err(VectorDimError::WrongLength {
2749 expected: 3,
2750 actual: 5
2751 })
2752 );
2753 assert_eq!(
2754 vec3.check_vector_dims(&Value::List(vec![])),
2755 Err(VectorDimError::WrongLength {
2756 expected: 3,
2757 actual: 0
2758 })
2759 );
2760 assert_eq!(
2761 vec3.check_vector_dims(&Value::List(vec![
2762 Value::Float(1.0),
2763 Value::String("x".into()),
2764 Value::Float(3.0),
2765 ])),
2766 Err(VectorDimError::NonNumericElement { index: 1 })
2767 );
2768 assert_eq!(
2769 vec3.check_vector_dims(&Value::List(vec![
2770 Value::Float(1.0),
2771 Value::Null,
2772 Value::Float(3.0)
2773 ])),
2774 Err(VectorDimError::NonNumericElement { index: 1 })
2775 );
2776 assert_eq!(
2777 vec3.check_vector_dims(&Value::String("not a vector".into())),
2778 Err(VectorDimError::NotAVector { actual: "String" })
2779 );
2780
2781 assert!(multi2.check_vector_dims(&Value::List(vec![])).is_ok());
2784 assert!(
2785 multi2
2786 .check_vector_dims(&Value::List(vec![flist(&[1.0, 2.0]), flist(&[3.0, 4.0])]))
2787 .is_ok()
2788 );
2789 assert_eq!(
2790 multi2.check_vector_dims(&Value::List(vec![
2791 flist(&[1.0, 2.0]),
2792 flist(&[9.0, 9.0, 9.0])
2793 ])),
2794 Err(VectorDimError::TokenWrongLength {
2795 token: 1,
2796 expected: 2,
2797 actual: 3
2798 })
2799 );
2800 assert_eq!(
2801 multi2.check_vector_dims(&Value::List(vec![Value::String("tok".into())])),
2802 Err(VectorDimError::TokenNotAVector {
2803 token: 0,
2804 actual: "String"
2805 })
2806 );
2807 assert_eq!(
2808 multi2.check_vector_dims(&Value::Vector(vec![1.0, 2.0])),
2809 Err(VectorDimError::NotATokenList { actual: "Vector" })
2810 );
2811
2812 assert!(
2814 DataType::Int64
2815 .check_vector_dims(&Value::String("x".into()))
2816 .is_ok()
2817 );
2818 assert!(
2819 DataType::List(Box::new(DataType::Float64))
2820 .check_vector_dims(&Value::List(vec![Value::String("x".into())]))
2821 .is_ok()
2822 );
2823 assert!(
2824 DataType::SparseVector { dimensions: 8 }
2825 .check_vector_dims(&Value::Map(Default::default()))
2826 .is_ok()
2827 );
2828
2829 let msg = VectorDimError::WrongLength {
2831 expected: 4,
2832 actual: 5,
2833 }
2834 .to_string();
2835 assert!(msg.contains('4') && msg.contains('5'), "message: {msg}");
2836 }
2837
2838 #[tokio::test]
2839 async fn test_declare_property_idempotent_and_conflicting() -> Result<()> {
2840 let dir = tempdir()?;
2841 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2842 let path = ObjectStorePath::from("schema.json");
2843 let manager = SchemaManager::load_from_store(store.clone(), &path).await?;
2844
2845 manager.add_label("Doc")?;
2846 let vec4 = DataType::Vector { dimensions: 4 };
2847
2848 assert!(manager.declare_property("Doc", "embedding", vec4.clone(), true, None)?);
2850
2851 assert!(!manager.declare_property("Doc", "embedding", vec4.clone(), true, None)?);
2854 assert!(!manager.declare_property(
2855 "Doc",
2856 "embedding",
2857 vec4.clone(),
2858 true,
2859 Some("new docs".into())
2860 )?);
2861
2862 let err = manager
2865 .declare_property(
2866 "Doc",
2867 "embedding",
2868 DataType::Vector { dimensions: 8 },
2869 true,
2870 None,
2871 )
2872 .unwrap_err()
2873 .to_string();
2874 assert!(err.contains('4') && err.contains('8'), "message: {err}");
2875 assert!(!err.contains("already exists"), "message: {err}");
2876
2877 assert!(
2879 manager
2880 .declare_property("Doc", "embedding", vec4.clone(), false, None)
2881 .is_err()
2882 );
2883
2884 let schema = manager.schema();
2886 let meta = &schema.properties["Doc"]["embedding"];
2887 assert_eq!(meta.r#type, vec4);
2888 assert!(meta.nullable);
2889 Ok(())
2890 }
2891
2892 #[tokio::test]
2893 async fn test_schema_management() -> Result<()> {
2894 let dir = tempdir()?;
2895 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2896 let path = ObjectStorePath::from("schema.json");
2897 let manager = SchemaManager::load_from_store(store.clone(), &path).await?;
2898
2899 let lid = manager.add_label("Person")?;
2901 assert_eq!(lid, 1);
2902 assert!(manager.add_label("Person").is_err());
2903
2904 manager.add_property("Person", "name", DataType::String, false)?;
2906 assert!(
2907 manager
2908 .add_property("Person", "name", DataType::String, false)
2909 .is_err()
2910 );
2911
2912 let tid = manager.add_edge_type("knows", vec!["Person".into()], vec!["Person".into()])?;
2914 assert_eq!(tid, 1);
2915
2916 manager.save().await?;
2917 assert!(store.get(&path).await.is_ok());
2919
2920 let manager2 = SchemaManager::load_from_store(store, &path).await?;
2921 assert!(manager2.schema().labels.contains_key("Person"));
2922 assert!(
2923 manager2
2924 .schema()
2925 .properties
2926 .get("Person")
2927 .unwrap()
2928 .contains_key("name")
2929 );
2930
2931 Ok(())
2932 }
2933
2934 #[tokio::test]
2935 async fn test_reserved_property_names_rejected() -> Result<()> {
2936 let dir = tempdir()?;
2937 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2938 let path = ObjectStorePath::from("schema.json");
2939 let manager = SchemaManager::load_from_store(store, &path).await?;
2940
2941 manager.add_label("Tiny")?;
2942
2943 for reserved in &["ext_id", "overflow_json", "eid", "src_vid", "dst_vid", "op"] {
2947 let err = manager
2948 .add_property("Tiny", reserved, DataType::String, true)
2949 .expect_err(&format!("expected '{reserved}' to be rejected"));
2950 assert!(
2951 err.to_string().contains("reserved"),
2952 "error for '{reserved}' should mention 'reserved', got: {err}"
2953 );
2954 }
2955
2956 let err = manager
2961 .add_property("Tiny", "__set_struct__", DataType::String, true)
2962 .expect_err("expected '__set_struct__' to be rejected");
2963 assert!(
2964 err.to_string().contains("reserved"),
2965 "__set_struct__ rejection should mention 'reserved', got: {err}"
2966 );
2967
2968 for reserved in &["_vid", "_uid", "_eid", "_version", "_created_at"] {
2970 assert!(
2971 manager
2972 .add_property("Tiny", reserved, DataType::String, true)
2973 .is_err(),
2974 "expected '{reserved}' to be rejected"
2975 );
2976 }
2977
2978 manager.add_property("Tiny", "ext_id_foo", DataType::String, true)?;
2981 manager.add_property("Tiny", "user_op", DataType::String, true)?;
2982 manager.add_property("Tiny", "type_name", DataType::String, true)?;
2983
2984 manager.add_edge_type("knows", vec!["Tiny".into()], vec!["Tiny".into()])?;
2986 assert!(
2987 manager
2988 .add_property("knows", "src_vid", DataType::Int64, true)
2989 .is_err()
2990 );
2991
2992 assert!(
2994 manager
2995 .add_generated_property(
2996 "Tiny",
2997 "ext_id",
2998 DataType::String,
2999 "concat('x', name)".into()
3000 )
3001 .is_err()
3002 );
3003
3004 Ok(())
3005 }
3006
3007 #[test]
3008 fn test_normalize_function_names() {
3009 assert_eq!(
3010 SchemaManager::normalize_function_names("lower(email)"),
3011 "LOWER(email)"
3012 );
3013 assert_eq!(
3014 SchemaManager::normalize_function_names("LOWER(email)"),
3015 "LOWER(email)"
3016 );
3017 assert_eq!(
3018 SchemaManager::normalize_function_names("Lower(email)"),
3019 "LOWER(email)"
3020 );
3021 assert_eq!(
3022 SchemaManager::normalize_function_names("trim(lower(email))"),
3023 "TRIM(LOWER(email))"
3024 );
3025 }
3026
3027 #[test]
3028 fn test_generated_column_name_case_insensitive() {
3029 let col1 = SchemaManager::generated_column_name("lower(email)");
3030 let col2 = SchemaManager::generated_column_name("LOWER(email)");
3031 let col3 = SchemaManager::generated_column_name("Lower(email)");
3032 assert_eq!(col1, col2);
3033 assert_eq!(col2, col3);
3034 assert!(col1.starts_with("_gen_LOWER_email_"));
3035 }
3036
3037 #[test]
3038 fn test_index_metadata_serde_backward_compat() {
3039 let json = r#"{
3041 "type": "Scalar",
3042 "name": "idx_person_name",
3043 "label": "Person",
3044 "properties": ["name"],
3045 "index_type": "BTree",
3046 "where_clause": null
3047 }"#;
3048 let def: IndexDefinition = serde_json::from_str(json).unwrap();
3049 let meta = def.metadata();
3050 assert_eq!(meta.status, IndexStatus::Online);
3051 assert!(meta.last_built_at.is_none());
3052 assert!(meta.row_count_at_build.is_none());
3053 }
3054
3055 #[test]
3056 fn test_index_metadata_serde_roundtrip() {
3057 let now = Utc::now();
3058 let def = IndexDefinition::Scalar(ScalarIndexConfig {
3059 name: "idx_test".to_string(),
3060 label: "Test".to_string(),
3061 properties: vec!["prop".to_string()],
3062 index_type: ScalarIndexType::BTree,
3063 where_clause: None,
3064 metadata: IndexMetadata {
3065 status: IndexStatus::Building,
3066 last_built_at: Some(now),
3067 row_count_at_build: Some(42),
3068 },
3069 });
3070
3071 let json = serde_json::to_string(&def).unwrap();
3072 let parsed: IndexDefinition = serde_json::from_str(&json).unwrap();
3073 assert_eq!(parsed.metadata().status, IndexStatus::Building);
3074 assert_eq!(parsed.metadata().row_count_at_build, Some(42));
3075 assert!(parsed.metadata().last_built_at.is_some());
3076 }
3077
3078 #[tokio::test]
3079 async fn test_update_index_metadata() -> Result<()> {
3080 let dir = tempdir()?;
3081 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3082 let path = ObjectStorePath::from("schema.json");
3083 let manager = SchemaManager::load_from_store(store, &path).await?;
3084
3085 manager.add_label("Person")?;
3086 let idx = IndexDefinition::Scalar(ScalarIndexConfig {
3087 name: "idx_test".to_string(),
3088 label: "Person".to_string(),
3089 properties: vec!["name".to_string()],
3090 index_type: ScalarIndexType::BTree,
3091 where_clause: None,
3092 metadata: Default::default(),
3093 });
3094 manager.add_index(idx)?;
3095
3096 let initial = manager.get_index("idx_test").unwrap();
3098 assert_eq!(initial.metadata().status, IndexStatus::Online);
3099
3100 manager.update_index_metadata("idx_test", |m| {
3102 m.status = IndexStatus::Building;
3103 m.row_count_at_build = Some(100);
3104 })?;
3105
3106 let updated = manager.get_index("idx_test").unwrap();
3107 assert_eq!(updated.metadata().status, IndexStatus::Building);
3108 assert_eq!(updated.metadata().row_count_at_build, Some(100));
3109
3110 assert!(manager.update_index_metadata("nope", |_| {}).is_err());
3112
3113 Ok(())
3114 }
3115
3116 #[tokio::test]
3121 async fn add_internal_property_reports_newly_added() -> Result<()> {
3122 let dir = tempdir()?;
3123 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3124 let path = ObjectStorePath::from("schema.json");
3125 let manager = SchemaManager::load_from_store(store, &path).await?;
3126 manager.add_label("Doc")?;
3127
3128 let dt = DataType::Vector { dimensions: 16 };
3129 assert!(manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
3131 assert!(!manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
3133 assert!(
3135 manager
3136 .add_internal_property("Doc", "__fde_x", DataType::Vector { dimensions: 8 }, true)
3137 .is_err()
3138 );
3139 Ok(())
3140 }
3141
3142 #[tokio::test]
3147 async fn test_add_index_is_upsert_by_name() -> Result<()> {
3148 let dir = tempdir()?;
3149 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3150 let path = ObjectStorePath::from("schema.json");
3151 let manager = SchemaManager::load_from_store(store, &path).await?;
3152 manager.add_label("Person")?;
3153
3154 let initial = IndexDefinition::Scalar(ScalarIndexConfig {
3155 name: "idx_test".to_string(),
3156 label: "Person".to_string(),
3157 properties: vec!["name".to_string()],
3158 index_type: ScalarIndexType::BTree,
3159 where_clause: None,
3160 metadata: IndexMetadata {
3161 status: IndexStatus::Building,
3162 ..Default::default()
3163 },
3164 });
3165 manager.add_index(initial.clone())?;
3166 assert_eq!(manager.schema().indexes.len(), 1);
3167
3168 manager.add_index(initial.clone())?;
3170 assert_eq!(
3171 manager.schema().indexes.len(),
3172 1,
3173 "duplicate add_index by name must not append"
3174 );
3175
3176 let mut updated_cfg = match initial {
3178 IndexDefinition::Scalar(c) => c,
3179 _ => unreachable!(),
3180 };
3181 updated_cfg.metadata.status = IndexStatus::Online;
3182 updated_cfg.metadata.row_count_at_build = Some(42);
3183 manager.add_index(IndexDefinition::Scalar(updated_cfg))?;
3184 assert_eq!(manager.schema().indexes.len(), 1);
3185 let stored = manager.get_index("idx_test").unwrap();
3186 assert_eq!(stored.metadata().status, IndexStatus::Online);
3187 assert_eq!(stored.metadata().row_count_at_build, Some(42));
3188
3189 let other = IndexDefinition::Scalar(ScalarIndexConfig {
3191 name: "idx_other".to_string(),
3192 label: "Person".to_string(),
3193 properties: vec!["age".to_string()],
3194 index_type: ScalarIndexType::BTree,
3195 where_clause: None,
3196 metadata: IndexMetadata::default(),
3197 });
3198 manager.add_index(other)?;
3199 assert_eq!(manager.schema().indexes.len(), 2);
3200
3201 Ok(())
3202 }
3203
3204 #[tokio::test]
3207 async fn test_load_dedups_bloated_indexes() -> Result<()> {
3208 let dir = tempdir()?;
3209 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3210 let path = ObjectStorePath::from("schema.json");
3211
3212 let mut schema = Schema::default();
3216 schema.labels.insert(
3217 "Person".to_string(),
3218 LabelMeta {
3219 id: 1,
3220 created_at: chrono::Utc::now(),
3221 state: SchemaElementState::Active,
3222 description: None,
3223 },
3224 );
3225 let make = |status: IndexStatus, count: Option<u64>| {
3226 IndexDefinition::Scalar(ScalarIndexConfig {
3227 name: "idx_dup".to_string(),
3228 label: "Person".to_string(),
3229 properties: vec!["name".to_string()],
3230 index_type: ScalarIndexType::BTree,
3231 where_clause: None,
3232 metadata: IndexMetadata {
3233 status,
3234 row_count_at_build: count,
3235 ..Default::default()
3236 },
3237 })
3238 };
3239 for _ in 0..49 {
3240 schema.indexes.push(make(IndexStatus::Building, None));
3241 }
3242 schema.indexes.push(make(IndexStatus::Online, Some(123)));
3243 let json = serde_json::to_string_pretty(&schema)?;
3244 store.put(&path, json.into()).await?;
3245
3246 let manager = SchemaManager::load_from_store(store, &path).await?;
3247 let schema = manager.schema();
3248 assert_eq!(
3249 schema.indexes.len(),
3250 1,
3251 "load() must collapse 50 duplicates by name to 1"
3252 );
3253 assert_eq!(schema.indexes[0].metadata().status, IndexStatus::Online);
3255 assert_eq!(schema.indexes[0].metadata().row_count_at_build, Some(123));
3256
3257 Ok(())
3258 }
3259
3260 #[test]
3261 fn test_vector_index_for_property_skips_non_online() {
3262 let mut schema = Schema::default();
3263 schema.labels.insert(
3264 "Document".to_string(),
3265 LabelMeta {
3266 id: 1,
3267 created_at: chrono::Utc::now(),
3268 state: SchemaElementState::Active,
3269 description: None,
3270 },
3271 );
3272
3273 schema
3275 .indexes
3276 .push(IndexDefinition::Vector(VectorIndexConfig {
3277 name: "vec_doc_embedding".to_string(),
3278 label: "Document".to_string(),
3279 property: "embedding".to_string(),
3280 index_type: VectorIndexType::Flat,
3281 metric: DistanceMetric::Cosine,
3282 embedding_config: None,
3283 metadata: IndexMetadata {
3284 status: IndexStatus::Stale,
3285 ..Default::default()
3286 },
3287 }));
3288
3289 assert!(
3291 schema
3292 .vector_index_for_property("Document", "embedding")
3293 .is_none()
3294 );
3295
3296 if let IndexDefinition::Vector(cfg) = &mut schema.indexes[0] {
3298 cfg.metadata.status = IndexStatus::Online;
3299 }
3300 let result = schema.vector_index_for_property("Document", "embedding");
3301 assert!(result.is_some());
3302 assert_eq!(result.unwrap().metric, DistanceMetric::Cosine);
3303 }
3304
3305 #[tokio::test]
3306 async fn with_overlay_empty_clones_primary_in_isolation() -> Result<()> {
3307 use crate::core::fork::SchemaDelta;
3308
3309 let dir = tempdir()?;
3310 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3311 let path = ObjectStorePath::from("schema.json");
3312 let primary = SchemaManager::load_from_store(store, &path).await?;
3313 primary.add_label("Person")?;
3314
3315 let overlay = primary.with_overlay(&SchemaDelta::empty());
3316 assert_eq!(overlay.schema().labels.len(), 1);
3317
3318 overlay.add_label("Forked")?;
3321 assert!(overlay.schema().labels.contains_key("Forked"));
3322 assert!(!primary.schema().labels.contains_key("Forked"));
3323
3324 Ok(())
3325 }
3326
3327 #[tokio::test]
3328 async fn with_overlay_merges_added_labels_and_edge_types() -> Result<()> {
3329 use crate::core::fork::SchemaDelta;
3330 use chrono::Utc;
3331
3332 let dir = tempdir()?;
3333 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3334 let path = ObjectStorePath::from("schema.json");
3335 let primary = SchemaManager::load_from_store(store, &path).await?;
3336 primary.add_label("Existing")?;
3337
3338 let label_meta = LabelMeta {
3339 id: 99,
3340 created_at: Utc::now(),
3341 state: SchemaElementState::Active,
3342 description: None,
3343 };
3344 let edge_meta = EdgeTypeMeta {
3345 id: 99,
3346 src_labels: vec!["NewLabel".into()],
3347 dst_labels: vec!["NewLabel".into()],
3348 state: SchemaElementState::Active,
3349 description: None,
3350 };
3351 let delta = SchemaDelta {
3352 added_labels: vec![("NewLabel".to_string(), label_meta)],
3353 added_edge_types: vec![("NewEdge".to_string(), edge_meta)],
3354 added_properties: vec![],
3355 };
3356
3357 let overlay = primary.with_overlay(&delta);
3358 let merged = overlay.schema();
3359 assert!(merged.labels.contains_key("Existing"));
3360 assert!(merged.labels.contains_key("NewLabel"));
3361 assert!(merged.edge_types.contains_key("NewEdge"));
3362
3363 assert!(!primary.schema().labels.contains_key("NewLabel"));
3365 Ok(())
3366 }
3367
3368 #[tokio::test]
3373 async fn test_get_or_assign_edge_type_id_concurrent() -> Result<()> {
3374 let dir = tempdir()?;
3375 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3376 let path = ObjectStorePath::from("schema.json");
3377 let manager = Arc::new(SchemaManager::load_from_store(store, &path).await?);
3378
3379 let mut handles = Vec::new();
3380 for _ in 0..16 {
3381 let m = manager.clone();
3382 handles.push(std::thread::spawn(move || {
3383 m.get_or_assign_edge_type_id("RACED")
3384 }));
3385 }
3386 let ids: Vec<u32> = handles.into_iter().map(|h| h.join().unwrap()).collect();
3387 assert!(
3388 ids.iter().all(|&id| id == ids[0]),
3389 "all racers must observe one id, got {ids:?}"
3390 );
3391 assert_eq!(manager.get_or_assign_edge_type_id("RACED"), ids[0]);
3393
3394 manager.add_label("A")?;
3396 let declared = manager.add_edge_type("DECLARED", vec!["A".into()], vec!["A".into()])?;
3397 assert_eq!(manager.get_or_assign_edge_type_id("DECLARED"), declared);
3398 Ok(())
3399 }
3400
3401 #[test]
3406 fn test_new_schemaless_edge_type_bumps_schema_version() {
3407 let mut schema = Schema::default();
3408 let v0 = schema.schema_version;
3409
3410 let id1 = schema.get_or_assign_edge_type_id("FRESH");
3411 assert_eq!(
3412 schema.schema_version,
3413 v0.wrapping_add(1),
3414 "minting a new edge type must bump schema_version"
3415 );
3416
3417 let id1_again = schema.get_or_assign_edge_type_id("FRESH");
3419 assert_eq!(id1, id1_again);
3420 assert_eq!(
3421 schema.schema_version,
3422 v0.wrapping_add(1),
3423 "resolving an existing edge type must not bump schema_version"
3424 );
3425
3426 let _id2 = schema.get_or_assign_edge_type_id("OTHER");
3428 assert_eq!(
3429 schema.schema_version,
3430 v0.wrapping_add(2),
3431 "a second new edge type must bump schema_version again"
3432 );
3433 }
3434
3435 #[test]
3439 fn validate_schema_element_name_rejects_unsafe() {
3440 for bad in ["", " ", "a/b", "a b", "a\nb", "a\\b", "x\0y"] {
3441 assert!(
3442 SchemaManager::validate_schema_element_name("Label", bad).is_err(),
3443 "expected {bad:?} to be rejected"
3444 );
3445 }
3446 for good in ["Person", "My.Label", "edge_2", "KNOWS"] {
3447 assert!(
3448 SchemaManager::validate_schema_element_name("Label", good).is_ok(),
3449 "expected {good:?} to be accepted"
3450 );
3451 }
3452 let long = "x".repeat(MAX_SCHEMA_NAME_LEN + 1);
3454 assert!(SchemaManager::validate_schema_element_name("Label", &long).is_err());
3455 }
3456}