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
20mod index_types;
21pub use index_types::*;
22
23#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
24#[non_exhaustive]
25pub enum SchemaElementState {
26 Active,
27 Hidden {
28 since: DateTime<Utc>,
29 last_active_snapshot: String, },
31 Tombstone {
32 since: DateTime<Utc>,
33 },
34}
35
36use arrow_schema::{DataType as ArrowDataType, Field, Fields, TimeUnit};
37
38pub fn datetime_struct_fields() -> Fields {
45 Fields::from(vec![
46 Field::new(
47 "nanos_since_epoch",
48 ArrowDataType::Timestamp(TimeUnit::Nanosecond, None),
49 true,
50 ),
51 Field::new("offset_seconds", ArrowDataType::Int32, true),
52 Field::new("timezone_name", ArrowDataType::Utf8, true),
53 ])
54}
55
56pub fn time_struct_fields() -> Fields {
62 Fields::from(vec![
63 Field::new(
64 "nanos_since_midnight",
65 ArrowDataType::Time64(TimeUnit::Nanosecond),
66 true,
67 ),
68 Field::new("offset_seconds", ArrowDataType::Int32, true),
69 ])
70}
71
72pub fn is_datetime_struct(arrow_dt: &ArrowDataType) -> bool {
74 matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == datetime_struct_fields())
75}
76
77pub fn is_time_struct(arrow_dt: &ArrowDataType) -> bool {
79 matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == time_struct_fields())
80}
81
82pub fn sparse_vector_struct_fields() -> Fields {
89 Fields::from(vec![
90 Field::new(
91 "indices",
92 ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::UInt32, true))),
93 false,
94 ),
95 Field::new(
96 "values",
97 ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::Float32, true))),
98 false,
99 ),
100 ])
101}
102
103pub fn is_sparse_vector_struct(arrow_dt: &ArrowDataType) -> bool {
105 matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == sparse_vector_struct_fields())
106}
107
108pub fn raw_bytes_field_metadata() -> HashMap<String, String> {
116 HashMap::from([("uni_raw_bytes".to_string(), "true".to_string())])
117}
118
119#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
120#[non_exhaustive]
121pub enum CrdtType {
122 GCounter,
123 GSet,
124 ORSet,
125 LWWRegister,
126 LWWMap,
127 Rga,
128 VectorClock,
129 VCRegister,
130}
131
132impl CrdtType {
133 #[must_use]
145 pub fn type_name(&self) -> &'static str {
146 match self {
147 CrdtType::GCounter => "GCounter",
148 CrdtType::GSet => "GSet",
149 CrdtType::ORSet => "ORSet",
150 CrdtType::LWWRegister => "LWWRegister",
151 CrdtType::LWWMap => "LWWMap",
152 CrdtType::Rga => "Rga",
153 CrdtType::VectorClock => "VectorClock",
154 CrdtType::VCRegister => "VCRegister",
155 }
156 }
157}
158
159#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq)]
160pub enum PointType {
161 Geographic, Cartesian2D, Cartesian3D, }
165
166#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
167#[non_exhaustive]
168pub enum DataType {
169 String,
170 Int32,
171 Int64,
172 Float32,
173 Float64,
174 Bool,
175 Timestamp,
176 Date,
177 Time,
178 DateTime,
179 Duration,
180 CypherValue,
181 Bytes,
182 Point(PointType),
183 Vector {
184 dimensions: usize,
185 },
186 SparseVector {
189 dimensions: usize,
190 },
191 BinaryVector {
196 dimensions: usize,
197 },
198 Btic,
199 Crdt(CrdtType),
200 List(Box<DataType>),
201 Map(Box<DataType>, Box<DataType>),
202}
203
204impl DataType {
205 #[allow(non_upper_case_globals)]
207 pub const Float: DataType = DataType::Float64;
208 #[allow(non_upper_case_globals)]
209 pub const Int: DataType = DataType::Int64;
210
211 pub fn to_arrow(&self) -> ArrowDataType {
212 match self {
213 DataType::String => ArrowDataType::Utf8,
214 DataType::Int32 => ArrowDataType::Int32,
215 DataType::Int64 => ArrowDataType::Int64,
216 DataType::Float32 => ArrowDataType::Float32,
217 DataType::Float64 => ArrowDataType::Float64,
218 DataType::Bool => ArrowDataType::Boolean,
219 DataType::Timestamp => {
220 ArrowDataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
221 }
222 DataType::Date => ArrowDataType::Date32,
223 DataType::Time => ArrowDataType::Struct(time_struct_fields()),
224 DataType::DateTime => ArrowDataType::Struct(datetime_struct_fields()),
225 DataType::Duration => ArrowDataType::LargeBinary, DataType::CypherValue => ArrowDataType::LargeBinary, DataType::Bytes => ArrowDataType::LargeBinary, DataType::Point(pt) => match pt {
229 PointType::Geographic => ArrowDataType::Struct(Fields::from(vec![
230 Field::new("latitude", ArrowDataType::Float64, false),
231 Field::new("longitude", ArrowDataType::Float64, false),
232 Field::new("crs", ArrowDataType::Utf8, false),
233 ])),
234 PointType::Cartesian2D => ArrowDataType::Struct(Fields::from(vec![
235 Field::new("x", ArrowDataType::Float64, false),
236 Field::new("y", ArrowDataType::Float64, false),
237 Field::new("crs", ArrowDataType::Utf8, false),
238 ])),
239 PointType::Cartesian3D => ArrowDataType::Struct(Fields::from(vec![
240 Field::new("x", ArrowDataType::Float64, false),
241 Field::new("y", ArrowDataType::Float64, false),
242 Field::new("z", ArrowDataType::Float64, false),
243 Field::new("crs", ArrowDataType::Utf8, false),
244 ])),
245 },
246 DataType::Vector { dimensions } => ArrowDataType::FixedSizeList(
247 Arc::new(Field::new("item", ArrowDataType::Float32, true)),
248 *dimensions as i32,
249 ),
250 DataType::SparseVector { .. } => ArrowDataType::Struct(sparse_vector_struct_fields()),
251 DataType::BinaryVector { dimensions } => ArrowDataType::FixedSizeList(
252 Arc::new(Field::new("item", ArrowDataType::UInt8, true)),
253 *dimensions as i32,
254 ),
255 DataType::Btic => ArrowDataType::FixedSizeBinary(24),
256 DataType::Crdt(_) => ArrowDataType::Binary, DataType::List(inner) => {
258 let item = Field::new("item", inner.to_arrow(), true);
262 let item = if matches!(**inner, DataType::Bytes) {
263 item.with_metadata(raw_bytes_field_metadata())
264 } else {
265 item
266 };
267 ArrowDataType::List(Arc::new(item))
268 }
269 DataType::Map(key, value) => {
270 let value_field = if value.map_value_is_typed() {
277 let f = Field::new("value", value.to_arrow(), true);
278 if matches!(**value, DataType::Bytes) {
279 f.with_metadata(raw_bytes_field_metadata())
280 } else {
281 f
282 }
283 } else {
284 Field::new("value", ArrowDataType::LargeBinary, true)
285 };
286 ArrowDataType::List(Arc::new(Field::new(
287 "item",
288 ArrowDataType::Struct(Fields::from(vec![
289 Field::new("key", key.to_arrow(), false),
290 value_field,
291 ])),
292 true,
293 )))
294 }
295 }
296 }
297
298 pub fn map_value_is_typed(&self) -> bool {
303 matches!(
304 self,
305 DataType::String
306 | DataType::Int64
307 | DataType::Int32
308 | DataType::Float64
309 | DataType::Float32
310 | DataType::Bool
311 | DataType::Bytes
312 )
313 }
314
315 pub fn accepts(&self, value: &crate::value::Value) -> bool {
341 use crate::value::{TemporalValue, Value};
342
343 if matches!(value, Value::Null) {
345 return true;
346 }
347
348 match self {
349 DataType::CypherValue | DataType::Crdt(_) | DataType::Point(_) => true,
351
352 DataType::String => matches!(value, Value::String(_)),
353 DataType::Int32 | DataType::Int64 => matches!(value, Value::Int(_)),
354 DataType::Float32 | DataType::Float64 => {
356 matches!(value, Value::Int(_) | Value::Float(_))
357 }
358 DataType::Bool => matches!(value, Value::Bool(_)),
359
360 DataType::Timestamp => matches!(
363 value,
364 Value::String(_)
365 | Value::Int(_)
366 | Value::Temporal(
367 TemporalValue::DateTime { .. } | TemporalValue::LocalDateTime { .. }
368 )
369 ),
370 DataType::DateTime => matches!(
371 value,
372 Value::Temporal(
373 TemporalValue::DateTime { .. } | TemporalValue::LocalDateTime { .. }
374 )
375 ),
376 DataType::Date => {
377 matches!(
378 value,
379 Value::Int(_) | Value::Temporal(TemporalValue::Date { .. })
380 )
381 }
382 DataType::Time => matches!(
383 value,
384 Value::Int(_)
385 | Value::Temporal(TemporalValue::Time { .. } | TemporalValue::LocalTime { .. })
386 ),
387 DataType::Duration => {
388 matches!(value, Value::Temporal(TemporalValue::Duration { .. }))
389 }
390 DataType::Bytes => matches!(value, Value::Bytes(_)),
391 DataType::Btic => matches!(
393 value,
394 Value::String(_) | Value::List(_) | Value::Temporal(TemporalValue::Btic { .. })
395 ),
396 DataType::Vector { .. } => matches!(value, Value::Vector(_) | Value::List(_)),
399 DataType::SparseVector { .. } => {
404 matches!(value, Value::SparseVector { .. } | Value::Map(_))
405 }
406 DataType::BinaryVector { .. } => {
410 matches!(value, Value::BinaryVector(_) | Value::List(_))
411 }
412 DataType::List(_) => matches!(value, Value::List(_)),
415 DataType::Map(_, _) => matches!(value, Value::Map(_)),
416 }
417 }
418
419 pub fn check_vector_dims(&self, value: &crate::value::Value) -> Result<(), VectorDimError> {
437 use crate::value::Value;
438
439 if matches!(value, Value::Null) {
440 return Ok(());
441 }
442
443 match self {
444 DataType::Vector { dimensions } => check_dense_vector_value(value, *dimensions),
445 DataType::BinaryVector { dimensions } => check_binary_vector_value(value, *dimensions),
446 DataType::List(inner) => {
447 let DataType::Vector { dimensions } = inner.as_ref() else {
448 return Ok(());
449 };
450 let Value::List(tokens) = value else {
451 return Err(VectorDimError::NotATokenList {
452 actual: value_variant_name(value),
453 });
454 };
455 for (token, token_value) in tokens.iter().enumerate() {
456 check_dense_vector_value(token_value, *dimensions)
457 .map_err(|e| e.for_token(token))?;
458 }
459 Ok(())
460 }
461 _ => Ok(()),
462 }
463 }
464}
465
466#[derive(Debug, Clone, PartialEq, Eq)]
472pub enum VectorDimError {
473 WrongLength {
475 expected: usize,
477 actual: usize,
479 },
480 NonNumericElement {
482 index: usize,
484 },
485 NotAVector {
487 actual: &'static str,
489 },
490 TokenWrongLength {
492 token: usize,
494 expected: usize,
496 actual: usize,
498 },
499 TokenNonNumericElement {
501 token: usize,
503 index: usize,
505 },
506 TokenNotAVector {
508 token: usize,
510 actual: &'static str,
512 },
513 NotATokenList {
515 actual: &'static str,
517 },
518}
519
520impl VectorDimError {
521 fn for_token(self, token: usize) -> Self {
523 match self {
524 Self::WrongLength { expected, actual } => Self::TokenWrongLength {
525 token,
526 expected,
527 actual,
528 },
529 Self::NonNumericElement { index } => Self::TokenNonNumericElement { token, index },
530 Self::NotAVector { actual } => Self::TokenNotAVector { token, actual },
531 other => other,
532 }
533 }
534}
535
536impl std::fmt::Display for VectorDimError {
537 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
538 match self {
539 Self::WrongLength { expected, actual } => write!(
540 f,
541 "got a vector of length {actual}, expected {expected} dimensions"
542 ),
543 Self::NonNumericElement { index } => {
544 write!(f, "element {index} is not numeric")
545 }
546 Self::NotAVector { actual } => {
547 write!(f, "got a non-vector value of type {actual}")
548 }
549 Self::TokenWrongLength {
550 token,
551 expected,
552 actual,
553 } => write!(
554 f,
555 "token {token} has {actual} dimensions, expected {expected}"
556 ),
557 Self::TokenNonNumericElement { token, index } => {
558 write!(f, "token {token} element {index} is not numeric")
559 }
560 Self::TokenNotAVector { token, actual } => {
561 write!(f, "token {token} is not a vector (got {actual})")
562 }
563 Self::NotATokenList { actual } => write!(
564 f,
565 "got a non-list value of type {actual} for a multi-vector column"
566 ),
567 }
568 }
569}
570
571impl std::error::Error for VectorDimError {}
572
573pub fn check_dense_vector_value(
587 value: &crate::value::Value,
588 dimensions: usize,
589) -> Result<(), VectorDimError> {
590 use crate::value::Value;
591
592 match value {
593 Value::Null => Ok(()),
594 Value::Vector(v) => {
595 if v.len() == dimensions {
596 Ok(())
597 } else {
598 Err(VectorDimError::WrongLength {
599 expected: dimensions,
600 actual: v.len(),
601 })
602 }
603 }
604 Value::List(items) => {
605 if items.len() != dimensions {
606 return Err(VectorDimError::WrongLength {
607 expected: dimensions,
608 actual: items.len(),
609 });
610 }
611 if let Some(index) = items.iter().position(|e| !e.is_number()) {
612 return Err(VectorDimError::NonNumericElement { index });
613 }
614 Ok(())
615 }
616 other => Err(VectorDimError::NotAVector {
617 actual: value_variant_name(other),
618 }),
619 }
620}
621
622pub fn check_binary_vector_value(
636 value: &crate::value::Value,
637 dimensions: usize,
638) -> Result<(), VectorDimError> {
639 use crate::value::Value;
640
641 match value {
642 Value::Null => Ok(()),
643 Value::BinaryVector(bytes) => {
644 if bytes.len() == dimensions {
645 Ok(())
646 } else {
647 Err(VectorDimError::WrongLength {
648 expected: dimensions,
649 actual: bytes.len(),
650 })
651 }
652 }
653 Value::List(items) => {
654 if items.len() != dimensions {
655 return Err(VectorDimError::WrongLength {
656 expected: dimensions,
657 actual: items.len(),
658 });
659 }
660 if let Some(index) = items
661 .iter()
662 .position(|e| !matches!(e.as_i64(), Some(0..=255)))
663 {
664 return Err(VectorDimError::NonNumericElement { index });
665 }
666 Ok(())
667 }
668 other => Err(VectorDimError::NotAVector {
669 actual: value_variant_name(other),
670 }),
671 }
672}
673
674fn value_variant_name(value: &crate::value::Value) -> &'static str {
676 use crate::value::Value;
677
678 match value {
679 Value::Null => "Null",
680 Value::Bool(_) => "Bool",
681 Value::Int(_) => "Int",
682 Value::Float(_) => "Float",
683 Value::String(_) => "String",
684 Value::Bytes(_) => "Bytes",
685 Value::List(_) => "List",
686 Value::Map(_) => "Map",
687 Value::Node(_) => "Node",
688 Value::Edge(_) => "Edge",
689 Value::Path(_) => "Path",
690 Value::Vector(_) => "Vector",
691 Value::SparseVector { .. } => "SparseVector",
692 Value::BinaryVector(_) => "BinaryVector",
693 Value::Temporal(_) => "Temporal",
694 }
695}
696
697fn default_created_at() -> DateTime<Utc> {
698 Utc::now()
699}
700
701fn default_state() -> SchemaElementState {
702 SchemaElementState::Active
703}
704
705fn default_version_1() -> u32 {
706 1
707}
708
709#[derive(Clone, Debug, Serialize, Deserialize)]
710pub struct PropertyMeta {
711 pub r#type: DataType,
712 pub nullable: bool,
713 #[serde(default = "default_version_1")]
714 pub added_in: u32, #[serde(default = "default_state")]
716 pub state: SchemaElementState,
717 #[serde(default)]
718 pub generation_expression: Option<String>,
719 #[serde(default, skip_serializing_if = "Option::is_none")]
720 pub description: Option<String>,
721}
722
723#[derive(Clone, Debug, Serialize, Deserialize)]
724pub struct LabelMeta {
725 pub id: u16, #[serde(default = "default_created_at")]
727 pub created_at: DateTime<Utc>,
728 #[serde(default = "default_state")]
729 pub state: SchemaElementState,
730 #[serde(default, skip_serializing_if = "Option::is_none")]
731 pub description: Option<String>,
732}
733
734#[derive(Clone, Debug, Serialize, Deserialize)]
735pub struct EdgeTypeMeta {
736 pub id: u32,
738 pub src_labels: Vec<String>,
739 pub dst_labels: Vec<String>,
740 #[serde(default = "default_state")]
741 pub state: SchemaElementState,
742 #[serde(default, skip_serializing_if = "Option::is_none")]
743 pub description: Option<String>,
744}
745
746#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
747#[non_exhaustive]
748pub enum ConstraintType {
749 Unique {
750 properties: Vec<String>,
751 },
752 Exists {
753 property: String,
754 },
755 Check {
756 expression: String,
757 },
758 NodeKey {
762 properties: Vec<String>,
763 },
764}
765
766impl ConstraintType {
767 #[must_use]
773 pub fn unique_properties(&self) -> Option<&[String]> {
774 match self {
775 ConstraintType::Unique { properties } | ConstraintType::NodeKey { properties } => {
776 Some(properties)
777 }
778 _ => None,
779 }
780 }
781}
782
783#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
784#[non_exhaustive]
785pub enum ConstraintTarget {
786 Label(String),
787 EdgeType(String),
788}
789
790#[derive(Clone, Debug, Serialize, Deserialize)]
791pub struct Constraint {
792 pub name: String,
793 pub constraint_type: ConstraintType,
794 pub target: ConstraintTarget,
795 pub enabled: bool,
796}
797
798#[derive(Clone, Debug, Serialize, Deserialize)]
804pub struct SchemalessEdgeTypeRegistry {
805 name_to_id: HashMap<String, u32>,
806 id_to_name: HashMap<u32, String>,
807 next_local_id: u32,
809}
810
811impl SchemalessEdgeTypeRegistry {
812 pub fn new() -> Self {
813 Self {
814 name_to_id: HashMap::new(),
815 id_to_name: HashMap::new(),
816 next_local_id: 1,
817 }
818 }
819
820 pub fn get_or_assign_id(&mut self, type_name: &str) -> u32 {
822 if let Some(&id) = self.name_to_id.get(type_name) {
823 return id;
824 }
825
826 let id = make_schemaless_id(self.next_local_id);
827 self.next_local_id += 1;
828
829 self.name_to_id.insert(type_name.to_string(), id);
830 self.id_to_name.insert(id, type_name.to_string());
831
832 id
833 }
834
835 pub fn type_name_by_id(&self, type_id: u32) -> Option<&str> {
837 self.id_to_name.get(&type_id).map(String::as_str)
838 }
839
840 pub fn contains(&self, type_name: &str) -> bool {
842 self.name_to_id.contains_key(type_name)
843 }
844
845 pub fn id_by_name(&self, type_name: &str) -> Option<u32> {
847 self.name_to_id.get(type_name).copied()
848 }
849
850 pub fn id_by_name_case_insensitive(&self, type_name: &str) -> Option<u32> {
852 self.name_to_id
853 .iter()
854 .find(|(k, _)| k.eq_ignore_ascii_case(type_name))
855 .map(|(_, &id)| id)
856 }
857
858 pub fn all_type_ids(&self) -> Vec<u32> {
860 self.id_to_name.keys().copied().collect()
861 }
862
863 pub fn is_empty(&self) -> bool {
865 self.name_to_id.is_empty()
866 }
867}
868
869impl Default for SchemalessEdgeTypeRegistry {
870 fn default() -> Self {
871 Self::new()
872 }
873}
874
875pub const VIRTUAL_LABEL_ID_START: u16 = 0xFF00;
881pub const VIRTUAL_LABEL_ID_SENTINEL: u16 = 0xFFFF;
883
884const MAX_SCHEMA_NAME_LEN: usize = 255;
889
890#[inline]
892pub fn is_virtual_label_id(id: u16) -> bool {
893 (VIRTUAL_LABEL_ID_START..VIRTUAL_LABEL_ID_SENTINEL).contains(&id)
894}
895
896#[derive(Clone, Debug, Serialize, Deserialize)]
897pub struct Schema {
898 pub schema_version: u32,
899 pub labels: HashMap<String, LabelMeta>,
900 pub edge_types: HashMap<String, EdgeTypeMeta>,
901 pub properties: HashMap<String, HashMap<String, PropertyMeta>>,
902 #[serde(default)]
903 pub indexes: Vec<IndexDefinition>,
904 #[serde(default)]
905 pub constraints: Vec<Constraint>,
906 #[serde(default)]
908 pub schemaless_registry: SchemalessEdgeTypeRegistry,
909}
910
911impl Default for Schema {
912 fn default() -> Self {
913 Self {
914 schema_version: 1,
915 labels: HashMap::new(),
916 edge_types: HashMap::new(),
917 properties: HashMap::new(),
918 indexes: Vec::new(),
919 constraints: Vec::new(),
920 schemaless_registry: SchemalessEdgeTypeRegistry::new(),
921 }
922 }
923}
924
925impl Schema {
926 fn bump_version(&mut self) {
935 self.schema_version = self.schema_version.wrapping_add(1);
936 }
937
938 pub fn label_name_by_id(&self, label_id: u16) -> Option<&str> {
943 self.labels
944 .iter()
945 .find(|(_, meta)| meta.id == label_id)
946 .map(|(name, _)| name.as_str())
947 }
948
949 pub fn label_id_by_name(&self, label_name: &str) -> Option<u16> {
951 self.labels.get(label_name).map(|meta| meta.id)
952 }
953
954 pub fn edge_type_name_by_id(&self, type_id: u32) -> Option<&str> {
959 self.edge_types
960 .iter()
961 .find(|(_, meta)| meta.id == type_id)
962 .map(|(name, _)| name.as_str())
963 }
964
965 pub fn edge_type_id_by_name(&self, type_name: &str) -> Option<u32> {
967 self.edge_types.get(type_name).map(|meta| meta.id)
968 }
969
970 pub fn vector_index_for_property(
975 &self,
976 label: &str,
977 property: &str,
978 ) -> Option<&VectorIndexConfig> {
979 self.indexes.iter().find_map(|idx| {
980 if let IndexDefinition::Vector(config) = idx
981 && config.label == label
982 && config.property == property
983 && config.metadata.status == IndexStatus::Online
984 {
985 return Some(config);
986 }
987 None
988 })
989 }
990
991 pub fn sparse_index_for_property(
993 &self,
994 label: &str,
995 property: &str,
996 ) -> Option<&SparseVectorIndexConfig> {
997 self.indexes.iter().find_map(|idx| {
998 if let IndexDefinition::Sparse(config) = idx
999 && config.label == label
1000 && config.property == property
1001 && config.metadata.status == IndexStatus::Online
1002 {
1003 return Some(config);
1004 }
1005 None
1006 })
1007 }
1008
1009 pub fn fulltext_index_for_property(
1014 &self,
1015 label: &str,
1016 property: &str,
1017 ) -> Option<&FullTextIndexConfig> {
1018 self.indexes.iter().find_map(|idx| {
1019 if let IndexDefinition::FullText(config) = idx
1020 && config.label == label
1021 && config.properties.iter().any(|p| p == property)
1022 && config.metadata.status == IndexStatus::Online
1023 {
1024 return Some(config);
1025 }
1026 None
1027 })
1028 }
1029
1030 pub fn get_label_case_insensitive(&self, name: &str) -> Option<&LabelMeta> {
1035 self.labels
1036 .iter()
1037 .find(|(k, _)| k.eq_ignore_ascii_case(name))
1038 .map(|(_, v)| v)
1039 }
1040
1041 pub fn canonical_label_name(&self, name: &str) -> Option<String> {
1048 self.labels
1049 .iter()
1050 .find(|(k, _)| k.eq_ignore_ascii_case(name))
1051 .map(|(k, _)| k.clone())
1052 }
1053
1054 pub fn label_id_by_name_case_insensitive(&self, label_name: &str) -> Option<u16> {
1056 self.get_label_case_insensitive(label_name)
1057 .map(|meta| meta.id)
1058 }
1059
1060 pub fn get_edge_type_case_insensitive(&self, name: &str) -> Option<&EdgeTypeMeta> {
1065 self.edge_types
1066 .iter()
1067 .find(|(k, _)| k.eq_ignore_ascii_case(name))
1068 .map(|(_, v)| v)
1069 }
1070
1071 pub fn edge_type_id_by_name_case_insensitive(&self, type_name: &str) -> Option<u32> {
1073 self.get_edge_type_case_insensitive(type_name)
1074 .map(|meta| meta.id)
1075 }
1076
1077 pub fn edge_type_id_unified_case_insensitive(&self, type_name: &str) -> Option<u32> {
1080 self.edge_type_id_by_name_case_insensitive(type_name)
1081 .or_else(|| {
1082 self.schemaless_registry
1083 .id_by_name_case_insensitive(type_name)
1084 })
1085 }
1086
1087 pub fn get_or_assign_edge_type_id(&mut self, type_name: &str) -> u32 {
1093 if let Some(id) = self.edge_type_id_unified(type_name) {
1094 return id;
1095 }
1096 let id = self.schemaless_registry.get_or_assign_id(type_name);
1104 self.bump_version();
1105 id
1106 }
1107
1108 pub fn edge_type_id_unified(&self, type_name: &str) -> Option<u32> {
1115 self.edge_type_id_by_name(type_name)
1116 .or_else(|| self.schemaless_registry.id_by_name(type_name))
1117 }
1118
1119 pub fn edge_type_name_by_id_unified(&self, type_id: u32) -> Option<String> {
1123 if is_schemaless_edge_type(type_id) {
1124 self.schemaless_registry
1125 .type_name_by_id(type_id)
1126 .map(str::to_owned)
1127 } else {
1128 self.edge_type_name_by_id(type_id).map(str::to_owned)
1129 }
1130 }
1131
1132 pub fn all_edge_type_ids(&self) -> Vec<u32> {
1135 let mut ids: Vec<u32> = self.edge_types.values().map(|m| m.id).collect();
1136 ids.extend(self.schemaless_registry.all_type_ids());
1137 ids.sort_unstable();
1138 ids
1139 }
1140}
1141
1142pub struct SchemaManager {
1143 store: Arc<dyn ObjectStore>,
1144 path: ObjectStorePath,
1145 schema: RwLock<Arc<Schema>>,
1146}
1147
1148impl SchemaManager {
1149 pub async fn load(path: impl AsRef<Path>) -> Result<Self> {
1150 let path = path.as_ref();
1151 let parent = path
1152 .parent()
1153 .ok_or_else(|| anyhow!("Invalid schema path"))?;
1154 let filename = path
1155 .file_name()
1156 .ok_or_else(|| anyhow!("Invalid schema filename"))?
1157 .to_str()
1158 .ok_or_else(|| anyhow!("Invalid utf8 filename"))?;
1159
1160 let store = Arc::new(LocalFileSystem::new_with_prefix(parent)?);
1161 let obj_path = ObjectStorePath::from(filename);
1162
1163 Self::load_from_store(store, &obj_path).await
1164 }
1165
1166 pub async fn load_from_store(
1167 store: Arc<dyn ObjectStore>,
1168 path: &ObjectStorePath,
1169 ) -> Result<Self> {
1170 match store.get(path).await {
1171 Ok(result) => {
1172 let bytes = result.bytes().await?;
1173 let content = String::from_utf8(bytes.to_vec())?;
1174 let mut schema: Schema = serde_json::from_str(&content)?;
1175 let original_len = schema.indexes.len();
1183 if original_len > 0 {
1184 let mut seen: std::collections::HashSet<String> =
1185 std::collections::HashSet::with_capacity(original_len);
1186 let mut dedup: Vec<IndexDefinition> = schema
1187 .indexes
1188 .iter()
1189 .rev()
1190 .filter(|idx| seen.insert(idx.name().to_string()))
1191 .cloned()
1192 .collect();
1193 dedup.reverse();
1194 if dedup.len() != original_len {
1195 tracing::warn!(
1196 collapsed = original_len - dedup.len(),
1197 kept = dedup.len(),
1198 "schema.indexes: collapsed duplicate entries on load (issue #63)"
1199 );
1200 schema.indexes = dedup;
1201 }
1202 }
1203 Ok(Self {
1204 store,
1205 path: path.clone(),
1206 schema: RwLock::new(Arc::new(schema)),
1207 })
1208 }
1209 Err(object_store::Error::NotFound { .. }) => Ok(Self {
1210 store,
1211 path: path.clone(),
1212 schema: RwLock::new(Arc::new(Schema::default())),
1213 }),
1214 Err(e) => Err(anyhow::Error::from(e)),
1215 }
1216 }
1217
1218 pub async fn save(&self) -> Result<()> {
1219 let content = {
1220 let schema_guard = acquire_read(&self.schema, "schema")?;
1221 serde_json::to_string_pretty(&**schema_guard)?
1222 };
1223 self.store
1224 .put(&self.path, content.into())
1225 .await
1226 .map_err(anyhow::Error::from)?;
1227 Ok(())
1228 }
1229
1230 pub fn path(&self) -> &ObjectStorePath {
1231 &self.path
1232 }
1233
1234 pub fn schema(&self) -> Arc<Schema> {
1235 self.schema
1236 .read()
1237 .expect("Schema lock poisoned - a thread panicked while holding it")
1238 .clone()
1239 }
1240
1241 fn normalize_function_names(expr: &str) -> String {
1244 let mut result = String::with_capacity(expr.len());
1245 let mut chars = expr.chars().peekable();
1246
1247 while let Some(ch) = chars.next() {
1248 if ch.is_alphabetic() {
1249 let mut ident = String::new();
1251 ident.push(ch);
1252
1253 while let Some(&next) = chars.peek() {
1254 if next.is_alphanumeric() || next == '_' {
1255 ident.push(chars.next().unwrap());
1256 } else {
1257 break;
1258 }
1259 }
1260
1261 if chars.peek() == Some(&'(') {
1263 result.push_str(&ident.to_uppercase());
1264 } else {
1265 result.push_str(&ident); }
1267 } else {
1268 result.push(ch);
1269 }
1270 }
1271
1272 result
1273 }
1274
1275 pub fn generated_column_name(expr: &str) -> String {
1283 let normalized = Self::normalize_function_names(expr);
1285
1286 let sanitized = normalized
1287 .replace(|c: char| !c.is_alphanumeric(), "_")
1288 .trim_matches('_')
1289 .to_string();
1290
1291 const FNV_OFFSET_BASIS: u64 = 14695981039346656037;
1293 const FNV_PRIME: u64 = 1099511628211;
1294
1295 let mut hash = FNV_OFFSET_BASIS;
1296 for byte in normalized.as_bytes() {
1297 hash ^= *byte as u64;
1298 hash = hash.wrapping_mul(FNV_PRIME);
1299 }
1300
1301 format!("_gen_{}_{:x}", sanitized, hash)
1302 }
1303
1304 pub fn replace_schema(&self, new_schema: Schema) {
1305 let mut schema = self
1306 .schema
1307 .write()
1308 .expect("Schema lock poisoned - a thread panicked while holding it");
1309 *schema = Arc::new(new_schema);
1310 }
1311
1312 #[must_use]
1325 pub fn with_overlay(&self, overlay: &crate::core::fork::SchemaDelta) -> Arc<Self> {
1326 let primary = self.schema();
1327 let merged = if overlay.is_empty() {
1328 (*primary).clone()
1329 } else {
1330 let mut merged = (*primary).clone();
1331 for (name, label) in &overlay.added_labels {
1332 merged.labels.insert(name.clone(), label.clone());
1333 }
1334 for (name, edge_type) in &overlay.added_edge_types {
1335 merged.edge_types.insert(name.clone(), edge_type.clone());
1336 }
1337 for addition in &overlay.added_properties {
1338 let props = merged.properties.entry(addition.owner.clone()).or_default();
1339 props.insert(
1340 addition.property.clone(),
1341 PropertyMeta {
1342 r#type: addition.data_type.clone(),
1343 nullable: addition.nullable,
1344 added_in: merged.schema_version,
1345 state: SchemaElementState::Active,
1346 generation_expression: None,
1347 description: None,
1348 },
1349 );
1350 }
1351 merged
1352 };
1353
1354 Arc::new(Self {
1355 store: self.store.clone(),
1356 path: self.path.clone(),
1357 schema: RwLock::new(Arc::new(merged)),
1358 })
1359 }
1360
1361 pub fn next_label_id(&self) -> u16 {
1362 self.schema()
1363 .labels
1364 .values()
1365 .map(|l| l.id)
1366 .max()
1367 .unwrap_or(0)
1368 + 1
1369 }
1370
1371 pub fn next_type_id(&self) -> u32 {
1372 let max_schema_id = self
1373 .schema()
1374 .edge_types
1375 .values()
1376 .map(|t| t.id)
1377 .max()
1378 .unwrap_or(0);
1379
1380 if max_schema_id >= MAX_SCHEMA_TYPE_ID {
1382 panic!("Schema edge type ID exhaustion");
1383 }
1384
1385 max_schema_id + 1
1386 }
1387
1388 pub fn validate_schema_element_name(kind: &str, name: &str) -> Result<()> {
1406 if name.is_empty() || name.chars().all(char::is_whitespace) {
1407 return Err(anyhow!(
1408 "{kind} name must be non-empty and not all whitespace"
1409 ));
1410 }
1411 if name.len() > MAX_SCHEMA_NAME_LEN {
1412 return Err(anyhow!("{kind} name exceeds {MAX_SCHEMA_NAME_LEN} bytes"));
1413 }
1414 if let Some(c) = name
1415 .chars()
1416 .find(|c| c.is_control() || c.is_whitespace() || matches!(c, '/' | '\\'))
1417 {
1418 return Err(anyhow!(
1419 "{kind} name '{name}' contains an unsafe character ({c:?})"
1420 ));
1421 }
1422 Ok(())
1423 }
1424
1425 pub fn add_label(&self, name: &str) -> Result<u16> {
1426 self.add_label_with_desc(name, None)
1427 }
1428
1429 pub fn add_label_with_desc(&self, name: &str, description: Option<String>) -> Result<u16> {
1430 Self::validate_schema_element_name("Label", name)?;
1431 let mut guard = acquire_write(&self.schema, "schema")?;
1432 let schema = Arc::make_mut(&mut *guard);
1433 if schema.labels.contains_key(name) {
1434 return Err(anyhow!("Label '{}' already exists", name));
1435 }
1436
1437 let id = schema.labels.values().map(|l| l.id).max().unwrap_or(0) + 1;
1438 if id >= VIRTUAL_LABEL_ID_START {
1439 return Err(anyhow!(
1440 "Native label space exhausted (next id {id:#x} would enter the \
1441 virtual range {VIRTUAL_LABEL_ID_START:#x}..{VIRTUAL_LABEL_ID_SENTINEL:#x} \
1442 reserved for catalog-resolved labels)"
1443 ));
1444 }
1445 schema.labels.insert(
1446 name.to_string(),
1447 LabelMeta {
1448 id,
1449 created_at: Utc::now(),
1450 state: SchemaElementState::Active,
1451 description,
1452 },
1453 );
1454 schema.bump_version();
1455 Ok(id)
1456 }
1457
1458 pub fn add_edge_type(
1459 &self,
1460 name: &str,
1461 src_labels: Vec<String>,
1462 dst_labels: Vec<String>,
1463 ) -> Result<u32> {
1464 self.add_edge_type_with_desc(name, src_labels, dst_labels, None)
1465 }
1466
1467 pub fn add_edge_type_with_desc(
1468 &self,
1469 name: &str,
1470 src_labels: Vec<String>,
1471 dst_labels: Vec<String>,
1472 description: Option<String>,
1473 ) -> Result<u32> {
1474 Self::validate_schema_element_name("Edge type", name)?;
1475 let mut guard = acquire_write(&self.schema, "schema")?;
1476 let schema = Arc::make_mut(&mut *guard);
1477 if schema.edge_types.contains_key(name) {
1478 return Err(anyhow!("Edge type '{}' already exists", name));
1479 }
1480
1481 let id = schema.edge_types.values().map(|t| t.id).max().unwrap_or(0) + 1;
1482
1483 if id >= VIRTUAL_EDGE_TYPE_ID_START {
1488 return Err(anyhow!(
1489 "Native edge type space exhausted (next id {id:#x} would enter the \
1490 virtual range {VIRTUAL_EDGE_TYPE_ID_START:#x}..{VIRTUAL_EDGE_TYPE_ID_SENTINEL:#x} \
1491 reserved for catalog-resolved edge types)"
1492 ));
1493 }
1494
1495 schema.edge_types.insert(
1496 name.to_string(),
1497 EdgeTypeMeta {
1498 id,
1499 src_labels,
1500 dst_labels,
1501 state: SchemaElementState::Active,
1502 description,
1503 },
1504 );
1505 schema.bump_version();
1506 Ok(id)
1507 }
1508
1509 pub fn get_or_assign_edge_type_id(&self, type_name: &str) -> u32 {
1518 {
1519 let guard = acquire_read(&self.schema, "schema")
1520 .expect("Schema lock poisoned - a thread panicked while holding it");
1521 if let Some(id) = guard.edge_type_id_unified(type_name) {
1522 return id;
1523 }
1524 }
1525 let mut guard = acquire_write(&self.schema, "schema")
1526 .expect("Schema lock poisoned - a thread panicked while holding it");
1527 let schema = Arc::make_mut(&mut *guard);
1528 schema.get_or_assign_edge_type_id(type_name)
1529 }
1530
1531 pub fn edge_type_name_by_id_unified(&self, type_id: u32) -> Option<String> {
1533 let schema = acquire_read(&self.schema, "schema")
1534 .expect("Schema lock poisoned - a thread panicked while holding it");
1535 schema.edge_type_name_by_id_unified(type_id)
1536 }
1537
1538 pub fn add_property(
1539 &self,
1540 label_or_type: &str,
1541 prop_name: &str,
1542 data_type: DataType,
1543 nullable: bool,
1544 ) -> Result<()> {
1545 self.add_property_with_desc(label_or_type, prop_name, data_type, nullable, None)
1546 }
1547
1548 pub fn add_property_with_desc(
1549 &self,
1550 label_or_type: &str,
1551 prop_name: &str,
1552 data_type: DataType,
1553 nullable: bool,
1554 description: Option<String>,
1555 ) -> Result<()> {
1556 validate_property_name(prop_name)?;
1557 let mut guard = acquire_write(&self.schema, "schema")?;
1558 let schema = Arc::make_mut(&mut *guard);
1559 let version = schema.schema_version;
1560 let props = schema
1561 .properties
1562 .entry(label_or_type.to_string())
1563 .or_default();
1564
1565 if props.contains_key(prop_name) {
1566 return Err(anyhow!(
1567 "Property '{}' already exists for '{}'",
1568 prop_name,
1569 label_or_type
1570 ));
1571 }
1572
1573 props.insert(
1574 prop_name.to_string(),
1575 PropertyMeta {
1576 r#type: data_type,
1577 nullable,
1578 added_in: version,
1579 state: SchemaElementState::Active,
1580 generation_expression: None,
1581 description,
1582 },
1583 );
1584 schema.bump_version();
1586 Ok(())
1587 }
1588
1589 pub fn declare_property(
1607 &self,
1608 label_or_type: &str,
1609 prop_name: &str,
1610 data_type: DataType,
1611 nullable: bool,
1612 description: Option<String>,
1613 ) -> Result<bool> {
1614 validate_property_name(prop_name)?;
1615 let mut guard = acquire_write(&self.schema, "schema")?;
1616 let schema = Arc::make_mut(&mut *guard);
1617 let version = schema.schema_version;
1618 let props = schema
1619 .properties
1620 .entry(label_or_type.to_string())
1621 .or_default();
1622
1623 if let Some(existing) = props.get(prop_name) {
1624 if existing.r#type == data_type && existing.nullable == nullable {
1625 return Ok(false); }
1627 return Err(anyhow!(
1628 "Property '{}' on '{}' is declared as {:?} (nullable: {}); cannot re-declare \
1629 as {:?} (nullable: {}). Property types are immutable — use a new property \
1630 name or migrate the data",
1631 prop_name,
1632 label_or_type,
1633 existing.r#type,
1634 existing.nullable,
1635 data_type,
1636 nullable
1637 ));
1638 }
1639
1640 props.insert(
1641 prop_name.to_string(),
1642 PropertyMeta {
1643 r#type: data_type,
1644 nullable,
1645 added_in: version,
1646 state: SchemaElementState::Active,
1647 generation_expression: None,
1648 description,
1649 },
1650 );
1651 schema.bump_version();
1653 Ok(true)
1654 }
1655
1656 pub fn add_internal_property(
1667 &self,
1668 label_or_type: &str,
1669 prop_name: &str,
1670 data_type: DataType,
1671 nullable: bool,
1672 ) -> Result<bool> {
1673 validate_reserved_property_name(prop_name)?;
1674 let mut guard = acquire_write(&self.schema, "schema")?;
1675 let schema = Arc::make_mut(&mut *guard);
1676 let version = schema.schema_version;
1677 let props = schema
1678 .properties
1679 .entry(label_or_type.to_string())
1680 .or_default();
1681
1682 if let Some(existing) = props.get(prop_name) {
1683 if existing.r#type == data_type {
1684 return Ok(false); }
1686 return Err(anyhow!(
1687 "Internal property '{}' already exists for '{}' with a different type",
1688 prop_name,
1689 label_or_type
1690 ));
1691 }
1692
1693 props.insert(
1694 prop_name.to_string(),
1695 PropertyMeta {
1696 r#type: data_type,
1697 nullable,
1698 added_in: version,
1699 state: SchemaElementState::Active,
1700 generation_expression: None,
1701 description: None,
1702 },
1703 );
1704 schema.bump_version();
1705 Ok(true)
1706 }
1707
1708 pub fn add_generated_property(
1709 &self,
1710 label_or_type: &str,
1711 prop_name: &str,
1712 data_type: DataType,
1713 expr: String,
1714 ) -> Result<()> {
1715 validate_reserved_property_name(prop_name)?;
1718 let mut guard = acquire_write(&self.schema, "schema")?;
1719 let schema = Arc::make_mut(&mut *guard);
1720 let version = schema.schema_version;
1721 let props = schema
1722 .properties
1723 .entry(label_or_type.to_string())
1724 .or_default();
1725
1726 if props.contains_key(prop_name) {
1727 return Err(anyhow!("Property '{}' already exists", prop_name));
1728 }
1729
1730 props.insert(
1731 prop_name.to_string(),
1732 PropertyMeta {
1733 r#type: data_type,
1734 nullable: true,
1735 added_in: version,
1736 state: SchemaElementState::Active,
1737 generation_expression: Some(expr),
1738 description: None,
1739 },
1740 );
1741 schema.bump_version();
1743 Ok(())
1744 }
1745
1746 pub fn set_label_description(&self, name: &str, description: Option<String>) -> Result<()> {
1747 let mut guard = acquire_write(&self.schema, "schema")?;
1748 let schema = Arc::make_mut(&mut *guard);
1749 let meta = schema
1750 .labels
1751 .get_mut(name)
1752 .ok_or_else(|| anyhow!("Label '{}' does not exist", name))?;
1753 meta.description = description;
1754 Ok(())
1755 }
1756
1757 pub fn set_edge_type_description(&self, name: &str, description: Option<String>) -> Result<()> {
1758 let mut guard = acquire_write(&self.schema, "schema")?;
1759 let schema = Arc::make_mut(&mut *guard);
1760 let meta = schema
1761 .edge_types
1762 .get_mut(name)
1763 .ok_or_else(|| anyhow!("Edge type '{}' does not exist", name))?;
1764 meta.description = description;
1765 Ok(())
1766 }
1767
1768 pub fn set_property_description(
1769 &self,
1770 entity: &str,
1771 prop_name: &str,
1772 description: Option<String>,
1773 ) -> Result<()> {
1774 let mut guard = acquire_write(&self.schema, "schema")?;
1775 let schema = Arc::make_mut(&mut *guard);
1776 let props = schema
1777 .properties
1778 .get_mut(entity)
1779 .ok_or_else(|| anyhow!("Entity '{}' does not exist", entity))?;
1780 let meta = props
1781 .get_mut(prop_name)
1782 .ok_or_else(|| anyhow!("Property '{}' does not exist on '{}'", prop_name, entity))?;
1783 meta.description = description;
1784 Ok(())
1785 }
1786
1787 pub fn add_index(&self, index_def: IndexDefinition) -> Result<()> {
1796 let mut guard = acquire_write(&self.schema, "schema")?;
1797 let schema = Arc::make_mut(&mut *guard);
1798 if let Some(existing) = schema
1799 .indexes
1800 .iter_mut()
1801 .find(|i| i.name() == index_def.name())
1802 {
1803 *existing = index_def;
1804 } else {
1805 schema.indexes.push(index_def);
1806 }
1807 schema.bump_version();
1808 Ok(())
1809 }
1810
1811 pub fn get_index(&self, name: &str) -> Option<IndexDefinition> {
1812 let schema = self.schema.read().expect("Schema lock poisoned");
1813 schema.indexes.iter().find(|i| i.name() == name).cloned()
1814 }
1815
1816 pub fn update_index_metadata(
1821 &self,
1822 index_name: &str,
1823 f: impl FnOnce(&mut IndexMetadata),
1824 ) -> Result<()> {
1825 let mut guard = acquire_write(&self.schema, "schema")?;
1826 let schema = Arc::make_mut(&mut *guard);
1827 let idx = schema
1828 .indexes
1829 .iter_mut()
1830 .find(|i| i.name() == index_name)
1831 .ok_or_else(|| anyhow!("Index '{}' not found", index_name))?;
1832 f(idx.metadata_mut());
1833 Ok(())
1834 }
1835
1836 pub fn remove_index(&self, name: &str) -> Result<()> {
1837 let mut guard = acquire_write(&self.schema, "schema")?;
1838 let schema = Arc::make_mut(&mut *guard);
1839 if let Some(pos) = schema.indexes.iter().position(|i| i.name() == name) {
1840 schema.indexes.remove(pos);
1841 schema.bump_version();
1842 Ok(())
1843 } else {
1844 Err(anyhow!("Index '{}' not found", name))
1845 }
1846 }
1847
1848 pub fn add_constraint(&self, constraint: Constraint) -> Result<()> {
1849 let mut guard = acquire_write(&self.schema, "schema")?;
1850 let schema = Arc::make_mut(&mut *guard);
1851 if schema.constraints.iter().any(|c| c.name == constraint.name) {
1852 return Err(anyhow!("Constraint '{}' already exists", constraint.name));
1853 }
1854 schema.constraints.push(constraint);
1855 schema.bump_version();
1856 Ok(())
1857 }
1858
1859 pub fn drop_constraint(&self, name: &str, if_exists: bool) -> Result<()> {
1860 let mut guard = acquire_write(&self.schema, "schema")?;
1861 let schema = Arc::make_mut(&mut *guard);
1862 if let Some(pos) = schema.constraints.iter().position(|c| c.name == name) {
1863 schema.constraints.remove(pos);
1864 schema.bump_version();
1865 Ok(())
1866 } else if if_exists {
1867 Ok(())
1868 } else {
1869 Err(anyhow!("Constraint '{}' not found", name))
1870 }
1871 }
1872
1873 pub fn drop_property(&self, label_or_type: &str, prop_name: &str) -> Result<()> {
1874 let mut guard = acquire_write(&self.schema, "schema")?;
1875 let schema = Arc::make_mut(&mut *guard);
1876 let Some(props) = schema.properties.get_mut(label_or_type) else {
1877 return Err(anyhow!("Label or Edge Type '{}' not found", label_or_type));
1878 };
1879 if props.remove(prop_name).is_none() {
1880 return Err(anyhow!(
1881 "Property '{}' not found for '{}'",
1882 prop_name,
1883 label_or_type
1884 ));
1885 }
1886 schema.bump_version();
1887 Ok(())
1888 }
1889
1890 pub fn rename_property(
1891 &self,
1892 label_or_type: &str,
1893 old_name: &str,
1894 new_name: &str,
1895 ) -> Result<()> {
1896 validate_property_name(new_name)?;
1901 let mut guard = acquire_write(&self.schema, "schema")?;
1902 let schema = Arc::make_mut(&mut *guard);
1903 let Some(props) = schema.properties.get_mut(label_or_type) else {
1904 return Err(anyhow!("Label or Edge Type '{}' not found", label_or_type));
1905 };
1906 let Some(meta) = props.remove(old_name) else {
1907 return Err(anyhow!(
1908 "Property '{}' not found for '{}'",
1909 old_name,
1910 label_or_type
1911 ));
1912 };
1913 if props.contains_key(new_name) {
1914 props.insert(old_name.to_string(), meta); return Err(anyhow!("Property '{}' already exists", new_name));
1917 }
1918 props.insert(new_name.to_string(), meta);
1919 schema.bump_version();
1920 Ok(())
1921 }
1922
1923 pub fn drop_label(&self, name: &str, if_exists: bool) -> Result<()> {
1924 let mut guard = acquire_write(&self.schema, "schema")?;
1925 let schema = Arc::make_mut(&mut *guard);
1926 if let Some(label_meta) = schema.labels.get_mut(name) {
1927 label_meta.state = SchemaElementState::Tombstone { since: Utc::now() };
1928 schema.bump_version();
1930 Ok(())
1931 } else if if_exists {
1932 Ok(())
1933 } else {
1934 Err(anyhow!("Label '{}' not found", name))
1935 }
1936 }
1937
1938 pub fn drop_edge_type(&self, name: &str, if_exists: bool) -> Result<()> {
1939 let mut guard = acquire_write(&self.schema, "schema")?;
1940 let schema = Arc::make_mut(&mut *guard);
1941 if let Some(edge_meta) = schema.edge_types.get_mut(name) {
1942 edge_meta.state = SchemaElementState::Tombstone { since: Utc::now() };
1943 schema.bump_version();
1945 Ok(())
1946 } else if if_exists {
1947 Ok(())
1948 } else {
1949 Err(anyhow!("Edge Type '{}' not found", name))
1950 }
1951 }
1952}
1953
1954pub fn validate_identifier(name: &str) -> Result<()> {
1956 if name.is_empty() || name.len() > 64 {
1958 return Err(anyhow!("Identifier '{}' must be 1-64 characters", name));
1959 }
1960
1961 let first = name.chars().next().unwrap();
1963 if !first.is_alphabetic() && first != '_' {
1964 return Err(anyhow!(
1965 "Identifier '{}' must start with letter or underscore",
1966 name
1967 ));
1968 }
1969
1970 if !name.chars().all(|c| c.is_alphanumeric() || c == '_') {
1972 return Err(anyhow!(
1973 "Identifier '{}' must contain only alphanumeric and underscore",
1974 name
1975 ));
1976 }
1977
1978 const RESERVED: &[&str] = &[
1980 "MATCH", "CREATE", "DELETE", "SET", "RETURN", "WHERE", "MERGE", "CALL", "YIELD", "WITH",
1981 "UNION", "ORDER", "LIMIT",
1982 ];
1983 if RESERVED.contains(&name.to_uppercase().as_str()) {
1984 return Err(anyhow!("Identifier '{}' cannot be a reserved word", name));
1985 }
1986
1987 Ok(())
1988}
1989
1990pub fn validate_property_name(name: &str) -> Result<()> {
1997 if name.starts_with('_') {
1998 return Err(anyhow!(
1999 "Property name '{}' is reserved: names starting with '_' are reserved by the storage layer",
2000 name
2001 ));
2002 }
2003 validate_reserved_property_name(name)
2004}
2005
2006fn validate_reserved_property_name(name: &str) -> Result<()> {
2013 const RESERVED_PROPS: &[&str] = &[
2022 "ext_id",
2023 "overflow_json",
2024 "eid",
2025 "src_vid",
2026 "dst_vid",
2027 "op",
2028 "__set_struct__",
2036 ];
2037 if RESERVED_PROPS.contains(&name) {
2038 return Err(anyhow!(
2039 "Property name '{}' is reserved by the storage layer; please choose a different name",
2040 name
2041 ));
2042 }
2043 Ok(())
2044}
2045
2046#[cfg(test)]
2047mod tests {
2048 use super::*;
2049 use crate::value::{TemporalValue, Value};
2050 use object_store::local::LocalFileSystem;
2051 use tempfile::tempdir;
2052
2053 #[test]
2054 fn binary_vector_metrics_exact() {
2055 assert_eq!(
2058 DistanceMetric::Hamming.compute_distance_binary(&[0x00], &[0xFF]),
2059 8.0
2060 );
2061 assert_eq!(
2062 DistanceMetric::Hamming.compute_distance_binary(&[0xA5, 0x0F], &[0xA5, 0x00]),
2063 4.0
2064 );
2065 assert_eq!(
2066 DistanceMetric::Hamming.compute_distance_binary(&[0xA5], &[0xA5]),
2067 0.0
2068 );
2069
2070 let j = DistanceMetric::Jaccard.compute_distance_binary(&[0b1100], &[0b1010]);
2073 assert!((j - (2.0 / 3.0)).abs() < 1e-6, "got {j}");
2074 assert_eq!(
2076 DistanceMetric::Jaccard.compute_distance_binary(&[0xFF], &[0xFF]),
2077 0.0
2078 );
2079 assert_eq!(
2081 DistanceMetric::Jaccard.compute_distance_binary(&[0x00, 0x00], &[0x00, 0x00]),
2082 0.0
2083 );
2084 }
2085
2086 #[test]
2087 fn binary_metrics_are_binary_and_route_correctly() {
2088 assert!(DistanceMetric::Hamming.is_binary());
2089 assert!(DistanceMetric::Jaccard.is_binary());
2090 assert!(!DistanceMetric::L2.is_binary());
2091 assert!(!DistanceMetric::L1.is_binary());
2092 }
2093
2094 #[test]
2095 #[should_panic(expected = "binary-vector metric")]
2096 fn float_compute_distance_rejects_binary_metric() {
2097 DistanceMetric::Hamming.compute_distance(&[1.0], &[0.0]);
2098 }
2099
2100 #[test]
2101 fn check_binary_vector_value_guards() {
2102 let ty = DataType::BinaryVector { dimensions: 3 };
2103 assert!(
2104 ty.check_vector_dims(&Value::BinaryVector(vec![1, 2, 3]))
2105 .is_ok()
2106 );
2107 assert!(ty.check_vector_dims(&Value::Null).is_ok());
2108 assert!(
2110 ty.check_vector_dims(&Value::BinaryVector(vec![1, 2]))
2111 .is_err()
2112 );
2113 assert!(
2115 ty.check_vector_dims(&Value::List(vec![
2116 Value::Int(0),
2117 Value::Int(255),
2118 Value::Int(128)
2119 ]))
2120 .is_ok()
2121 );
2122 assert!(
2124 ty.check_vector_dims(&Value::List(vec![
2125 Value::Int(0),
2126 Value::Int(256),
2127 Value::Int(1)
2128 ]))
2129 .is_err()
2130 );
2131 }
2132
2133 #[test]
2134 fn test_datatype_accepts_matrix() {
2135 let dt = || TemporalValue::DateTime {
2136 nanos_since_epoch: 0,
2137 offset_seconds: 0,
2138 timezone_name: None,
2139 };
2140
2141 for ty in [
2143 DataType::String,
2144 DataType::Int64,
2145 DataType::Bool,
2146 DataType::DateTime,
2147 DataType::Float64,
2148 ] {
2149 assert!(ty.accepts(&Value::Null), "{ty:?} must accept Null");
2150 }
2151
2152 assert!(DataType::String.accepts(&Value::String("x".into())));
2154 assert!(DataType::Int64.accepts(&Value::Int(1)));
2155 assert!(DataType::Bool.accepts(&Value::Bool(true)));
2156 assert!(DataType::DateTime.accepts(&Value::Temporal(dt())));
2157
2158 assert!(
2160 DataType::Float64.accepts(&Value::Int(3)),
2161 "Int widens to Float"
2162 );
2163 assert!(DataType::Int32.accepts(&Value::Int(3)), "Int fits Int32");
2164 assert!(DataType::Timestamp.accepts(&Value::Temporal(dt())));
2165 assert!(
2166 DataType::Timestamp.accepts(&Value::String("2026-01-01T00:00:00Z".into())),
2167 "storage parses strings for non-struct Timestamp columns"
2168 );
2169
2170 assert!(
2172 !DataType::DateTime.accepts(&Value::String("2026-01-01T00:00:00Z".into())),
2173 "String into a DateTime struct column nulls silently — reject here"
2174 );
2175 assert!(!DataType::Bool.accepts(&Value::Int(1)));
2176 assert!(!DataType::Int64.accepts(&Value::Bool(true)));
2177 assert!(!DataType::Int64.accepts(&Value::Float(1.5)));
2178 assert!(
2179 !DataType::String.accepts(&Value::Int(10)),
2180 "no implicit stringification"
2181 );
2182 assert!(!DataType::Duration.accepts(&Value::String("P1D".into())));
2183
2184 assert!(DataType::CypherValue.accepts(&Value::Map(Default::default())));
2186 }
2187
2188 #[test]
2189 fn test_check_vector_dims_matrix() {
2190 let vec3 = DataType::Vector { dimensions: 3 };
2191 let multi2 = DataType::List(Box::new(DataType::Vector { dimensions: 2 }));
2192 let flist = |vals: &[f64]| Value::List(vals.iter().map(|f| Value::Float(*f)).collect());
2193
2194 assert!(vec3.check_vector_dims(&Value::Null).is_ok());
2196 assert!(multi2.check_vector_dims(&Value::Null).is_ok());
2197
2198 assert!(
2200 vec3.check_vector_dims(&Value::Vector(vec![1.0, 2.0, 3.0]))
2201 .is_ok()
2202 );
2203 assert!(vec3.check_vector_dims(&flist(&[1.0, 2.0, 3.0])).is_ok());
2204 assert!(
2205 vec3.check_vector_dims(&Value::List(vec![
2206 Value::Int(1),
2207 Value::Float(2.0),
2208 Value::Int(3)
2209 ]))
2210 .is_ok()
2211 );
2212
2213 assert_eq!(
2215 vec3.check_vector_dims(&Value::Vector(vec![1.0, 2.0])),
2216 Err(VectorDimError::WrongLength {
2217 expected: 3,
2218 actual: 2
2219 })
2220 );
2221 assert_eq!(
2222 vec3.check_vector_dims(&flist(&[1.0, 2.0, 3.0, 4.0, 5.0])),
2223 Err(VectorDimError::WrongLength {
2224 expected: 3,
2225 actual: 5
2226 })
2227 );
2228 assert_eq!(
2229 vec3.check_vector_dims(&Value::List(vec![])),
2230 Err(VectorDimError::WrongLength {
2231 expected: 3,
2232 actual: 0
2233 })
2234 );
2235 assert_eq!(
2236 vec3.check_vector_dims(&Value::List(vec![
2237 Value::Float(1.0),
2238 Value::String("x".into()),
2239 Value::Float(3.0),
2240 ])),
2241 Err(VectorDimError::NonNumericElement { index: 1 })
2242 );
2243 assert_eq!(
2244 vec3.check_vector_dims(&Value::List(vec![
2245 Value::Float(1.0),
2246 Value::Null,
2247 Value::Float(3.0)
2248 ])),
2249 Err(VectorDimError::NonNumericElement { index: 1 })
2250 );
2251 assert_eq!(
2252 vec3.check_vector_dims(&Value::String("not a vector".into())),
2253 Err(VectorDimError::NotAVector { actual: "String" })
2254 );
2255
2256 assert!(multi2.check_vector_dims(&Value::List(vec![])).is_ok());
2259 assert!(
2260 multi2
2261 .check_vector_dims(&Value::List(vec![flist(&[1.0, 2.0]), flist(&[3.0, 4.0])]))
2262 .is_ok()
2263 );
2264 assert_eq!(
2265 multi2.check_vector_dims(&Value::List(vec![
2266 flist(&[1.0, 2.0]),
2267 flist(&[9.0, 9.0, 9.0])
2268 ])),
2269 Err(VectorDimError::TokenWrongLength {
2270 token: 1,
2271 expected: 2,
2272 actual: 3
2273 })
2274 );
2275 assert_eq!(
2276 multi2.check_vector_dims(&Value::List(vec![Value::String("tok".into())])),
2277 Err(VectorDimError::TokenNotAVector {
2278 token: 0,
2279 actual: "String"
2280 })
2281 );
2282 assert_eq!(
2283 multi2.check_vector_dims(&Value::Vector(vec![1.0, 2.0])),
2284 Err(VectorDimError::NotATokenList { actual: "Vector" })
2285 );
2286
2287 assert!(
2289 DataType::Int64
2290 .check_vector_dims(&Value::String("x".into()))
2291 .is_ok()
2292 );
2293 assert!(
2294 DataType::List(Box::new(DataType::Float64))
2295 .check_vector_dims(&Value::List(vec![Value::String("x".into())]))
2296 .is_ok()
2297 );
2298 assert!(
2299 DataType::SparseVector { dimensions: 8 }
2300 .check_vector_dims(&Value::Map(Default::default()))
2301 .is_ok()
2302 );
2303
2304 let msg = VectorDimError::WrongLength {
2306 expected: 4,
2307 actual: 5,
2308 }
2309 .to_string();
2310 assert!(msg.contains('4') && msg.contains('5'), "message: {msg}");
2311 }
2312
2313 #[tokio::test]
2314 async fn test_declare_property_idempotent_and_conflicting() -> Result<()> {
2315 let dir = tempdir()?;
2316 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2317 let path = ObjectStorePath::from("schema.json");
2318 let manager = SchemaManager::load_from_store(store.clone(), &path).await?;
2319
2320 manager.add_label("Doc")?;
2321 let vec4 = DataType::Vector { dimensions: 4 };
2322
2323 assert!(manager.declare_property("Doc", "embedding", vec4.clone(), true, None)?);
2325
2326 assert!(!manager.declare_property("Doc", "embedding", vec4.clone(), true, None)?);
2329 assert!(!manager.declare_property(
2330 "Doc",
2331 "embedding",
2332 vec4.clone(),
2333 true,
2334 Some("new docs".into())
2335 )?);
2336
2337 let err = manager
2340 .declare_property(
2341 "Doc",
2342 "embedding",
2343 DataType::Vector { dimensions: 8 },
2344 true,
2345 None,
2346 )
2347 .unwrap_err()
2348 .to_string();
2349 assert!(err.contains('4') && err.contains('8'), "message: {err}");
2350 assert!(!err.contains("already exists"), "message: {err}");
2351
2352 assert!(
2354 manager
2355 .declare_property("Doc", "embedding", vec4.clone(), false, None)
2356 .is_err()
2357 );
2358
2359 let schema = manager.schema();
2361 let meta = &schema.properties["Doc"]["embedding"];
2362 assert_eq!(meta.r#type, vec4);
2363 assert!(meta.nullable);
2364 Ok(())
2365 }
2366
2367 #[tokio::test]
2368 async fn test_schema_management() -> Result<()> {
2369 let dir = tempdir()?;
2370 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2371 let path = ObjectStorePath::from("schema.json");
2372 let manager = SchemaManager::load_from_store(store.clone(), &path).await?;
2373
2374 let lid = manager.add_label("Person")?;
2376 assert_eq!(lid, 1);
2377 assert!(manager.add_label("Person").is_err());
2378
2379 manager.add_property("Person", "name", DataType::String, false)?;
2381 assert!(
2382 manager
2383 .add_property("Person", "name", DataType::String, false)
2384 .is_err()
2385 );
2386
2387 let tid = manager.add_edge_type("knows", vec!["Person".into()], vec!["Person".into()])?;
2389 assert_eq!(tid, 1);
2390
2391 manager.save().await?;
2392 assert!(store.get(&path).await.is_ok());
2394
2395 let manager2 = SchemaManager::load_from_store(store, &path).await?;
2396 assert!(manager2.schema().labels.contains_key("Person"));
2397 assert!(
2398 manager2
2399 .schema()
2400 .properties
2401 .get("Person")
2402 .unwrap()
2403 .contains_key("name")
2404 );
2405
2406 Ok(())
2407 }
2408
2409 #[tokio::test]
2410 async fn test_reserved_property_names_rejected() -> Result<()> {
2411 let dir = tempdir()?;
2412 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2413 let path = ObjectStorePath::from("schema.json");
2414 let manager = SchemaManager::load_from_store(store, &path).await?;
2415
2416 manager.add_label("Tiny")?;
2417
2418 for reserved in &["ext_id", "overflow_json", "eid", "src_vid", "dst_vid", "op"] {
2422 let err = manager
2423 .add_property("Tiny", reserved, DataType::String, true)
2424 .expect_err(&format!("expected '{reserved}' to be rejected"));
2425 assert!(
2426 err.to_string().contains("reserved"),
2427 "error for '{reserved}' should mention 'reserved', got: {err}"
2428 );
2429 }
2430
2431 let err = manager
2436 .add_property("Tiny", "__set_struct__", DataType::String, true)
2437 .expect_err("expected '__set_struct__' to be rejected");
2438 assert!(
2439 err.to_string().contains("reserved"),
2440 "__set_struct__ rejection should mention 'reserved', got: {err}"
2441 );
2442
2443 for reserved in &["_vid", "_uid", "_eid", "_version", "_created_at"] {
2445 assert!(
2446 manager
2447 .add_property("Tiny", reserved, DataType::String, true)
2448 .is_err(),
2449 "expected '{reserved}' to be rejected"
2450 );
2451 }
2452
2453 manager.add_property("Tiny", "ext_id_foo", DataType::String, true)?;
2456 manager.add_property("Tiny", "user_op", DataType::String, true)?;
2457 manager.add_property("Tiny", "type_name", DataType::String, true)?;
2458
2459 manager.add_edge_type("knows", vec!["Tiny".into()], vec!["Tiny".into()])?;
2461 assert!(
2462 manager
2463 .add_property("knows", "src_vid", DataType::Int64, true)
2464 .is_err()
2465 );
2466
2467 assert!(
2469 manager
2470 .add_generated_property(
2471 "Tiny",
2472 "ext_id",
2473 DataType::String,
2474 "concat('x', name)".into()
2475 )
2476 .is_err()
2477 );
2478
2479 Ok(())
2480 }
2481
2482 #[test]
2483 fn test_normalize_function_names() {
2484 assert_eq!(
2485 SchemaManager::normalize_function_names("lower(email)"),
2486 "LOWER(email)"
2487 );
2488 assert_eq!(
2489 SchemaManager::normalize_function_names("LOWER(email)"),
2490 "LOWER(email)"
2491 );
2492 assert_eq!(
2493 SchemaManager::normalize_function_names("Lower(email)"),
2494 "LOWER(email)"
2495 );
2496 assert_eq!(
2497 SchemaManager::normalize_function_names("trim(lower(email))"),
2498 "TRIM(LOWER(email))"
2499 );
2500 }
2501
2502 #[test]
2503 fn test_generated_column_name_case_insensitive() {
2504 let col1 = SchemaManager::generated_column_name("lower(email)");
2505 let col2 = SchemaManager::generated_column_name("LOWER(email)");
2506 let col3 = SchemaManager::generated_column_name("Lower(email)");
2507 assert_eq!(col1, col2);
2508 assert_eq!(col2, col3);
2509 assert!(col1.starts_with("_gen_LOWER_email_"));
2510 }
2511
2512 #[test]
2513 fn test_index_metadata_serde_backward_compat() {
2514 let json = r#"{
2516 "type": "Scalar",
2517 "name": "idx_person_name",
2518 "label": "Person",
2519 "properties": ["name"],
2520 "index_type": "BTree",
2521 "where_clause": null
2522 }"#;
2523 let def: IndexDefinition = serde_json::from_str(json).unwrap();
2524 let meta = def.metadata();
2525 assert_eq!(meta.status, IndexStatus::Online);
2526 assert!(meta.last_built_at.is_none());
2527 assert!(meta.row_count_at_build.is_none());
2528 }
2529
2530 #[test]
2531 fn test_index_metadata_serde_roundtrip() {
2532 let now = Utc::now();
2533 let def = IndexDefinition::Scalar(ScalarIndexConfig {
2534 name: "idx_test".to_string(),
2535 label: "Test".to_string(),
2536 properties: vec!["prop".to_string()],
2537 index_type: ScalarIndexType::BTree,
2538 where_clause: None,
2539 metadata: IndexMetadata {
2540 status: IndexStatus::Building,
2541 last_built_at: Some(now),
2542 row_count_at_build: Some(42),
2543 },
2544 });
2545
2546 let json = serde_json::to_string(&def).unwrap();
2547 let parsed: IndexDefinition = serde_json::from_str(&json).unwrap();
2548 assert_eq!(parsed.metadata().status, IndexStatus::Building);
2549 assert_eq!(parsed.metadata().row_count_at_build, Some(42));
2550 assert!(parsed.metadata().last_built_at.is_some());
2551 }
2552
2553 #[tokio::test]
2554 async fn test_update_index_metadata() -> Result<()> {
2555 let dir = tempdir()?;
2556 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2557 let path = ObjectStorePath::from("schema.json");
2558 let manager = SchemaManager::load_from_store(store, &path).await?;
2559
2560 manager.add_label("Person")?;
2561 let idx = IndexDefinition::Scalar(ScalarIndexConfig {
2562 name: "idx_test".to_string(),
2563 label: "Person".to_string(),
2564 properties: vec!["name".to_string()],
2565 index_type: ScalarIndexType::BTree,
2566 where_clause: None,
2567 metadata: Default::default(),
2568 });
2569 manager.add_index(idx)?;
2570
2571 let initial = manager.get_index("idx_test").unwrap();
2573 assert_eq!(initial.metadata().status, IndexStatus::Online);
2574
2575 manager.update_index_metadata("idx_test", |m| {
2577 m.status = IndexStatus::Building;
2578 m.row_count_at_build = Some(100);
2579 })?;
2580
2581 let updated = manager.get_index("idx_test").unwrap();
2582 assert_eq!(updated.metadata().status, IndexStatus::Building);
2583 assert_eq!(updated.metadata().row_count_at_build, Some(100));
2584
2585 assert!(manager.update_index_metadata("nope", |_| {}).is_err());
2587
2588 Ok(())
2589 }
2590
2591 #[tokio::test]
2596 async fn add_internal_property_reports_newly_added() -> Result<()> {
2597 let dir = tempdir()?;
2598 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2599 let path = ObjectStorePath::from("schema.json");
2600 let manager = SchemaManager::load_from_store(store, &path).await?;
2601 manager.add_label("Doc")?;
2602
2603 let dt = DataType::Vector { dimensions: 16 };
2604 assert!(manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
2606 assert!(!manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
2608 assert!(
2610 manager
2611 .add_internal_property("Doc", "__fde_x", DataType::Vector { dimensions: 8 }, true)
2612 .is_err()
2613 );
2614 Ok(())
2615 }
2616
2617 #[tokio::test]
2622 async fn test_add_index_is_upsert_by_name() -> Result<()> {
2623 let dir = tempdir()?;
2624 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2625 let path = ObjectStorePath::from("schema.json");
2626 let manager = SchemaManager::load_from_store(store, &path).await?;
2627 manager.add_label("Person")?;
2628
2629 let initial = IndexDefinition::Scalar(ScalarIndexConfig {
2630 name: "idx_test".to_string(),
2631 label: "Person".to_string(),
2632 properties: vec!["name".to_string()],
2633 index_type: ScalarIndexType::BTree,
2634 where_clause: None,
2635 metadata: IndexMetadata {
2636 status: IndexStatus::Building,
2637 ..Default::default()
2638 },
2639 });
2640 manager.add_index(initial.clone())?;
2641 assert_eq!(manager.schema().indexes.len(), 1);
2642
2643 manager.add_index(initial.clone())?;
2645 assert_eq!(
2646 manager.schema().indexes.len(),
2647 1,
2648 "duplicate add_index by name must not append"
2649 );
2650
2651 let mut updated_cfg = match initial {
2653 IndexDefinition::Scalar(c) => c,
2654 _ => unreachable!(),
2655 };
2656 updated_cfg.metadata.status = IndexStatus::Online;
2657 updated_cfg.metadata.row_count_at_build = Some(42);
2658 manager.add_index(IndexDefinition::Scalar(updated_cfg))?;
2659 assert_eq!(manager.schema().indexes.len(), 1);
2660 let stored = manager.get_index("idx_test").unwrap();
2661 assert_eq!(stored.metadata().status, IndexStatus::Online);
2662 assert_eq!(stored.metadata().row_count_at_build, Some(42));
2663
2664 let other = IndexDefinition::Scalar(ScalarIndexConfig {
2666 name: "idx_other".to_string(),
2667 label: "Person".to_string(),
2668 properties: vec!["age".to_string()],
2669 index_type: ScalarIndexType::BTree,
2670 where_clause: None,
2671 metadata: IndexMetadata::default(),
2672 });
2673 manager.add_index(other)?;
2674 assert_eq!(manager.schema().indexes.len(), 2);
2675
2676 Ok(())
2677 }
2678
2679 #[tokio::test]
2682 async fn test_load_dedups_bloated_indexes() -> Result<()> {
2683 let dir = tempdir()?;
2684 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2685 let path = ObjectStorePath::from("schema.json");
2686
2687 let mut schema = Schema::default();
2691 schema.labels.insert(
2692 "Person".to_string(),
2693 LabelMeta {
2694 id: 1,
2695 created_at: chrono::Utc::now(),
2696 state: SchemaElementState::Active,
2697 description: None,
2698 },
2699 );
2700 let make = |status: IndexStatus, count: Option<u64>| {
2701 IndexDefinition::Scalar(ScalarIndexConfig {
2702 name: "idx_dup".to_string(),
2703 label: "Person".to_string(),
2704 properties: vec!["name".to_string()],
2705 index_type: ScalarIndexType::BTree,
2706 where_clause: None,
2707 metadata: IndexMetadata {
2708 status,
2709 row_count_at_build: count,
2710 ..Default::default()
2711 },
2712 })
2713 };
2714 for _ in 0..49 {
2715 schema.indexes.push(make(IndexStatus::Building, None));
2716 }
2717 schema.indexes.push(make(IndexStatus::Online, Some(123)));
2718 let json = serde_json::to_string_pretty(&schema)?;
2719 store.put(&path, json.into()).await?;
2720
2721 let manager = SchemaManager::load_from_store(store, &path).await?;
2722 let schema = manager.schema();
2723 assert_eq!(
2724 schema.indexes.len(),
2725 1,
2726 "load() must collapse 50 duplicates by name to 1"
2727 );
2728 assert_eq!(schema.indexes[0].metadata().status, IndexStatus::Online);
2730 assert_eq!(schema.indexes[0].metadata().row_count_at_build, Some(123));
2731
2732 Ok(())
2733 }
2734
2735 #[test]
2736 fn test_vector_index_for_property_skips_non_online() {
2737 let mut schema = Schema::default();
2738 schema.labels.insert(
2739 "Document".to_string(),
2740 LabelMeta {
2741 id: 1,
2742 created_at: chrono::Utc::now(),
2743 state: SchemaElementState::Active,
2744 description: None,
2745 },
2746 );
2747
2748 schema
2750 .indexes
2751 .push(IndexDefinition::Vector(VectorIndexConfig {
2752 name: "vec_doc_embedding".to_string(),
2753 label: "Document".to_string(),
2754 property: "embedding".to_string(),
2755 index_type: VectorIndexType::Flat,
2756 metric: DistanceMetric::Cosine,
2757 embedding_config: None,
2758 metadata: IndexMetadata {
2759 status: IndexStatus::Stale,
2760 ..Default::default()
2761 },
2762 }));
2763
2764 assert!(
2766 schema
2767 .vector_index_for_property("Document", "embedding")
2768 .is_none()
2769 );
2770
2771 if let IndexDefinition::Vector(cfg) = &mut schema.indexes[0] {
2773 cfg.metadata.status = IndexStatus::Online;
2774 }
2775 let result = schema.vector_index_for_property("Document", "embedding");
2776 assert!(result.is_some());
2777 assert_eq!(result.unwrap().metric, DistanceMetric::Cosine);
2778 }
2779
2780 #[tokio::test]
2781 async fn with_overlay_empty_clones_primary_in_isolation() -> Result<()> {
2782 use crate::core::fork::SchemaDelta;
2783
2784 let dir = tempdir()?;
2785 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2786 let path = ObjectStorePath::from("schema.json");
2787 let primary = SchemaManager::load_from_store(store, &path).await?;
2788 primary.add_label("Person")?;
2789
2790 let overlay = primary.with_overlay(&SchemaDelta::empty());
2791 assert_eq!(overlay.schema().labels.len(), 1);
2792
2793 overlay.add_label("Forked")?;
2796 assert!(overlay.schema().labels.contains_key("Forked"));
2797 assert!(!primary.schema().labels.contains_key("Forked"));
2798
2799 Ok(())
2800 }
2801
2802 #[tokio::test]
2803 async fn with_overlay_merges_added_labels_and_edge_types() -> Result<()> {
2804 use crate::core::fork::SchemaDelta;
2805 use chrono::Utc;
2806
2807 let dir = tempdir()?;
2808 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2809 let path = ObjectStorePath::from("schema.json");
2810 let primary = SchemaManager::load_from_store(store, &path).await?;
2811 primary.add_label("Existing")?;
2812
2813 let label_meta = LabelMeta {
2814 id: 99,
2815 created_at: Utc::now(),
2816 state: SchemaElementState::Active,
2817 description: None,
2818 };
2819 let edge_meta = EdgeTypeMeta {
2820 id: 99,
2821 src_labels: vec!["NewLabel".into()],
2822 dst_labels: vec!["NewLabel".into()],
2823 state: SchemaElementState::Active,
2824 description: None,
2825 };
2826 let delta = SchemaDelta {
2827 added_labels: vec![("NewLabel".to_string(), label_meta)],
2828 added_edge_types: vec![("NewEdge".to_string(), edge_meta)],
2829 added_properties: vec![],
2830 };
2831
2832 let overlay = primary.with_overlay(&delta);
2833 let merged = overlay.schema();
2834 assert!(merged.labels.contains_key("Existing"));
2835 assert!(merged.labels.contains_key("NewLabel"));
2836 assert!(merged.edge_types.contains_key("NewEdge"));
2837
2838 assert!(!primary.schema().labels.contains_key("NewLabel"));
2840 Ok(())
2841 }
2842
2843 #[tokio::test]
2848 async fn test_get_or_assign_edge_type_id_concurrent() -> Result<()> {
2849 let dir = tempdir()?;
2850 let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2851 let path = ObjectStorePath::from("schema.json");
2852 let manager = Arc::new(SchemaManager::load_from_store(store, &path).await?);
2853
2854 let mut handles = Vec::new();
2855 for _ in 0..16 {
2856 let m = manager.clone();
2857 handles.push(std::thread::spawn(move || {
2858 m.get_or_assign_edge_type_id("RACED")
2859 }));
2860 }
2861 let ids: Vec<u32> = handles.into_iter().map(|h| h.join().unwrap()).collect();
2862 assert!(
2863 ids.iter().all(|&id| id == ids[0]),
2864 "all racers must observe one id, got {ids:?}"
2865 );
2866 assert_eq!(manager.get_or_assign_edge_type_id("RACED"), ids[0]);
2868
2869 manager.add_label("A")?;
2871 let declared = manager.add_edge_type("DECLARED", vec!["A".into()], vec!["A".into()])?;
2872 assert_eq!(manager.get_or_assign_edge_type_id("DECLARED"), declared);
2873 Ok(())
2874 }
2875
2876 #[test]
2881 fn test_new_schemaless_edge_type_bumps_schema_version() {
2882 let mut schema = Schema::default();
2883 let v0 = schema.schema_version;
2884
2885 let id1 = schema.get_or_assign_edge_type_id("FRESH");
2886 assert_eq!(
2887 schema.schema_version,
2888 v0.wrapping_add(1),
2889 "minting a new edge type must bump schema_version"
2890 );
2891
2892 let id1_again = schema.get_or_assign_edge_type_id("FRESH");
2894 assert_eq!(id1, id1_again);
2895 assert_eq!(
2896 schema.schema_version,
2897 v0.wrapping_add(1),
2898 "resolving an existing edge type must not bump schema_version"
2899 );
2900
2901 let _id2 = schema.get_or_assign_edge_type_id("OTHER");
2903 assert_eq!(
2904 schema.schema_version,
2905 v0.wrapping_add(2),
2906 "a second new edge type must bump schema_version again"
2907 );
2908 }
2909
2910 #[test]
2914 fn validate_schema_element_name_rejects_unsafe() {
2915 for bad in ["", " ", "a/b", "a b", "a\nb", "a\\b", "x\0y"] {
2916 assert!(
2917 SchemaManager::validate_schema_element_name("Label", bad).is_err(),
2918 "expected {bad:?} to be rejected"
2919 );
2920 }
2921 for good in ["Person", "My.Label", "edge_2", "KNOWS"] {
2922 assert!(
2923 SchemaManager::validate_schema_element_name("Label", good).is_ok(),
2924 "expected {good:?} to be accepted"
2925 );
2926 }
2927 let long = "x".repeat(MAX_SCHEMA_NAME_LEN + 1);
2929 assert!(SchemaManager::validate_schema_element_name("Label", &long).is_err());
2930 }
2931}