Skip to main content

uni_common/core/
schema.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2024-2026 Dragonscale Team
3
4use 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, // SnapshotId
30    },
31    Tombstone {
32        since: DateTime<Utc>,
33    },
34}
35
36use arrow_schema::{DataType as ArrowDataType, Field, Fields, TimeUnit};
37
38/// Returns the canonical struct field definitions for DateTime encoding in Arrow.
39///
40/// DateTime is encoded as a 3-field struct to preserve timezone information:
41/// - `nanos_since_epoch`: i64 nanoseconds since Unix epoch (UTC)
42/// - `offset_seconds`: i32 seconds offset from UTC (e.g., +3600 for +01:00)
43/// - `timezone_name`: Optional IANA timezone name (e.g., "America/New_York")
44pub 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
56/// Returns the canonical struct field definitions for Time encoding in Arrow.
57///
58/// Time is encoded as a 2-field struct to preserve timezone offset:
59/// - `nanos_since_midnight`: i64 nanoseconds since midnight (0-86,399,999,999,999)
60/// - `offset_seconds`: i32 seconds offset from UTC (e.g., +3600 for +01:00)
61pub 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
72/// Detects if an Arrow DataType is the canonical DateTime struct.
73pub fn is_datetime_struct(arrow_dt: &ArrowDataType) -> bool {
74    matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == datetime_struct_fields())
75}
76
77/// Detects if an Arrow DataType is the canonical Time struct.
78pub fn is_time_struct(arrow_dt: &ArrowDataType) -> bool {
79    matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == time_struct_fields())
80}
81
82/// The canonical Arrow struct fields for a `DataType::SparseVector` column:
83/// `Struct { indices: List<UInt32>, values: List<Float32> }`. Two parallel
84/// variable-length lists in one struct. Both lists are non-null — an empty
85/// sparse vector stores as two empty lists, never null. The write and read
86/// sides both route through this one definition so they cannot drift (the same
87/// lockstep discipline as the temporal structs above).
88pub 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
103/// Detects if an Arrow DataType is the canonical SparseVector struct.
104pub fn is_sparse_vector_struct(arrow_dt: &ArrowDataType) -> bool {
105    matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == sparse_vector_struct_fields())
106}
107
108/// Field metadata marking an Arrow `LargeBinary` field as a raw `DataType::Bytes`
109/// value rather than a tagged CypherValue/Duration blob.
110///
111/// Stamped on the child field of `List(Bytes)` / `Map(_, Bytes)` container types so
112/// the read path decodes each element verbatim instead of through the tagged codec
113/// (which would read `byte[0]` as a type tag). CV-encoded containers carry no such
114/// marker and keep the codec path. See the read-side honoring in `arrow_convert`.
115pub 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    /// Returns the canonical variant name for this CRDT type.
134    ///
135    /// The returned strings must stay in sync with `uni_crdt::Crdt::type_name`,
136    /// so a written CRDT value can be validated against its schema-declared
137    /// variant (see uni-store's write-time CRDT enforcement).
138    ///
139    /// # Examples
140    /// ```
141    /// use uni_common::core::schema::CrdtType;
142    /// assert_eq!(CrdtType::GCounter.type_name(), "GCounter");
143    /// ```
144    #[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,  // WGS84
162    Cartesian2D, // x, y
163    Cartesian3D, // x, y, z
164}
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    /// Learned-sparse vector (SPLADE / BGE-M3). `dimensions` is the term-space
187    /// cardinality (max term id + 1) used for validation and index config.
188    SparseVector {
189        dimensions: usize,
190    },
191    /// Binary/bit vector for Hamming/Jaccard similarity. `dimensions` is the
192    /// number of `u8` lanes (each lane holds 8 bits), stored as
193    /// `FixedSizeList<UInt8>`. Exact/brute-force only — Lance ANN over binary
194    /// metrics is not wired, so a `BinaryVector` column builds no ANN index.
195    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    // Alias for compatibility/convenience if needed, but preferable to use exact types.
206    #[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, // Lance doesn't support Interval(MonthDayNano); use CypherValue codec
226            DataType::CypherValue => ArrowDataType::LargeBinary, // MessagePack-tagged binary encoding
227            DataType::Bytes => ArrowDataType::LargeBinary, // raw byte buffer (no codec wrapping)
228            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, // Store CRDT as binary MessagePack
257            DataType::List(inner) => {
258                // A raw `Bytes` element maps to Arrow `LargeBinary`, indistinguishable
259                // from a CV-encoded element by type alone; mark the child field so the
260                // read path decodes it verbatim rather than through the tagged codec.
261                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                // The value child's Arrow storage MUST agree with `build_map_column` in
271                // uni-store: typed scalars use their own Arrow type; `Bytes` is a
272                // raw-bytes-marked `LargeBinary`; every other (nested/non-scalar) value type
273                // is CypherValue-encoded into an UNMARKED `LargeBinary` (decoded back through
274                // the tagged codec on read). Gated by `map_value_is_typed` so the two sites
275                // can't drift.
276                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    /// Whether a `Map(_, self)` VALUE is stored as a typed Arrow child (this set) versus a
299    /// CypherValue-encoded `LargeBinary` fallback child (everything else, e.g. `Vector`,
300    /// `List`, `Map`, temporal). This MUST stay in lockstep with the explicit value-type
301    /// arms of `build_map_column` in uni-store (the `_` arm there is the CV fallback).
302    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    /// Returns `true` if `value` is directly storable in this column type without loss.
316    ///
317    /// This is the schema-level type guard used by the write path. `Value::Null` is
318    /// always accepted — column nullability is enforced separately by the `nullable`
319    /// flag, not here. `CypherValue`, `Crdt`, and `Point` columns accept any value.
320    /// For every other declared type, only the `Value` variants that the storage layer
321    /// persists *without silently nulling* are accepted (see the per-type converters in
322    /// `uni-store`'s `arrow_convert`), plus the intentional lossless widenings
323    /// `Int`→`Float`, `Int`→`Int32`, and `Temporal`→`Timestamp`.
324    ///
325    /// A `Value::String` destined for a `Date`/`Time`/`DateTime`/`Duration` column is
326    /// intentionally *not* accepted here: the write path first coerces such strings into
327    /// the proper `Temporal` value (matching the Cypher temporal constructors), then the
328    /// coerced value passes this check. This keeps `accepts` a pure, allocation-free
329    /// predicate.
330    ///
331    /// # Examples
332    /// ```
333    /// use uni_common::core::schema::DataType;
334    /// use uni_common::Value;
335    ///
336    /// assert!(DataType::Float64.accepts(&Value::Int(3))); // Int widens to Float
337    /// assert!(DataType::Bool.accepts(&Value::Null)); // Null always accepted
338    /// assert!(!DataType::DateTime.accepts(&Value::String("2026-01-01T00:00:00Z".into())));
339    /// ```
340    pub fn accepts(&self, value: &crate::value::Value) -> bool {
341        use crate::value::{TemporalValue, Value};
342
343        // Null is universally accepted; nullability is a separate concern.
344        if matches!(value, Value::Null) {
345            return true;
346        }
347
348        match self {
349            // Opaque / dynamically-typed columns accept any value.
350            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            // Int widens to Float losslessly for the ranges we care about.
355            DataType::Float32 | DataType::Float64 => {
356                matches!(value, Value::Int(_) | Value::Float(_))
357            }
358            DataType::Bool => matches!(value, Value::Bool(_)),
359
360            // Non-struct timestamp column: storage parses strings and accepts ints,
361            // so both are lossless here (unlike the DateTime struct column below).
362            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            // FixedSizeBinary(24) converter accepts the Btic temporal, raw strings, and lists.
392            DataType::Btic => matches!(
393                value,
394                Value::String(_) | Value::List(_) | Value::Temporal(TemporalValue::Btic { .. })
395            ),
396            // Shape-only: declared dimensions are enforced by `check_vector_dims`,
397            // which the write paths call alongside this predicate (issue #137).
398            DataType::Vector { .. } => matches!(value, Value::Vector(_) | Value::List(_)),
399            // `Value::Map` is the degraded form a `SparseVector` collapses into
400            // when round-tripped through `#[serde(untagged)]` persistence (e.g.
401            // the WAL's serde_json mutation log); accept it like `Vector` accepts
402            // `List`. The column builder re-extracts `{indices, values}`.
403            DataType::SparseVector { .. } => {
404                matches!(value, Value::SparseVector { .. } | Value::Map(_))
405            }
406            // Shape-only: the declared lane count is enforced by
407            // `check_vector_dims`. A `Value::List` of byte-valued integers is the
408            // literal input form, coerced to `BinaryVector` by the write path.
409            DataType::BinaryVector { .. } => {
410                matches!(value, Value::BinaryVector(_) | Value::List(_))
411            }
412            // Shape-only: for `List(Vector)` multi-vector columns, per-token
413            // dimensions are enforced by `check_vector_dims` (issue #137).
414            DataType::List(_) => matches!(value, Value::List(_)),
415            DataType::Map(_, _) => matches!(value, Value::Map(_)),
416        }
417    }
418
419    /// Checks a value's dimensions against a declared vector column type.
420    ///
421    /// The dimension-aware companion to [`DataType::accepts`] (which is shape-only):
422    /// for `Vector { dimensions }` columns the value must be a `Value::Vector` or an
423    /// all-numeric `Value::List` of exactly `dimensions` elements; for
424    /// `List(Vector { dimensions })` multi-vector columns every token must satisfy the
425    /// same rule (an empty token list is a legal empty multi-vector). `Value::Null` is
426    /// always accepted — nullability is enforced separately — and every non-vector
427    /// `DataType` returns `Ok(())`, so callers may invoke this unconditionally.
428    ///
429    /// Guards against the silent data loss of issue #137, where wrong-length vectors
430    /// were accepted at write time and nulled at flush by the Arrow converters.
431    ///
432    /// # Errors
433    /// Returns a [`VectorDimError`] describing the first offending value: a length
434    /// mismatch, a non-numeric list element, a non-vector value in a vector column,
435    /// or their per-token counterparts for multi-vector columns.
436    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/// Why a value cannot be stored in a declared `VECTOR(dim)` or multi-vector column.
467///
468/// Produced by [`DataType::check_vector_dims`] and [`check_dense_vector_value`].
469/// Messages carry the declared and actual lengths so write-path errors are
470/// actionable; callers prefix the property name and declared type.
471#[derive(Debug, Clone, PartialEq, Eq)]
472pub enum VectorDimError {
473    /// The vector has the wrong number of elements.
474    WrongLength {
475        /// Declared column dimensions.
476        expected: usize,
477        /// Actual element count of the offending value.
478        actual: usize,
479    },
480    /// A list element is not `Int` or `Float`.
481    NonNumericElement {
482        /// Zero-based index of the offending element.
483        index: usize,
484    },
485    /// The value is not a vector or list at all.
486    NotAVector {
487        /// Variant name of the offending value.
488        actual: &'static str,
489    },
490    /// A multi-vector token has the wrong number of elements.
491    TokenWrongLength {
492        /// Zero-based token index within the multi-vector.
493        token: usize,
494        /// Declared per-token dimensions.
495        expected: usize,
496        /// Actual element count of the offending token.
497        actual: usize,
498    },
499    /// A multi-vector token contains a non-numeric element.
500    TokenNonNumericElement {
501        /// Zero-based token index within the multi-vector.
502        token: usize,
503        /// Zero-based index of the offending element within the token.
504        index: usize,
505    },
506    /// A multi-vector token is not a vector or list.
507    TokenNotAVector {
508        /// Zero-based token index within the multi-vector.
509        token: usize,
510        /// Variant name of the offending token.
511        actual: &'static str,
512    },
513    /// The value for a multi-vector column is not a list of tokens.
514    NotATokenList {
515        /// Variant name of the offending value.
516        actual: &'static str,
517    },
518}
519
520impl VectorDimError {
521    /// Maps a dense-kernel error to its per-token counterpart for multi-vector columns.
522    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
573/// Checks one dense vector value against a declared dimension count.
574///
575/// The per-token kernel behind [`DataType::check_vector_dims`], exposed so the
576/// storage-layer converters can validate individual multi-vector tokens with the
577/// same semantics: `Value::Null` is accepted (a legal null row), a `Value::Vector`
578/// must have exactly `dimensions` elements, and a `Value::List` must additionally
579/// be all-numeric (`Int` or `Float`).
580///
581/// # Errors
582/// Returns [`VectorDimError::WrongLength`] on a length mismatch (an empty list
583/// reports `actual: 0`), [`VectorDimError::NonNumericElement`] for a `String`,
584/// `Null`, or other non-numeric list element, and [`VectorDimError::NotAVector`]
585/// for any other value variant.
586pub 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
622/// Validates a value against a declared `BinaryVector(dimensions)` column.
623///
624/// The binary-vector companion to [`check_dense_vector_value`]: `dimensions` is
625/// the number of `u8` lanes. Accepts a [`Value::BinaryVector`](crate::value::Value::BinaryVector)
626/// of exactly that many bytes, or the literal input form — a
627/// [`Value::List`](crate::value::Value::List) of exactly that many integers each
628/// in `[0, 255]` (coerced to bytes by the write path).
629/// [`Value::Null`](crate::value::Value::Null) passes.
630///
631/// # Errors
632/// Returns [`VectorDimError::WrongLength`] on a lane-count mismatch,
633/// [`VectorDimError::NonNumericElement`] if a list element is not an integer in
634/// `[0, 255]`, or [`VectorDimError::NotAVector`] for any other value.
635pub 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
674/// Returns a short variant name for a `Value`, used in dimension-mismatch messages.
675fn 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, // SchemaVersion
715    #[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, // LabelId
726    #[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    /// See [`crate::core::edge_type::EdgeTypeId`] for bit-layout details.
737    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    /// Composite node key: the property tuple must be unique AND every listed
759    /// property must be present (non-null). Equivalent to `Unique` over the tuple
760    /// plus `Exists` on each member, enforced together at write time.
761    NodeKey {
762        properties: Vec<String>,
763    },
764}
765
766impl ConstraintType {
767    /// The property tuple whose combination must be unique.
768    ///
769    /// Returns `Some` for the uniqueness-enforcing kinds — `Unique` and `NodeKey`
770    /// (a node key is a unique tuple plus NOT-NULL) — and `None` otherwise. Lets
771    /// the write path share one key-collection/probe path across both kinds.
772    #[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/// Bidirectional registry for dynamically-assigned schemaless edge type IDs.
799///
800/// Edge types not defined in the schema are assigned IDs at runtime with
801/// bit 31 set (see [`crate::core::edge_type`]). This registry maintains
802/// the name-to-ID and ID-to-name mappings for those types.
803#[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 to assign (0 is reserved for invalid).
808    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    /// Returns the schemaless ID for `type_name`, assigning a new one if needed.
821    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    /// Looks up the edge type name for a schemaless ID.
836    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    /// Returns `true` if `type_name` has already been assigned a schemaless ID.
841    pub fn contains(&self, type_name: &str) -> bool {
842        self.name_to_id.contains_key(type_name)
843    }
844
845    /// Looks up the schemaless ID for `type_name` (exact match, read-only).
846    pub fn id_by_name(&self, type_name: &str) -> Option<u32> {
847        self.name_to_id.get(type_name).copied()
848    }
849
850    /// Looks up the edge type ID for `type_name` with case-insensitive matching.
851    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    /// Returns all registered schemaless type IDs.
859    pub fn all_type_ids(&self) -> Vec<u32> {
860        self.id_to_name.keys().copied().collect()
861    }
862
863    /// Returns true if the registry has any schemaless types.
864    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
875/// First virtual (catalog-resolved) label ID. Label IDs in
876/// `VIRTUAL_LABEL_ID_START..VIRTUAL_LABEL_ID_SENTINEL` are owned by
877/// plugin-registered `CatalogProvider`s and allocated lazily by the
878/// planner via `PluginRegistry::register_virtual_label`. Native label
879/// allocation (`SchemaManager::add_label`) refuses IDs in this range.
880pub const VIRTUAL_LABEL_ID_START: u16 = 0xFF00;
881/// Sentinel "no label" marker, kept distinct from any allocatable ID.
882pub const VIRTUAL_LABEL_ID_SENTINEL: u16 = 0xFFFF;
883
884/// Maximum byte length of a label or edge-type name. (L6)
885///
886/// Generous; the cap is hygiene — the name lands in on-disk dataset paths
887/// and Lance branch names — not a storage limit.
888const MAX_SCHEMA_NAME_LEN: usize = 255;
889
890/// Returns `true` if `id` is in the virtual (catalog-resolved) range.
891#[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    /// Registry for schemaless edge types (dynamically assigned IDs)
907    #[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    /// Bumps `schema_version` to invalidate cached query plans.
927    ///
928    /// Called at the end of every DDL mutation that changes the schema's
929    /// shape (labels, edge types, properties, indexes, constraints). The
930    /// plan-cache eviction guard keys on `schema_version`, so a stale plan
931    /// built against an older shape is discarded once this advances. Uses
932    /// wrapping arithmetic: the value is a coarse change token, not a count,
933    /// so wraparound only risks a missed eviction after 2^32 DDL operations.
934    fn bump_version(&mut self) {
935        self.schema_version = self.schema_version.wrapping_add(1);
936    }
937
938    /// Returns the label name for a given label ID.
939    ///
940    /// Performs a linear scan over all labels. This is efficient because
941    /// the number of labels in a schema is typically small.
942    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    /// Returns the label ID for a given label name.
950    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    /// Returns the edge type name for a given type ID.
955    ///
956    /// Performs a linear scan over all edge types. This is efficient because
957    /// the number of edge types in a schema is typically small.
958    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    /// Returns the edge type ID for a given type name.
966    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    /// Returns the vector index configuration for a given label and property.
971    ///
972    /// Performs a linear scan over all indexes. This is efficient because
973    /// the number of indexes in a schema is typically small.
974    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    /// Returns the scored sparse-vector index configuration for a label/property.
992    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    /// Returns the full-text index configuration for a given label and property.
1010    ///
1011    /// A full-text index covers one or more properties. This returns the config
1012    /// if the specified property is among the indexed properties.
1013    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    /// Get label metadata with case-insensitive lookup.
1031    ///
1032    /// This allows queries to match labels regardless of case, providing
1033    /// better user experience when label names vary in casing.
1034    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    /// Get the schema-canonical spelling of a label, matched case-insensitively.
1042    ///
1043    /// Returns the stored label name whose spelling differs only in case from
1044    /// `name`, or `None` if no such label is registered. Callers use this to
1045    /// normalize a user-supplied label to the canonical form the storage tier
1046    /// keys on, so case variants resolve to the same vertex table.
1047    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    /// Get label ID with case-insensitive lookup.
1055    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    /// Get edge type metadata with case-insensitive lookup.
1061    ///
1062    /// This allows queries to match edge types regardless of case, providing
1063    /// better user experience when type names vary in casing.
1064    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    /// Get edge type ID with case-insensitive lookup (schema-defined types only).
1072    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    /// Get edge type ID with case-insensitive lookup, checking both
1078    /// schema-defined and schemaless registries.
1079    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    /// Returns the edge type ID for `type_name`, checking the schema first
1088    /// and falling back to the schemaless registry (assigning a new ID if needed).
1089    ///
1090    /// Requires `&mut self` because it may assign a new schemaless ID.
1091    /// Use [`edge_type_id_by_name`](Self::edge_type_id_by_name) for read-only schema lookups.
1092    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        // Reaching here means the type is brand-new to *both* the schema map and
1097        // the schemaless registry (the early return above mirrors exactly what
1098        // `edge_type_id_unified` checks). Minting a new schemaless edge type
1099        // changes the result of `all_edge_type_ids()`, which untyped traversals
1100        // bake into cached plans keyed on `schema_version`. Bump the version so
1101        // those stale plans are evicted — otherwise a `MATCH ()-[r]->()` plan
1102        // built before this type existed silently drops edges of the new type.
1103        let id = self.schemaless_registry.get_or_assign_id(type_name);
1104        self.bump_version();
1105        id
1106    }
1107
1108    /// Read-only unified exact lookup: schema-defined edge type id, falling
1109    /// back to an already-assigned schemaless id.
1110    ///
1111    /// Mirrors exactly the checks [`Self::get_or_assign_edge_type_id`]
1112    /// performs before assigning, so a `Some` here means the assigning path
1113    /// would be a no-op — the basis for `SchemaManager`'s read-lock fast path.
1114    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    /// Returns the edge type name for `type_id`, checking both the schema
1120    /// and schemaless registries. Returns an owned `String` because the
1121    /// name may come from either registry.
1122    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    /// Returns all edge type IDs, including both schema-defined and schemaless types.
1133    /// Used when MATCH queries don't specify an edge type and need to scan all edges.
1134    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                // Self-heal catalogs that grew super-linearly under the
1176                // pre-fix `add_index` (issue rustic-ai/uni-db#63). Collapse
1177                // duplicate index entries by name, keeping the *last*
1178                // occurrence — matches the upsert semantics in `add_index`
1179                // and preserves whatever metadata the most recent rebuild
1180                // wrote. The dedup persists on the next mutation that
1181                // calls `save()`.
1182                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    /// Normalize function names in an expression to uppercase for case-insensitive matching.
1242    /// Examples: "lower(email)" -> "LOWER(email)", "trim(name)" -> "TRIM(name)"
1243    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                // Collect identifier
1250                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 followed by '(', it's a function call - uppercase it
1262                if chars.peek() == Some(&'(') {
1263                    result.push_str(&ident.to_uppercase());
1264                } else {
1265                    result.push_str(&ident); // Keep property names as-is
1266                }
1267            } else {
1268                result.push(ch);
1269            }
1270        }
1271
1272        result
1273    }
1274
1275    /// Generate a consistent internal column name for an expression index.
1276    /// Uses a hash suffix to ensure uniqueness for different expressions that
1277    /// might sanitize to the same string (e.g., "a+b" and "a-b" both become "a_b").
1278    ///
1279    /// IMPORTANT: Uses FNV-1a hash which is stable across Rust versions and platforms.
1280    /// DefaultHasher is not guaranteed to be stable and could break persistent data
1281    /// if the hash changes after a compiler upgrade.
1282    pub fn generated_column_name(expr: &str) -> String {
1283        // Normalize function names to uppercase for case-insensitive matching
1284        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        // FNV-1a 64-bit hash - stable across Rust versions and platforms
1292        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    /// Build a fork-scoped manager whose schema is `primary ⊕ overlay`.
1313    ///
1314    /// Used by `UniInner::at_fork` to give a forked session a schema view
1315    /// that includes any labels/edge-types/properties the fork has
1316    /// introduced on top of primary. The returned manager owns its own
1317    /// in-memory `Arc<Schema>` — mutations to it never reach primary's
1318    /// schema file. The returned manager is *not* intended for `.save()`;
1319    /// fork-overlay persistence is owned by the registry layer
1320    /// (`catalog/fork_schemas/{fork_id}.json`).
1321    ///
1322    /// In Phase 1 the delta is always empty, so the merge is a clone.
1323    /// Phase 2 starts populating it when on-the-fly label creation lands.
1324    #[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        // Ensure we stay in schema'd ID space (bit 31 = 0)
1381        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    /// Validate a label or edge-type name at definition time. (L6)
1389    ///
1390    /// Names flow into on-disk dataset paths (`vertices_{name}.lance`) and
1391    /// Lance branch names (`fork_{id}_{…}`); a name with a path separator,
1392    /// whitespace, or a control character corrupts those paths and breaks
1393    /// fork creation. Such names were never actually usable, so they are
1394    /// rejected up front rather than failing later. `.` is allowed
1395    /// (path-safe and common in qualified names).
1396    ///
1397    /// Public so the fork-create path can apply the same rule as a backstop
1398    /// over names that entered the schema through an infallible interning
1399    /// path (e.g. schemaless `get_or_assign_edge_type_id`).
1400    ///
1401    /// # Errors
1402    /// Returns an error if `name` is empty/all-whitespace, exceeds
1403    /// `MAX_SCHEMA_NAME_LEN` bytes, or contains a control, whitespace,
1404    /// `/`, or `\` character.
1405    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        // Stay in the schema-defined sub-range (bit 31 = 0, and below the
1484        // virtual reservation `VIRTUAL_EDGE_TYPE_ID_START`) — same bound as
1485        // `add_edge_type`, so the two entry points cannot disagree on the
1486        // legal ceiling.
1487        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    /// Delegates to [`Schema::get_or_assign_edge_type_id`].
1510    ///
1511    /// Read-lock fast path: the type name is almost always already known
1512    /// (it is constant per statement but resolved per row by the CREATE
1513    /// executor), and the slow path's write lock + `Arc::make_mut` deep-clones
1514    /// the whole `Schema` whenever the Arc is shared — which under SSI it
1515    /// always is. Double-checked: on a miss, `Schema::get_or_assign_edge_type_id`
1516    /// re-checks under the write lock, so two racing assigners converge on one id.
1517    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    /// Delegates to [`Schema::edge_type_name_by_id_unified`].
1532    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        // Bump after stamping `added_in` with the pre-bump `version`.
1585        schema.bump_version();
1586        Ok(())
1587    }
1588
1589    /// Declares a property, idempotent when an identical declaration already exists.
1590    ///
1591    /// The schema-builder counterpart to [`Self::add_property_with_desc`] (which
1592    /// hard-errors on *any* re-add, as DDL `ALTER` semantics require). Re-applying a
1593    /// schema is common — every `apply()` re-declares — so an existing property with
1594    /// the same `data_type` and `nullable` is a no-op (`Ok(false)`; a differing
1595    /// `description` is ignored as docs-only). A differing type or nullability is a
1596    /// hard conflict: silently swallowing it let `VECTOR(4)` be "re-declared" as
1597    /// `VECTOR(8)` while the column stayed 4-dimensional (issue #137).
1598    ///
1599    /// Returns `true` if this call newly inserted the property.
1600    ///
1601    /// # Errors
1602    /// Returns an error when the property exists with a different type or nullability
1603    /// (the message deliberately does not contain "already exists", which callers
1604    /// historically string-matched to ignore benign re-adds), or when the property
1605    /// name is invalid.
1606    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); // identical re-declaration (idempotent)
1626            }
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        // Bump after stamping `added_in` with the pre-bump `version`.
1652        schema.bump_version();
1653        Ok(true)
1654    }
1655
1656    /// Register an INTERNAL property (underscore-prefixed name allowed) that is
1657    /// materialised by the storage layer, not written by the user — e.g. the MUVERA
1658    /// `__fde_*` derived column. Bypasses the user-facing underscore-prefix rule but
1659    /// still rejects storage-layer name collisions. Idempotent: a no-op if the property
1660    /// already exists with the same type (so re-creating an index is safe).
1661    ///
1662    /// Returns `true` if this call newly inserted the property, `false` if it already
1663    /// existed (idempotent). The check-and-insert is atomic under the schema write lock,
1664    /// so for concurrent callers exactly one observes `true` — letting callers gate
1665    /// expensive one-time work (e.g. the MUVERA backfill) on the winner.
1666    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); // already present (idempotent re-registration)
1685            }
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        // System-generated `_gen_*` columns bypass the underscore-prefix rule
1716        // but must still avoid storage-layer column-name collisions.
1717        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        // Bump after stamping `added_in` with the pre-bump `version`.
1742        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    /// Register an index definition on the schema, **upsert by name**.
1788    ///
1789    /// If an index with the same `IndexDefinition::name()` already exists, it
1790    /// is replaced in place; otherwise the def is appended. Idempotent under
1791    /// repeat invocation, which makes `SchemaBuilder::apply()` re-applicable
1792    /// without bloating `schema.indexes` and lets the rebuild epilogue inside
1793    /// every `IndexManager::create_*_index` re-record metadata updates without
1794    /// duplicating entries (issue rustic-ai/uni-db#63).
1795    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    /// Updates the lifecycle metadata for an index by name.
1817    ///
1818    /// The closure receives a mutable reference to the index's `IndexMetadata`,
1819    /// allowing callers to update status, timestamps, etc.
1820    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 the new name like declare_property/add_property do — otherwise
1897        // a rename bypasses the reserved-storage-column guard and the leading-
1898        // underscore rule, letting a user property collide with an internal Arrow
1899        // column (e.g. `_vid`, `src_vid`, `overflow_json`).
1900        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            // Rollback removal? Or just error.
1915            props.insert(old_name.to_string(), meta); // Restore
1916            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            // Do not remove properties; they are implicitly tombstoned by the label
1929            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            // Do not remove properties; they are implicitly tombstoned by the edge type
1944            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
1954/// Validate identifier names to prevent injection and ensure compatibility.
1955pub fn validate_identifier(name: &str) -> Result<()> {
1956    // Length check
1957    if name.is_empty() || name.len() > 64 {
1958        return Err(anyhow!("Identifier '{}' must be 1-64 characters", name));
1959    }
1960
1961    // First character must be letter or underscore
1962    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    // Remaining characters: alphanumeric or underscore
1971    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    // Reserved words
1979    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
1990/// Reject user-declared property names that collide with internal Arrow column
1991/// names used by the storage layer.
1992///
1993/// Without this, declaring a property named e.g. `ext_id` produces an Arrow
1994/// schema with two `ext_id` fields at flush time, which Lance rejects with
1995/// "Duplicate field name" — silently losing all in-session writes on shutdown.
1996pub 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
2006/// Reject names that collide with storage-layer Arrow column names.
2007///
2008/// Used both by `validate_property_name` (user-facing path) and directly by
2009/// `add_generated_property` (system-generated `_gen_*` path) — the latter
2010/// needs to bypass the underscore-prefix rule but must still reject the
2011/// fixed-name collisions below.
2012fn validate_reserved_property_name(name: &str) -> Result<()> {
2013    // Unprefixed names that get appended alongside user properties in the
2014    // per-label vertex (`storage/vertex.rs`), per-edge-type edge
2015    // (`storage/edge.rs`), or per-edge-type delta (`storage/delta.rs`)
2016    // Arrow schemas — declaring one of these as a user property produces a
2017    // duplicate Arrow field and a Lance "Duplicate field name" error at
2018    // flush time. Fixed-schema-only columns (`type`, `props_json`,
2019    // `labels` in the main tables) are NOT listed: those tables don't
2020    // append user properties, so no collision can occur.
2021    const RESERVED_PROPS: &[&str] = &[
2022        "ext_id",
2023        "overflow_json",
2024        "eid",
2025        "src_vid",
2026        "dst_vid",
2027        "op",
2028        // Internal planner sentinel: a column-name marker used by
2029        // `mark_set_item_variables` (uni-query::query::planner) to request
2030        // narrow structural projection without full-schema expansion.
2031        // Reserved here defensively so an internal `add_generated_property`
2032        // path can't accidentally create a colliding user-facing column.
2033        // The user-facing `validate_property_name` already rejects this
2034        // via the underscore-prefix rule, so this is belt-and-suspenders.
2035        "__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        // Hamming = number of differing bits. 0x00 vs 0xFF = 8 bits; 0xA5 vs 0xA5
2056        // = 0; 0x0F vs 0x00 = 4 bits.
2057        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        // Jaccard = 1 − |A∩B|/|A∪B|. 0b1100 & 0b1010 = 0b1000 (1 bit);
2071        // 0b1100 | 0b1010 = 0b1110 (3 bits) → 1 − 1/3 = 2/3.
2072        let j = DistanceMetric::Jaccard.compute_distance_binary(&[0b1100], &[0b1010]);
2073        assert!((j - (2.0 / 3.0)).abs() < 1e-6, "got {j}");
2074        // Identical vectors → distance 0.
2075        assert_eq!(
2076            DistanceMetric::Jaccard.compute_distance_binary(&[0xFF], &[0xFF]),
2077            0.0
2078        );
2079        // Two all-zero vectors are defined as distance 0 (empty union).
2080        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        // Wrong lane count.
2109        assert!(
2110            ty.check_vector_dims(&Value::BinaryVector(vec![1, 2]))
2111                .is_err()
2112        );
2113        // List of byte-ints is the literal form.
2114        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        // Out-of-byte-range element.
2123        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        // Null is accepted by every type (nullability checked separately).
2142        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        // Exact-type matches.
2153        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        // Intentional lossless widenings remain allowed.
2159        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        // The #68 data-loss cases must be rejected (coercion handles strings separately).
2171        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        // Opaque columns accept anything.
2185        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        // Null is accepted everywhere (nullability enforced separately).
2195        assert!(vec3.check_vector_dims(&Value::Null).is_ok());
2196        assert!(multi2.check_vector_dims(&Value::Null).is_ok());
2197
2198        // Correct-dimension values pass; Int elements are numeric.
2199        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        // The #137 cases: wrong length, empty list, non-numeric element, wrong shape.
2214        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        // Multi-vector: empty token list is a legal empty multi-vector; each
2257        // token must match the declared per-token dimensions.
2258        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        // Non-vector declared types never object, so callers may check unconditionally.
2288        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        // Error rendering carries both lengths so write errors are actionable.
2305        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        // First declaration inserts.
2324        assert!(manager.declare_property("Doc", "embedding", vec4.clone(), true, None)?);
2325
2326        // Identical re-declaration is an idempotent no-op — the register-on-every-open
2327        // pattern; a differing description is docs-only and also ignored.
2328        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        // A dimension change is a conflict (#137 case c), and the message must not
2338        // contain "already exists" (historically string-matched and swallowed).
2339        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        // Nullability flips are conflicts too — they change NOT NULL enforcement.
2353        assert!(
2354            manager
2355                .declare_property("Doc", "embedding", vec4.clone(), false, None)
2356                .is_err()
2357        );
2358
2359        // The schema still holds the original declaration.
2360        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        // Labels
2375        let lid = manager.add_label("Person")?;
2376        assert_eq!(lid, 1);
2377        assert!(manager.add_label("Person").is_err());
2378
2379        // Properties
2380        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        // Edge types
2388        let tid = manager.add_edge_type("knows", vec!["Person".into()], vec!["Person".into()])?;
2389        assert_eq!(tid, 1);
2390
2391        manager.save().await?;
2392        // Check file exists
2393        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        // Unprefixed reserved names — these collide with internal Arrow
2419        // columns in storage tables and previously caused Lance
2420        // "Duplicate field name" errors at flush time.
2421        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        // Planner sentinel — reserved in RESERVED_PROPS (belt-and-suspenders
2432        // alongside the underscore-prefix rule). Confirms an internal
2433        // `add_generated_property` path cannot accidentally create a column
2434        // that collides with the SET-target structural-projection marker.
2435        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        // Leading-underscore pattern rule.
2444        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        // Names that merely contain a reserved substring should still be
2454        // accepted.
2455        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        // Same check applies to edge-type properties (single dispatch).
2460        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        // And to generated properties.
2468        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        // Simulate old JSON without metadata field
2515        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        // Verify initial status is Online
2572        let initial = manager.get_index("idx_test").unwrap();
2573        assert_eq!(initial.metadata().status, IndexStatus::Online);
2574
2575        // Update to Building
2576        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        // Non-existent index should error
2586        assert!(manager.update_index_metadata("nope", |_| {}).is_err());
2587
2588        Ok(())
2589    }
2590
2591    /// `add_internal_property` reports whether THIS call inserted the property: `true` on
2592    /// first insert, `false` on idempotent re-registration, `Err` on a type conflict. The
2593    /// MUVERA backfill gates on this (only the inserter backfills), so two concurrent
2594    /// creates of the same index can't both run the full-table rewrite (issue #107).
2595    #[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        // First registration: newly added.
2605        assert!(manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
2606        // Idempotent re-registration with the same type: NOT newly added.
2607        assert!(!manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
2608        // Same name, conflicting type: hard error (no silent divergence).
2609        assert!(
2610            manager
2611                .add_internal_property("Doc", "__fde_x", DataType::Vector { dimensions: 8 }, true)
2612                .is_err()
2613        );
2614        Ok(())
2615    }
2616
2617    /// `add_index` is upsert-by-name (issue rustic-ai/uni-db#63). Repeat
2618    /// invocations with the same `IndexDefinition::name()` must replace
2619    /// the entry in place rather than appending. Subsequent `add_index`
2620    /// calls also reflect metadata updates from the new definition.
2621    #[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        // Re-add the identical def — must remain a single entry.
2644        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        // Re-add with updated metadata — must replace in place, len unchanged.
2652        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        // A *different* name appends as a new entry.
2665        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    /// `load_from_store` self-heals catalogs that were bloated by the
2680    /// pre-fix `add_index` (kept the *last* def per name).
2681    #[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        // Seed disk with a hand-crafted bloated schema: 50 entries, all
2688        // sharing the same name. The last entry has distinct metadata so
2689        // we can assert "last writer wins" semantics.
2690        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        // Last-writer-wins: the kept entry is the final push (Online, 123).
2729        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        // Add a vector index with Stale status
2749        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        // Stale index should NOT be returned
2765        assert!(
2766            schema
2767                .vector_index_for_property("Document", "embedding")
2768                .is_none()
2769        );
2770
2771        // Set to Online — should now be returned
2772        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        // Phase 1 invariant: mutating the overlay manager must not bleed
2794        // into primary's schema.
2795        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        // Primary unchanged.
2839        assert!(!primary.schema().labels.contains_key("NewLabel"));
2840        Ok(())
2841    }
2842
2843    /// N threads racing `get_or_assign_edge_type_id` for the same new name
2844    /// must converge on a single id (the read-lock fast path double-checks
2845    /// under the write lock); a schema-defined type must win over the
2846    /// schemaless registry.
2847    #[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        // Fast path returns the same id afterwards.
2867        assert_eq!(manager.get_or_assign_edge_type_id("RACED"), ids[0]);
2868
2869        // Schema-defined type wins over the schemaless registry.
2870        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    /// Minting a brand-new schemaless edge type must bump `schema_version`
2877    /// (the plan cache keys on it; untyped traversals bake `all_edge_type_ids()`
2878    /// into the plan, so a stale plan would silently drop edges of the new
2879    /// type). Re-resolving an existing type must NOT bump. (review C5)
2880    #[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        // Re-resolving the same type is a no-op — no further bump.
2893        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        // A second distinct new type bumps again.
2902        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    /// L6: label/edge-type names with path separators, whitespace, or
2911    /// control chars are rejected at definition; benign names (incl. `.`)
2912    /// are accepted.
2913    #[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        // Over-length is rejected.
2928        let long = "x".repeat(MAX_SCHEMA_NAME_LEN + 1);
2929        assert!(SchemaManager::validate_schema_element_name("Label", &long).is_err());
2930    }
2931}