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
20#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
21#[non_exhaustive]
22pub enum SchemaElementState {
23    Active,
24    Hidden {
25        since: DateTime<Utc>,
26        last_active_snapshot: String, // SnapshotId
27    },
28    Tombstone {
29        since: DateTime<Utc>,
30    },
31}
32
33use arrow_schema::{DataType as ArrowDataType, Field, Fields, TimeUnit};
34
35/// Returns the canonical struct field definitions for DateTime encoding in Arrow.
36///
37/// DateTime is encoded as a 3-field struct to preserve timezone information:
38/// - `nanos_since_epoch`: i64 nanoseconds since Unix epoch (UTC)
39/// - `offset_seconds`: i32 seconds offset from UTC (e.g., +3600 for +01:00)
40/// - `timezone_name`: Optional IANA timezone name (e.g., "America/New_York")
41pub fn datetime_struct_fields() -> Fields {
42    Fields::from(vec![
43        Field::new(
44            "nanos_since_epoch",
45            ArrowDataType::Timestamp(TimeUnit::Nanosecond, None),
46            true,
47        ),
48        Field::new("offset_seconds", ArrowDataType::Int32, true),
49        Field::new("timezone_name", ArrowDataType::Utf8, true),
50    ])
51}
52
53/// Returns the canonical struct field definitions for Time encoding in Arrow.
54///
55/// Time is encoded as a 2-field struct to preserve timezone offset:
56/// - `nanos_since_midnight`: i64 nanoseconds since midnight (0-86,399,999,999,999)
57/// - `offset_seconds`: i32 seconds offset from UTC (e.g., +3600 for +01:00)
58pub fn time_struct_fields() -> Fields {
59    Fields::from(vec![
60        Field::new(
61            "nanos_since_midnight",
62            ArrowDataType::Time64(TimeUnit::Nanosecond),
63            true,
64        ),
65        Field::new("offset_seconds", ArrowDataType::Int32, true),
66    ])
67}
68
69/// Detects if an Arrow DataType is the canonical DateTime struct.
70pub fn is_datetime_struct(arrow_dt: &ArrowDataType) -> bool {
71    matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == datetime_struct_fields())
72}
73
74/// Detects if an Arrow DataType is the canonical Time struct.
75pub fn is_time_struct(arrow_dt: &ArrowDataType) -> bool {
76    matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == time_struct_fields())
77}
78
79/// The canonical Arrow struct fields for a `DataType::SparseVector` column:
80/// `Struct { indices: List<UInt32>, values: List<Float32> }`. Two parallel
81/// variable-length lists in one struct. Both lists are non-null — an empty
82/// sparse vector stores as two empty lists, never null. The write and read
83/// sides both route through this one definition so they cannot drift (the same
84/// lockstep discipline as the temporal structs above).
85pub fn sparse_vector_struct_fields() -> Fields {
86    Fields::from(vec![
87        Field::new(
88            "indices",
89            ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::UInt32, true))),
90            false,
91        ),
92        Field::new(
93            "values",
94            ArrowDataType::List(Arc::new(Field::new("item", ArrowDataType::Float32, true))),
95            false,
96        ),
97    ])
98}
99
100/// Detects if an Arrow DataType is the canonical SparseVector struct.
101pub fn is_sparse_vector_struct(arrow_dt: &ArrowDataType) -> bool {
102    matches!(arrow_dt, ArrowDataType::Struct(fields) if *fields == sparse_vector_struct_fields())
103}
104
105/// Field metadata marking an Arrow `LargeBinary` field as a raw `DataType::Bytes`
106/// value rather than a tagged CypherValue/Duration blob.
107///
108/// Stamped on the child field of `List(Bytes)` / `Map(_, Bytes)` container types so
109/// the read path decodes each element verbatim instead of through the tagged codec
110/// (which would read `byte[0]` as a type tag). CV-encoded containers carry no such
111/// marker and keep the codec path. See the read-side honoring in `arrow_convert`.
112pub fn raw_bytes_field_metadata() -> HashMap<String, String> {
113    HashMap::from([("uni_raw_bytes".to_string(), "true".to_string())])
114}
115
116#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
117#[non_exhaustive]
118pub enum CrdtType {
119    GCounter,
120    GSet,
121    ORSet,
122    LWWRegister,
123    LWWMap,
124    Rga,
125    VectorClock,
126    VCRegister,
127}
128
129impl CrdtType {
130    /// Returns the canonical variant name for this CRDT type.
131    ///
132    /// The returned strings must stay in sync with `uni_crdt::Crdt::type_name`,
133    /// so a written CRDT value can be validated against its schema-declared
134    /// variant (see uni-store's write-time CRDT enforcement).
135    ///
136    /// # Examples
137    /// ```
138    /// use uni_common::core::schema::CrdtType;
139    /// assert_eq!(CrdtType::GCounter.type_name(), "GCounter");
140    /// ```
141    #[must_use]
142    pub fn type_name(&self) -> &'static str {
143        match self {
144            CrdtType::GCounter => "GCounter",
145            CrdtType::GSet => "GSet",
146            CrdtType::ORSet => "ORSet",
147            CrdtType::LWWRegister => "LWWRegister",
148            CrdtType::LWWMap => "LWWMap",
149            CrdtType::Rga => "Rga",
150            CrdtType::VectorClock => "VectorClock",
151            CrdtType::VCRegister => "VCRegister",
152        }
153    }
154}
155
156#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq)]
157pub enum PointType {
158    Geographic,  // WGS84
159    Cartesian2D, // x, y
160    Cartesian3D, // x, y, z
161}
162
163#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
164#[non_exhaustive]
165pub enum DataType {
166    String,
167    Int32,
168    Int64,
169    Float32,
170    Float64,
171    Bool,
172    Timestamp,
173    Date,
174    Time,
175    DateTime,
176    Duration,
177    CypherValue,
178    Bytes,
179    Point(PointType),
180    Vector {
181        dimensions: usize,
182    },
183    /// Learned-sparse vector (SPLADE / BGE-M3). `dimensions` is the term-space
184    /// cardinality (max term id + 1) used for validation and index config.
185    SparseVector {
186        dimensions: usize,
187    },
188    /// Binary/bit vector for Hamming/Jaccard similarity. `dimensions` is the
189    /// number of `u8` lanes (each lane holds 8 bits), stored as
190    /// `FixedSizeList<UInt8>`. Exact/brute-force only — Lance ANN over binary
191    /// metrics is not wired, so a `BinaryVector` column builds no ANN index.
192    BinaryVector {
193        dimensions: usize,
194    },
195    Btic,
196    Crdt(CrdtType),
197    List(Box<DataType>),
198    Map(Box<DataType>, Box<DataType>),
199}
200
201impl DataType {
202    // Alias for compatibility/convenience if needed, but preferable to use exact types.
203    #[allow(non_upper_case_globals)]
204    pub const Float: DataType = DataType::Float64;
205    #[allow(non_upper_case_globals)]
206    pub const Int: DataType = DataType::Int64;
207
208    pub fn to_arrow(&self) -> ArrowDataType {
209        match self {
210            DataType::String => ArrowDataType::Utf8,
211            DataType::Int32 => ArrowDataType::Int32,
212            DataType::Int64 => ArrowDataType::Int64,
213            DataType::Float32 => ArrowDataType::Float32,
214            DataType::Float64 => ArrowDataType::Float64,
215            DataType::Bool => ArrowDataType::Boolean,
216            DataType::Timestamp => {
217                ArrowDataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
218            }
219            DataType::Date => ArrowDataType::Date32,
220            DataType::Time => ArrowDataType::Struct(time_struct_fields()),
221            DataType::DateTime => ArrowDataType::Struct(datetime_struct_fields()),
222            DataType::Duration => ArrowDataType::LargeBinary, // Lance doesn't support Interval(MonthDayNano); use CypherValue codec
223            DataType::CypherValue => ArrowDataType::LargeBinary, // MessagePack-tagged binary encoding
224            DataType::Bytes => ArrowDataType::LargeBinary, // raw byte buffer (no codec wrapping)
225            DataType::Point(pt) => match pt {
226                PointType::Geographic => ArrowDataType::Struct(Fields::from(vec![
227                    Field::new("latitude", ArrowDataType::Float64, false),
228                    Field::new("longitude", ArrowDataType::Float64, false),
229                    Field::new("crs", ArrowDataType::Utf8, false),
230                ])),
231                PointType::Cartesian2D => ArrowDataType::Struct(Fields::from(vec![
232                    Field::new("x", ArrowDataType::Float64, false),
233                    Field::new("y", ArrowDataType::Float64, false),
234                    Field::new("crs", ArrowDataType::Utf8, false),
235                ])),
236                PointType::Cartesian3D => ArrowDataType::Struct(Fields::from(vec![
237                    Field::new("x", ArrowDataType::Float64, false),
238                    Field::new("y", ArrowDataType::Float64, false),
239                    Field::new("z", ArrowDataType::Float64, false),
240                    Field::new("crs", ArrowDataType::Utf8, false),
241                ])),
242            },
243            DataType::Vector { dimensions } => ArrowDataType::FixedSizeList(
244                Arc::new(Field::new("item", ArrowDataType::Float32, true)),
245                *dimensions as i32,
246            ),
247            DataType::SparseVector { .. } => ArrowDataType::Struct(sparse_vector_struct_fields()),
248            DataType::BinaryVector { dimensions } => ArrowDataType::FixedSizeList(
249                Arc::new(Field::new("item", ArrowDataType::UInt8, true)),
250                *dimensions as i32,
251            ),
252            DataType::Btic => ArrowDataType::FixedSizeBinary(24),
253            DataType::Crdt(_) => ArrowDataType::Binary, // Store CRDT as binary MessagePack
254            DataType::List(inner) => {
255                // A raw `Bytes` element maps to Arrow `LargeBinary`, indistinguishable
256                // from a CV-encoded element by type alone; mark the child field so the
257                // read path decodes it verbatim rather than through the tagged codec.
258                let item = Field::new("item", inner.to_arrow(), true);
259                let item = if matches!(**inner, DataType::Bytes) {
260                    item.with_metadata(raw_bytes_field_metadata())
261                } else {
262                    item
263                };
264                ArrowDataType::List(Arc::new(item))
265            }
266            DataType::Map(key, value) => {
267                // The value child's Arrow storage MUST agree with `build_map_column` in
268                // uni-store: typed scalars use their own Arrow type; `Bytes` is a
269                // raw-bytes-marked `LargeBinary`; every other (nested/non-scalar) value type
270                // is CypherValue-encoded into an UNMARKED `LargeBinary` (decoded back through
271                // the tagged codec on read). Gated by `map_value_is_typed` so the two sites
272                // can't drift.
273                let value_field = if value.map_value_is_typed() {
274                    let f = Field::new("value", value.to_arrow(), true);
275                    if matches!(**value, DataType::Bytes) {
276                        f.with_metadata(raw_bytes_field_metadata())
277                    } else {
278                        f
279                    }
280                } else {
281                    Field::new("value", ArrowDataType::LargeBinary, true)
282                };
283                ArrowDataType::List(Arc::new(Field::new(
284                    "item",
285                    ArrowDataType::Struct(Fields::from(vec![
286                        Field::new("key", key.to_arrow(), false),
287                        value_field,
288                    ])),
289                    true,
290                )))
291            }
292        }
293    }
294
295    /// Whether a `Map(_, self)` VALUE is stored as a typed Arrow child (this set) versus a
296    /// CypherValue-encoded `LargeBinary` fallback child (everything else, e.g. `Vector`,
297    /// `List`, `Map`, temporal). This MUST stay in lockstep with the explicit value-type
298    /// arms of `build_map_column` in uni-store (the `_` arm there is the CV fallback).
299    pub fn map_value_is_typed(&self) -> bool {
300        matches!(
301            self,
302            DataType::String
303                | DataType::Int64
304                | DataType::Int32
305                | DataType::Float64
306                | DataType::Float32
307                | DataType::Bool
308                | DataType::Bytes
309        )
310    }
311
312    /// Returns `true` if `value` is directly storable in this column type without loss.
313    ///
314    /// This is the schema-level type guard used by the write path. `Value::Null` is
315    /// always accepted — column nullability is enforced separately by the `nullable`
316    /// flag, not here. `CypherValue`, `Crdt`, and `Point` columns accept any value.
317    /// For every other declared type, only the `Value` variants that the storage layer
318    /// persists *without silently nulling* are accepted (see the per-type converters in
319    /// `uni-store`'s `arrow_convert`), plus the intentional lossless widenings
320    /// `Int`→`Float`, `Int`→`Int32`, and `Temporal`→`Timestamp`.
321    ///
322    /// A `Value::String` destined for a `Date`/`Time`/`DateTime`/`Duration` column is
323    /// intentionally *not* accepted here: the write path first coerces such strings into
324    /// the proper `Temporal` value (matching the Cypher temporal constructors), then the
325    /// coerced value passes this check. This keeps `accepts` a pure, allocation-free
326    /// predicate.
327    ///
328    /// # Examples
329    /// ```
330    /// use uni_common::core::schema::DataType;
331    /// use uni_common::Value;
332    ///
333    /// assert!(DataType::Float64.accepts(&Value::Int(3))); // Int widens to Float
334    /// assert!(DataType::Bool.accepts(&Value::Null)); // Null always accepted
335    /// assert!(!DataType::DateTime.accepts(&Value::String("2026-01-01T00:00:00Z".into())));
336    /// ```
337    pub fn accepts(&self, value: &crate::value::Value) -> bool {
338        use crate::value::{TemporalValue, Value};
339
340        // Null is universally accepted; nullability is a separate concern.
341        if matches!(value, Value::Null) {
342            return true;
343        }
344
345        match self {
346            // Opaque / dynamically-typed columns accept any value.
347            DataType::CypherValue | DataType::Crdt(_) | DataType::Point(_) => true,
348
349            DataType::String => matches!(value, Value::String(_)),
350            DataType::Int32 | DataType::Int64 => matches!(value, Value::Int(_)),
351            // Int widens to Float losslessly for the ranges we care about.
352            DataType::Float32 | DataType::Float64 => {
353                matches!(value, Value::Int(_) | Value::Float(_))
354            }
355            DataType::Bool => matches!(value, Value::Bool(_)),
356
357            // Non-struct timestamp column: storage parses strings and accepts ints,
358            // so both are lossless here (unlike the DateTime struct column below).
359            DataType::Timestamp => matches!(
360                value,
361                Value::String(_)
362                    | Value::Int(_)
363                    | Value::Temporal(
364                        TemporalValue::DateTime { .. } | TemporalValue::LocalDateTime { .. }
365                    )
366            ),
367            DataType::DateTime => matches!(
368                value,
369                Value::Temporal(
370                    TemporalValue::DateTime { .. } | TemporalValue::LocalDateTime { .. }
371                )
372            ),
373            DataType::Date => {
374                matches!(
375                    value,
376                    Value::Int(_) | Value::Temporal(TemporalValue::Date { .. })
377                )
378            }
379            DataType::Time => matches!(
380                value,
381                Value::Int(_)
382                    | Value::Temporal(TemporalValue::Time { .. } | TemporalValue::LocalTime { .. })
383            ),
384            DataType::Duration => {
385                matches!(value, Value::Temporal(TemporalValue::Duration { .. }))
386            }
387            DataType::Bytes => matches!(value, Value::Bytes(_)),
388            // FixedSizeBinary(24) converter accepts the Btic temporal, raw strings, and lists.
389            DataType::Btic => matches!(
390                value,
391                Value::String(_) | Value::List(_) | Value::Temporal(TemporalValue::Btic { .. })
392            ),
393            // Shape-only: declared dimensions are enforced by `check_vector_dims`,
394            // which the write paths call alongside this predicate (issue #137).
395            DataType::Vector { .. } => matches!(value, Value::Vector(_) | Value::List(_)),
396            // `Value::Map` is the degraded form a `SparseVector` collapses into
397            // when round-tripped through `#[serde(untagged)]` persistence (e.g.
398            // the WAL's serde_json mutation log); accept it like `Vector` accepts
399            // `List`. The column builder re-extracts `{indices, values}`.
400            DataType::SparseVector { .. } => {
401                matches!(value, Value::SparseVector { .. } | Value::Map(_))
402            }
403            // Shape-only: the declared lane count is enforced by
404            // `check_vector_dims`. A `Value::List` of byte-valued integers is the
405            // literal input form, coerced to `BinaryVector` by the write path.
406            DataType::BinaryVector { .. } => {
407                matches!(value, Value::BinaryVector(_) | Value::List(_))
408            }
409            // Shape-only: for `List(Vector)` multi-vector columns, per-token
410            // dimensions are enforced by `check_vector_dims` (issue #137).
411            DataType::List(_) => matches!(value, Value::List(_)),
412            DataType::Map(_, _) => matches!(value, Value::Map(_)),
413        }
414    }
415
416    /// Checks a value's dimensions against a declared vector column type.
417    ///
418    /// The dimension-aware companion to [`DataType::accepts`] (which is shape-only):
419    /// for `Vector { dimensions }` columns the value must be a `Value::Vector` or an
420    /// all-numeric `Value::List` of exactly `dimensions` elements; for
421    /// `List(Vector { dimensions })` multi-vector columns every token must satisfy the
422    /// same rule (an empty token list is a legal empty multi-vector). `Value::Null` is
423    /// always accepted — nullability is enforced separately — and every non-vector
424    /// `DataType` returns `Ok(())`, so callers may invoke this unconditionally.
425    ///
426    /// Guards against the silent data loss of issue #137, where wrong-length vectors
427    /// were accepted at write time and nulled at flush by the Arrow converters.
428    ///
429    /// # Errors
430    /// Returns a [`VectorDimError`] describing the first offending value: a length
431    /// mismatch, a non-numeric list element, a non-vector value in a vector column,
432    /// or their per-token counterparts for multi-vector columns.
433    pub fn check_vector_dims(&self, value: &crate::value::Value) -> Result<(), VectorDimError> {
434        use crate::value::Value;
435
436        if matches!(value, Value::Null) {
437            return Ok(());
438        }
439
440        match self {
441            DataType::Vector { dimensions } => check_dense_vector_value(value, *dimensions),
442            DataType::BinaryVector { dimensions } => check_binary_vector_value(value, *dimensions),
443            DataType::List(inner) => {
444                let DataType::Vector { dimensions } = inner.as_ref() else {
445                    return Ok(());
446                };
447                let Value::List(tokens) = value else {
448                    return Err(VectorDimError::NotATokenList {
449                        actual: value_variant_name(value),
450                    });
451                };
452                for (token, token_value) in tokens.iter().enumerate() {
453                    check_dense_vector_value(token_value, *dimensions)
454                        .map_err(|e| e.for_token(token))?;
455                }
456                Ok(())
457            }
458            _ => Ok(()),
459        }
460    }
461}
462
463/// Why a value cannot be stored in a declared `VECTOR(dim)` or multi-vector column.
464///
465/// Produced by [`DataType::check_vector_dims`] and [`check_dense_vector_value`].
466/// Messages carry the declared and actual lengths so write-path errors are
467/// actionable; callers prefix the property name and declared type.
468#[derive(Debug, Clone, PartialEq, Eq)]
469pub enum VectorDimError {
470    /// The vector has the wrong number of elements.
471    WrongLength {
472        /// Declared column dimensions.
473        expected: usize,
474        /// Actual element count of the offending value.
475        actual: usize,
476    },
477    /// A list element is not `Int` or `Float`.
478    NonNumericElement {
479        /// Zero-based index of the offending element.
480        index: usize,
481    },
482    /// The value is not a vector or list at all.
483    NotAVector {
484        /// Variant name of the offending value.
485        actual: &'static str,
486    },
487    /// A multi-vector token has the wrong number of elements.
488    TokenWrongLength {
489        /// Zero-based token index within the multi-vector.
490        token: usize,
491        /// Declared per-token dimensions.
492        expected: usize,
493        /// Actual element count of the offending token.
494        actual: usize,
495    },
496    /// A multi-vector token contains a non-numeric element.
497    TokenNonNumericElement {
498        /// Zero-based token index within the multi-vector.
499        token: usize,
500        /// Zero-based index of the offending element within the token.
501        index: usize,
502    },
503    /// A multi-vector token is not a vector or list.
504    TokenNotAVector {
505        /// Zero-based token index within the multi-vector.
506        token: usize,
507        /// Variant name of the offending token.
508        actual: &'static str,
509    },
510    /// The value for a multi-vector column is not a list of tokens.
511    NotATokenList {
512        /// Variant name of the offending value.
513        actual: &'static str,
514    },
515}
516
517impl VectorDimError {
518    /// Maps a dense-kernel error to its per-token counterpart for multi-vector columns.
519    fn for_token(self, token: usize) -> Self {
520        match self {
521            Self::WrongLength { expected, actual } => Self::TokenWrongLength {
522                token,
523                expected,
524                actual,
525            },
526            Self::NonNumericElement { index } => Self::TokenNonNumericElement { token, index },
527            Self::NotAVector { actual } => Self::TokenNotAVector { token, actual },
528            other => other,
529        }
530    }
531}
532
533impl std::fmt::Display for VectorDimError {
534    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
535        match self {
536            Self::WrongLength { expected, actual } => write!(
537                f,
538                "got a vector of length {actual}, expected {expected} dimensions"
539            ),
540            Self::NonNumericElement { index } => {
541                write!(f, "element {index} is not numeric")
542            }
543            Self::NotAVector { actual } => {
544                write!(f, "got a non-vector value of type {actual}")
545            }
546            Self::TokenWrongLength {
547                token,
548                expected,
549                actual,
550            } => write!(
551                f,
552                "token {token} has {actual} dimensions, expected {expected}"
553            ),
554            Self::TokenNonNumericElement { token, index } => {
555                write!(f, "token {token} element {index} is not numeric")
556            }
557            Self::TokenNotAVector { token, actual } => {
558                write!(f, "token {token} is not a vector (got {actual})")
559            }
560            Self::NotATokenList { actual } => write!(
561                f,
562                "got a non-list value of type {actual} for a multi-vector column"
563            ),
564        }
565    }
566}
567
568impl std::error::Error for VectorDimError {}
569
570/// Checks one dense vector value against a declared dimension count.
571///
572/// The per-token kernel behind [`DataType::check_vector_dims`], exposed so the
573/// storage-layer converters can validate individual multi-vector tokens with the
574/// same semantics: `Value::Null` is accepted (a legal null row), a `Value::Vector`
575/// must have exactly `dimensions` elements, and a `Value::List` must additionally
576/// be all-numeric (`Int` or `Float`).
577///
578/// # Errors
579/// Returns [`VectorDimError::WrongLength`] on a length mismatch (an empty list
580/// reports `actual: 0`), [`VectorDimError::NonNumericElement`] for a `String`,
581/// `Null`, or other non-numeric list element, and [`VectorDimError::NotAVector`]
582/// for any other value variant.
583pub fn check_dense_vector_value(
584    value: &crate::value::Value,
585    dimensions: usize,
586) -> Result<(), VectorDimError> {
587    use crate::value::Value;
588
589    match value {
590        Value::Null => Ok(()),
591        Value::Vector(v) => {
592            if v.len() == dimensions {
593                Ok(())
594            } else {
595                Err(VectorDimError::WrongLength {
596                    expected: dimensions,
597                    actual: v.len(),
598                })
599            }
600        }
601        Value::List(items) => {
602            if items.len() != dimensions {
603                return Err(VectorDimError::WrongLength {
604                    expected: dimensions,
605                    actual: items.len(),
606                });
607            }
608            if let Some(index) = items.iter().position(|e| !e.is_number()) {
609                return Err(VectorDimError::NonNumericElement { index });
610            }
611            Ok(())
612        }
613        other => Err(VectorDimError::NotAVector {
614            actual: value_variant_name(other),
615        }),
616    }
617}
618
619/// Validates a value against a declared `BinaryVector(dimensions)` column.
620///
621/// The binary-vector companion to [`check_dense_vector_value`]: `dimensions` is
622/// the number of `u8` lanes. Accepts a [`Value::BinaryVector`](crate::value::Value::BinaryVector)
623/// of exactly that many bytes, or the literal input form — a
624/// [`Value::List`](crate::value::Value::List) of exactly that many integers each
625/// in `[0, 255]` (coerced to bytes by the write path).
626/// [`Value::Null`](crate::value::Value::Null) passes.
627///
628/// # Errors
629/// Returns [`VectorDimError::WrongLength`] on a lane-count mismatch,
630/// [`VectorDimError::NonNumericElement`] if a list element is not an integer in
631/// `[0, 255]`, or [`VectorDimError::NotAVector`] for any other value.
632pub fn check_binary_vector_value(
633    value: &crate::value::Value,
634    dimensions: usize,
635) -> Result<(), VectorDimError> {
636    use crate::value::Value;
637
638    match value {
639        Value::Null => Ok(()),
640        Value::BinaryVector(bytes) => {
641            if bytes.len() == dimensions {
642                Ok(())
643            } else {
644                Err(VectorDimError::WrongLength {
645                    expected: dimensions,
646                    actual: bytes.len(),
647                })
648            }
649        }
650        Value::List(items) => {
651            if items.len() != dimensions {
652                return Err(VectorDimError::WrongLength {
653                    expected: dimensions,
654                    actual: items.len(),
655                });
656            }
657            if let Some(index) = items
658                .iter()
659                .position(|e| !matches!(e.as_i64(), Some(0..=255)))
660            {
661                return Err(VectorDimError::NonNumericElement { index });
662            }
663            Ok(())
664        }
665        other => Err(VectorDimError::NotAVector {
666            actual: value_variant_name(other),
667        }),
668    }
669}
670
671/// Returns a short variant name for a `Value`, used in dimension-mismatch messages.
672fn value_variant_name(value: &crate::value::Value) -> &'static str {
673    use crate::value::Value;
674
675    match value {
676        Value::Null => "Null",
677        Value::Bool(_) => "Bool",
678        Value::Int(_) => "Int",
679        Value::Float(_) => "Float",
680        Value::String(_) => "String",
681        Value::Bytes(_) => "Bytes",
682        Value::List(_) => "List",
683        Value::Map(_) => "Map",
684        Value::Node(_) => "Node",
685        Value::Edge(_) => "Edge",
686        Value::Path(_) => "Path",
687        Value::Vector(_) => "Vector",
688        Value::SparseVector { .. } => "SparseVector",
689        Value::BinaryVector(_) => "BinaryVector",
690        Value::Temporal(_) => "Temporal",
691    }
692}
693
694fn default_created_at() -> DateTime<Utc> {
695    Utc::now()
696}
697
698fn default_state() -> SchemaElementState {
699    SchemaElementState::Active
700}
701
702fn default_version_1() -> u32 {
703    1
704}
705
706#[derive(Clone, Debug, Serialize, Deserialize)]
707pub struct PropertyMeta {
708    pub r#type: DataType,
709    pub nullable: bool,
710    #[serde(default = "default_version_1")]
711    pub added_in: u32, // SchemaVersion
712    #[serde(default = "default_state")]
713    pub state: SchemaElementState,
714    #[serde(default)]
715    pub generation_expression: Option<String>,
716    #[serde(default, skip_serializing_if = "Option::is_none")]
717    pub description: Option<String>,
718}
719
720#[derive(Clone, Debug, Serialize, Deserialize)]
721pub struct LabelMeta {
722    pub id: u16, // LabelId
723    #[serde(default = "default_created_at")]
724    pub created_at: DateTime<Utc>,
725    #[serde(default = "default_state")]
726    pub state: SchemaElementState,
727    #[serde(default, skip_serializing_if = "Option::is_none")]
728    pub description: Option<String>,
729}
730
731#[derive(Clone, Debug, Serialize, Deserialize)]
732pub struct EdgeTypeMeta {
733    /// See [`crate::core::edge_type::EdgeTypeId`] for bit-layout details.
734    pub id: u32,
735    pub src_labels: Vec<String>,
736    pub dst_labels: Vec<String>,
737    #[serde(default = "default_state")]
738    pub state: SchemaElementState,
739    #[serde(default, skip_serializing_if = "Option::is_none")]
740    pub description: Option<String>,
741}
742
743#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
744#[non_exhaustive]
745pub enum ConstraintType {
746    Unique {
747        properties: Vec<String>,
748    },
749    Exists {
750        property: String,
751    },
752    Check {
753        expression: String,
754    },
755    /// Composite node key: the property tuple must be unique AND every listed
756    /// property must be present (non-null). Equivalent to `Unique` over the tuple
757    /// plus `Exists` on each member, enforced together at write time.
758    NodeKey {
759        properties: Vec<String>,
760    },
761}
762
763impl ConstraintType {
764    /// The property tuple whose combination must be unique.
765    ///
766    /// Returns `Some` for the uniqueness-enforcing kinds — `Unique` and `NodeKey`
767    /// (a node key is a unique tuple plus NOT-NULL) — and `None` otherwise. Lets
768    /// the write path share one key-collection/probe path across both kinds.
769    #[must_use]
770    pub fn unique_properties(&self) -> Option<&[String]> {
771        match self {
772            ConstraintType::Unique { properties } | ConstraintType::NodeKey { properties } => {
773                Some(properties)
774            }
775            _ => None,
776        }
777    }
778}
779
780#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
781#[non_exhaustive]
782pub enum ConstraintTarget {
783    Label(String),
784    EdgeType(String),
785}
786
787#[derive(Clone, Debug, Serialize, Deserialize)]
788pub struct Constraint {
789    pub name: String,
790    pub constraint_type: ConstraintType,
791    pub target: ConstraintTarget,
792    pub enabled: bool,
793}
794
795/// Bidirectional registry for dynamically-assigned schemaless edge type IDs.
796///
797/// Edge types not defined in the schema are assigned IDs at runtime with
798/// bit 31 set (see [`crate::core::edge_type`]). This registry maintains
799/// the name-to-ID and ID-to-name mappings for those types.
800#[derive(Clone, Debug, Serialize, Deserialize)]
801pub struct SchemalessEdgeTypeRegistry {
802    name_to_id: HashMap<String, u32>,
803    id_to_name: HashMap<u32, String>,
804    /// Next local ID to assign (0 is reserved for invalid).
805    next_local_id: u32,
806}
807
808impl SchemalessEdgeTypeRegistry {
809    pub fn new() -> Self {
810        Self {
811            name_to_id: HashMap::new(),
812            id_to_name: HashMap::new(),
813            next_local_id: 1,
814        }
815    }
816
817    /// Returns the schemaless ID for `type_name`, assigning a new one if needed.
818    pub fn get_or_assign_id(&mut self, type_name: &str) -> u32 {
819        if let Some(&id) = self.name_to_id.get(type_name) {
820            return id;
821        }
822
823        let id = make_schemaless_id(self.next_local_id);
824        self.next_local_id += 1;
825
826        self.name_to_id.insert(type_name.to_string(), id);
827        self.id_to_name.insert(id, type_name.to_string());
828
829        id
830    }
831
832    /// Looks up the edge type name for a schemaless ID.
833    pub fn type_name_by_id(&self, type_id: u32) -> Option<&str> {
834        self.id_to_name.get(&type_id).map(String::as_str)
835    }
836
837    /// Returns `true` if `type_name` has already been assigned a schemaless ID.
838    pub fn contains(&self, type_name: &str) -> bool {
839        self.name_to_id.contains_key(type_name)
840    }
841
842    /// Looks up the schemaless ID for `type_name` (exact match, read-only).
843    pub fn id_by_name(&self, type_name: &str) -> Option<u32> {
844        self.name_to_id.get(type_name).copied()
845    }
846
847    /// Looks up the edge type ID for `type_name` with case-insensitive matching.
848    pub fn id_by_name_case_insensitive(&self, type_name: &str) -> Option<u32> {
849        self.name_to_id
850            .iter()
851            .find(|(k, _)| k.eq_ignore_ascii_case(type_name))
852            .map(|(_, &id)| id)
853    }
854
855    /// Returns all registered schemaless type IDs.
856    pub fn all_type_ids(&self) -> Vec<u32> {
857        self.id_to_name.keys().copied().collect()
858    }
859
860    /// Returns true if the registry has any schemaless types.
861    pub fn is_empty(&self) -> bool {
862        self.name_to_id.is_empty()
863    }
864}
865
866impl Default for SchemalessEdgeTypeRegistry {
867    fn default() -> Self {
868        Self::new()
869    }
870}
871
872/// First virtual (catalog-resolved) label ID. Label IDs in
873/// `VIRTUAL_LABEL_ID_START..VIRTUAL_LABEL_ID_SENTINEL` are owned by
874/// plugin-registered `CatalogProvider`s and allocated lazily by the
875/// planner via `PluginRegistry::register_virtual_label`. Native label
876/// allocation (`SchemaManager::add_label`) refuses IDs in this range.
877pub const VIRTUAL_LABEL_ID_START: u16 = 0xFF00;
878/// Sentinel "no label" marker, kept distinct from any allocatable ID.
879pub const VIRTUAL_LABEL_ID_SENTINEL: u16 = 0xFFFF;
880
881/// Maximum byte length of a label or edge-type name. (L6)
882///
883/// Generous; the cap is hygiene — the name lands in on-disk dataset paths
884/// and Lance branch names — not a storage limit.
885const MAX_SCHEMA_NAME_LEN: usize = 255;
886
887/// Returns `true` if `id` is in the virtual (catalog-resolved) range.
888#[inline]
889pub fn is_virtual_label_id(id: u16) -> bool {
890    (VIRTUAL_LABEL_ID_START..VIRTUAL_LABEL_ID_SENTINEL).contains(&id)
891}
892
893#[derive(Clone, Debug, Serialize, Deserialize)]
894pub struct Schema {
895    pub schema_version: u32,
896    pub labels: HashMap<String, LabelMeta>,
897    pub edge_types: HashMap<String, EdgeTypeMeta>,
898    pub properties: HashMap<String, HashMap<String, PropertyMeta>>,
899    #[serde(default)]
900    pub indexes: Vec<IndexDefinition>,
901    #[serde(default)]
902    pub constraints: Vec<Constraint>,
903    /// Registry for schemaless edge types (dynamically assigned IDs)
904    #[serde(default)]
905    pub schemaless_registry: SchemalessEdgeTypeRegistry,
906}
907
908impl Default for Schema {
909    fn default() -> Self {
910        Self {
911            schema_version: 1,
912            labels: HashMap::new(),
913            edge_types: HashMap::new(),
914            properties: HashMap::new(),
915            indexes: Vec::new(),
916            constraints: Vec::new(),
917            schemaless_registry: SchemalessEdgeTypeRegistry::new(),
918        }
919    }
920}
921
922impl Schema {
923    /// Bumps `schema_version` to invalidate cached query plans.
924    ///
925    /// Called at the end of every DDL mutation that changes the schema's
926    /// shape (labels, edge types, properties, indexes, constraints). The
927    /// plan-cache eviction guard keys on `schema_version`, so a stale plan
928    /// built against an older shape is discarded once this advances. Uses
929    /// wrapping arithmetic: the value is a coarse change token, not a count,
930    /// so wraparound only risks a missed eviction after 2^32 DDL operations.
931    fn bump_version(&mut self) {
932        self.schema_version = self.schema_version.wrapping_add(1);
933    }
934
935    /// Returns the label name for a given label ID.
936    ///
937    /// Performs a linear scan over all labels. This is efficient because
938    /// the number of labels in a schema is typically small.
939    pub fn label_name_by_id(&self, label_id: u16) -> Option<&str> {
940        self.labels
941            .iter()
942            .find(|(_, meta)| meta.id == label_id)
943            .map(|(name, _)| name.as_str())
944    }
945
946    /// Returns the label ID for a given label name.
947    pub fn label_id_by_name(&self, label_name: &str) -> Option<u16> {
948        self.labels.get(label_name).map(|meta| meta.id)
949    }
950
951    /// Returns the edge type name for a given type ID.
952    ///
953    /// Performs a linear scan over all edge types. This is efficient because
954    /// the number of edge types in a schema is typically small.
955    pub fn edge_type_name_by_id(&self, type_id: u32) -> Option<&str> {
956        self.edge_types
957            .iter()
958            .find(|(_, meta)| meta.id == type_id)
959            .map(|(name, _)| name.as_str())
960    }
961
962    /// Returns the edge type ID for a given type name.
963    pub fn edge_type_id_by_name(&self, type_name: &str) -> Option<u32> {
964        self.edge_types.get(type_name).map(|meta| meta.id)
965    }
966
967    /// Returns the vector index configuration for a given label and property.
968    ///
969    /// Performs a linear scan over all indexes. This is efficient because
970    /// the number of indexes in a schema is typically small.
971    pub fn vector_index_for_property(
972        &self,
973        label: &str,
974        property: &str,
975    ) -> Option<&VectorIndexConfig> {
976        self.indexes.iter().find_map(|idx| {
977            if let IndexDefinition::Vector(config) = idx
978                && config.label == label
979                && config.property == property
980                && config.metadata.status == IndexStatus::Online
981            {
982                return Some(config);
983            }
984            None
985        })
986    }
987
988    /// Returns the scored sparse-vector index configuration for a label/property.
989    pub fn sparse_index_for_property(
990        &self,
991        label: &str,
992        property: &str,
993    ) -> Option<&SparseVectorIndexConfig> {
994        self.indexes.iter().find_map(|idx| {
995            if let IndexDefinition::Sparse(config) = idx
996                && config.label == label
997                && config.property == property
998                && config.metadata.status == IndexStatus::Online
999            {
1000                return Some(config);
1001            }
1002            None
1003        })
1004    }
1005
1006    /// Returns the full-text index configuration for a given label and property.
1007    ///
1008    /// A full-text index covers one or more properties. This returns the config
1009    /// if the specified property is among the indexed properties.
1010    pub fn fulltext_index_for_property(
1011        &self,
1012        label: &str,
1013        property: &str,
1014    ) -> Option<&FullTextIndexConfig> {
1015        self.indexes.iter().find_map(|idx| {
1016            if let IndexDefinition::FullText(config) = idx
1017                && config.label == label
1018                && config.properties.iter().any(|p| p == property)
1019                && config.metadata.status == IndexStatus::Online
1020            {
1021                return Some(config);
1022            }
1023            None
1024        })
1025    }
1026
1027    /// Get label metadata with case-insensitive lookup.
1028    ///
1029    /// This allows queries to match labels regardless of case, providing
1030    /// better user experience when label names vary in casing.
1031    pub fn get_label_case_insensitive(&self, name: &str) -> Option<&LabelMeta> {
1032        self.labels
1033            .iter()
1034            .find(|(k, _)| k.eq_ignore_ascii_case(name))
1035            .map(|(_, v)| v)
1036    }
1037
1038    /// Get the schema-canonical spelling of a label, matched case-insensitively.
1039    ///
1040    /// Returns the stored label name whose spelling differs only in case from
1041    /// `name`, or `None` if no such label is registered. Callers use this to
1042    /// normalize a user-supplied label to the canonical form the storage tier
1043    /// keys on, so case variants resolve to the same vertex table.
1044    pub fn canonical_label_name(&self, name: &str) -> Option<String> {
1045        self.labels
1046            .iter()
1047            .find(|(k, _)| k.eq_ignore_ascii_case(name))
1048            .map(|(k, _)| k.clone())
1049    }
1050
1051    /// Get label ID with case-insensitive lookup.
1052    pub fn label_id_by_name_case_insensitive(&self, label_name: &str) -> Option<u16> {
1053        self.get_label_case_insensitive(label_name)
1054            .map(|meta| meta.id)
1055    }
1056
1057    /// Get edge type metadata with case-insensitive lookup.
1058    ///
1059    /// This allows queries to match edge types regardless of case, providing
1060    /// better user experience when type names vary in casing.
1061    pub fn get_edge_type_case_insensitive(&self, name: &str) -> Option<&EdgeTypeMeta> {
1062        self.edge_types
1063            .iter()
1064            .find(|(k, _)| k.eq_ignore_ascii_case(name))
1065            .map(|(_, v)| v)
1066    }
1067
1068    /// Get edge type ID with case-insensitive lookup (schema-defined types only).
1069    pub fn edge_type_id_by_name_case_insensitive(&self, type_name: &str) -> Option<u32> {
1070        self.get_edge_type_case_insensitive(type_name)
1071            .map(|meta| meta.id)
1072    }
1073
1074    /// Get edge type ID with case-insensitive lookup, checking both
1075    /// schema-defined and schemaless registries.
1076    pub fn edge_type_id_unified_case_insensitive(&self, type_name: &str) -> Option<u32> {
1077        self.edge_type_id_by_name_case_insensitive(type_name)
1078            .or_else(|| {
1079                self.schemaless_registry
1080                    .id_by_name_case_insensitive(type_name)
1081            })
1082    }
1083
1084    /// Returns the edge type ID for `type_name`, checking the schema first
1085    /// and falling back to the schemaless registry (assigning a new ID if needed).
1086    ///
1087    /// Requires `&mut self` because it may assign a new schemaless ID.
1088    /// Use [`edge_type_id_by_name`](Self::edge_type_id_by_name) for read-only schema lookups.
1089    pub fn get_or_assign_edge_type_id(&mut self, type_name: &str) -> u32 {
1090        if let Some(id) = self.edge_type_id_unified(type_name) {
1091            return id;
1092        }
1093        // Reaching here means the type is brand-new to *both* the schema map and
1094        // the schemaless registry (the early return above mirrors exactly what
1095        // `edge_type_id_unified` checks). Minting a new schemaless edge type
1096        // changes the result of `all_edge_type_ids()`, which untyped traversals
1097        // bake into cached plans keyed on `schema_version`. Bump the version so
1098        // those stale plans are evicted — otherwise a `MATCH ()-[r]->()` plan
1099        // built before this type existed silently drops edges of the new type.
1100        let id = self.schemaless_registry.get_or_assign_id(type_name);
1101        self.bump_version();
1102        id
1103    }
1104
1105    /// Read-only unified exact lookup: schema-defined edge type id, falling
1106    /// back to an already-assigned schemaless id.
1107    ///
1108    /// Mirrors exactly the checks [`Self::get_or_assign_edge_type_id`]
1109    /// performs before assigning, so a `Some` here means the assigning path
1110    /// would be a no-op — the basis for `SchemaManager`'s read-lock fast path.
1111    pub fn edge_type_id_unified(&self, type_name: &str) -> Option<u32> {
1112        self.edge_type_id_by_name(type_name)
1113            .or_else(|| self.schemaless_registry.id_by_name(type_name))
1114    }
1115
1116    /// Returns the edge type name for `type_id`, checking both the schema
1117    /// and schemaless registries. Returns an owned `String` because the
1118    /// name may come from either registry.
1119    pub fn edge_type_name_by_id_unified(&self, type_id: u32) -> Option<String> {
1120        if is_schemaless_edge_type(type_id) {
1121            self.schemaless_registry
1122                .type_name_by_id(type_id)
1123                .map(str::to_owned)
1124        } else {
1125            self.edge_type_name_by_id(type_id).map(str::to_owned)
1126        }
1127    }
1128
1129    /// Returns all edge type IDs, including both schema-defined and schemaless types.
1130    /// Used when MATCH queries don't specify an edge type and need to scan all edges.
1131    pub fn all_edge_type_ids(&self) -> Vec<u32> {
1132        let mut ids: Vec<u32> = self.edge_types.values().map(|m| m.id).collect();
1133        ids.extend(self.schemaless_registry.all_type_ids());
1134        ids.sort_unstable();
1135        ids
1136    }
1137}
1138
1139/// Lifecycle status of an index.
1140#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
1141pub enum IndexStatus {
1142    /// Index is up-to-date and available for queries.
1143    #[default]
1144    Online,
1145    /// Index is currently being rebuilt.
1146    Building,
1147    /// Index is outdated and scheduled for rebuild.
1148    Stale,
1149    /// Index rebuild failed after exhausting retries.
1150    Failed,
1151}
1152
1153/// Metadata tracking the lifecycle state of an index.
1154#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
1155pub struct IndexMetadata {
1156    /// Current lifecycle status.
1157    #[serde(default)]
1158    pub status: IndexStatus,
1159    /// When the index was last successfully built.
1160    #[serde(default)]
1161    pub last_built_at: Option<DateTime<Utc>>,
1162    /// Row count of the dataset when the index was last built.
1163    #[serde(default)]
1164    pub row_count_at_build: Option<u64>,
1165}
1166
1167#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1168#[serde(tag = "type")]
1169#[non_exhaustive]
1170pub enum IndexDefinition {
1171    Vector(VectorIndexConfig),
1172    FullText(FullTextIndexConfig),
1173    Scalar(ScalarIndexConfig),
1174    Inverted(InvertedIndexConfig),
1175    JsonFullText(JsonFtsIndexConfig),
1176    /// Scored sparse-vector (SPLADE / learned-sparse) inverted index.
1177    Sparse(SparseVectorIndexConfig),
1178}
1179
1180impl IndexDefinition {
1181    /// Returns the index name for any variant.
1182    pub fn name(&self) -> &str {
1183        match self {
1184            IndexDefinition::Vector(c) => &c.name,
1185            IndexDefinition::FullText(c) => &c.name,
1186            IndexDefinition::Scalar(c) => &c.name,
1187            IndexDefinition::Inverted(c) => &c.name,
1188            IndexDefinition::JsonFullText(c) => &c.name,
1189            IndexDefinition::Sparse(c) => &c.name,
1190        }
1191    }
1192
1193    /// Returns the label this index is defined on.
1194    pub fn label(&self) -> &str {
1195        match self {
1196            IndexDefinition::Vector(c) => &c.label,
1197            IndexDefinition::FullText(c) => &c.label,
1198            IndexDefinition::Scalar(c) => &c.label,
1199            IndexDefinition::Inverted(c) => &c.label,
1200            IndexDefinition::JsonFullText(c) => &c.label,
1201            IndexDefinition::Sparse(c) => &c.label,
1202        }
1203    }
1204
1205    /// Returns a reference to the index lifecycle metadata.
1206    pub fn metadata(&self) -> &IndexMetadata {
1207        match self {
1208            IndexDefinition::Vector(c) => &c.metadata,
1209            IndexDefinition::FullText(c) => &c.metadata,
1210            IndexDefinition::Scalar(c) => &c.metadata,
1211            IndexDefinition::Inverted(c) => &c.metadata,
1212            IndexDefinition::JsonFullText(c) => &c.metadata,
1213            IndexDefinition::Sparse(c) => &c.metadata,
1214        }
1215    }
1216
1217    /// Returns a mutable reference to the index lifecycle metadata.
1218    pub fn metadata_mut(&mut self) -> &mut IndexMetadata {
1219        match self {
1220            IndexDefinition::Vector(c) => &mut c.metadata,
1221            IndexDefinition::FullText(c) => &mut c.metadata,
1222            IndexDefinition::Scalar(c) => &mut c.metadata,
1223            IndexDefinition::Inverted(c) => &mut c.metadata,
1224            IndexDefinition::JsonFullText(c) => &mut c.metadata,
1225            IndexDefinition::Sparse(c) => &mut c.metadata,
1226        }
1227    }
1228}
1229
1230#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1231pub struct InvertedIndexConfig {
1232    pub name: String,
1233    pub label: String,
1234    pub property: String,
1235    #[serde(default = "default_normalize")]
1236    pub normalize: bool,
1237    #[serde(default = "default_max_terms_per_doc")]
1238    pub max_terms_per_doc: usize,
1239    #[serde(default)]
1240    pub metadata: IndexMetadata,
1241}
1242
1243fn default_normalize() -> bool {
1244    true
1245}
1246
1247fn default_max_terms_per_doc() -> usize {
1248    10_000
1249}
1250
1251/// Configuration for a scored sparse-vector (SPLADE / learned-sparse) index.
1252///
1253/// The index stores per-term postings `(term_id, vids, weights, max_impact)`
1254/// and scores by dot product. `quantize` controls 8-bit weight quantization at
1255/// the postings boundary (≈ lossless, ~4× smaller; default on). P2 block-max
1256/// pruning knobs are added in a later milestone.
1257#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1258pub struct SparseVectorIndexConfig {
1259    pub name: String,
1260    pub label: String,
1261    pub property: String,
1262    /// Term-space cardinality (max term id + 1), for validation/config.
1263    pub dimensions: usize,
1264    /// Quantize stored weights to 8-bit (per-term scale). Default on.
1265    #[serde(default = "default_sparse_quantize")]
1266    pub quantize: bool,
1267    /// Auto-embedding source. When set, a declared text column is embedded into
1268    /// this sparse column via the xervo sparse model at write time (and a text
1269    /// query is embedded at query time) — mirrors `VectorIndexConfig`.
1270    #[serde(default)]
1271    pub embedding_config: Option<EmbeddingConfig>,
1272    #[serde(default)]
1273    pub metadata: IndexMetadata,
1274}
1275
1276fn default_sparse_quantize() -> bool {
1277    true
1278}
1279
1280#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1281pub struct VectorIndexConfig {
1282    pub name: String,
1283    pub label: String,
1284    pub property: String,
1285    pub index_type: VectorIndexType,
1286    pub metric: DistanceMetric,
1287    pub embedding_config: Option<EmbeddingConfig>,
1288    #[serde(default)]
1289    pub metadata: IndexMetadata,
1290}
1291
1292#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1293pub struct EmbeddingConfig {
1294    /// Model alias in the Uni-Xervo catalog (for example: "embed/default").
1295    pub alias: String,
1296    pub source_properties: Vec<String>,
1297    pub batch_size: usize,
1298    /// Prefix prepended to text before embedding during auto-embed (document side).
1299    /// Example: `"search_document: "` for Nomic models. Include any trailing space.
1300    #[serde(default)]
1301    pub document_prefix: Option<String>,
1302    /// Prefix prepended to text before embedding during query-time embed calls.
1303    /// Example: `"search_query: "` for Nomic models. Include any trailing space.
1304    #[serde(default)]
1305    pub query_prefix: Option<String>,
1306}
1307
1308#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1309#[non_exhaustive]
1310pub enum VectorIndexType {
1311    Flat,
1312    IvfFlat {
1313        num_partitions: u32,
1314    },
1315    IvfPq {
1316        num_partitions: u32,
1317        num_sub_vectors: u32,
1318        bits_per_subvector: u8,
1319    },
1320    IvfSq {
1321        num_partitions: u32,
1322    },
1323    IvfRq {
1324        num_partitions: u32,
1325        #[serde(default)]
1326        num_bits: Option<u8>,
1327    },
1328    HnswFlat {
1329        m: u32,
1330        ef_construction: u32,
1331        #[serde(default)]
1332        num_partitions: Option<u32>,
1333    },
1334    HnswSq {
1335        m: u32,
1336        ef_construction: u32,
1337        #[serde(default)]
1338        num_partitions: Option<u32>,
1339    },
1340    HnswPq {
1341        m: u32,
1342        ef_construction: u32,
1343        num_sub_vectors: u32,
1344        #[serde(default)]
1345        num_partitions: Option<u32>,
1346    },
1347    /// MUVERA (arXiv:2405.19504) Fixed-Dimensional Encoding for multi-vector
1348    /// (ColBERT/MaxSim) columns. The source multi-vector is encoded into a single
1349    /// derived `Vector<fde_dim>` column, and `inner` is the single-vector ANN index
1350    /// type built over that derived column (always with the `Dot` metric — the FDE
1351    /// inner product approximates MaxSim). The exact MaxSim re-rank still uses the
1352    /// `VectorIndexConfig.metric`. See `uni_query_functions::muvera`.
1353    Muvera {
1354        /// SimHash hyperplanes per repetition (`2^k_sim` buckets).
1355        k_sim: u32,
1356        /// Independent repetitions concatenated into the FDE.
1357        reps: u32,
1358        /// Inner-projection target dim (`0` = no projection, use the source dim).
1359        d_proj: u32,
1360        /// Master seed; persisted so query-time encoding matches doc-time encoding.
1361        seed: u64,
1362        /// The single-vector ANN index built over the derived FDE column.
1363        inner: Box<VectorIndexType>,
1364    },
1365}
1366
1367#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1368#[non_exhaustive]
1369pub enum DistanceMetric {
1370    Cosine,
1371    L2,
1372    Dot,
1373    /// L1 / Manhattan distance (`Σ|xᵢ − yᵢ|`). Exact/brute-force only — Lance
1374    /// ANN indexes do not support it, so an L1 column cannot build an ANN index.
1375    L1,
1376    /// Hamming distance over binary vectors: the number of differing bits
1377    /// (`Σ popcount(aᵢ ⊕ bᵢ)` across `u8` lanes). Applies only to
1378    /// [`DataType::BinaryVector`] columns. Exact/brute-force only — Lance ANN
1379    /// over binary metrics is not wired here.
1380    Hamming,
1381    /// Jaccard distance over binary vectors: `1 − |A ∩ B| / |A ∪ B|` computed
1382    /// bitwise (two all-zero vectors are defined as distance `0`). Applies only
1383    /// to [`DataType::BinaryVector`] columns. Exact/brute-force only — Lance has
1384    /// no Jaccard ANN.
1385    Jaccard,
1386}
1387
1388impl DistanceMetric {
1389    /// Computes the distance between two vectors using this metric.
1390    ///
1391    /// All metrics follow LanceDB conventions so that lower values indicate
1392    /// greater similarity:
1393    /// - **L2**: squared Euclidean distance.
1394    /// - **Cosine**: `1.0 - cosine_similarity` (range \[0, 2\]).
1395    /// - **Dot**: negative dot product.
1396    /// - **L1**: Manhattan distance (`Σ|xᵢ − yᵢ|`).
1397    ///
1398    /// # Panics
1399    ///
1400    /// Panics if `a` and `b` have different lengths.
1401    pub fn compute_distance(&self, a: &[f32], b: &[f32]) -> f32 {
1402        assert_eq!(a.len(), b.len(), "vector dimension mismatch");
1403        match self {
1404            DistanceMetric::L2 => a.iter().zip(b).map(|(x, y)| (x - y).powi(2)).sum(),
1405            DistanceMetric::L1 => a.iter().zip(b).map(|(x, y)| (x - y).abs()).sum(),
1406            DistanceMetric::Cosine => {
1407                let dot: f32 = a.iter().zip(b).map(|(x, y)| x * y).sum();
1408                let norm_a: f32 = a.iter().map(|x| x.powi(2)).sum::<f32>().sqrt();
1409                let norm_b: f32 = b.iter().map(|x| x.powi(2)).sum::<f32>().sqrt();
1410                let denom = norm_a * norm_b;
1411                if denom == 0.0 { 1.0 } else { 1.0 - dot / denom }
1412            }
1413            DistanceMetric::Dot => {
1414                let dot: f32 = a.iter().zip(b).map(|(x, y)| x * y).sum();
1415                -dot
1416            }
1417            // Binary metrics operate on `&[u8]`, not float lanes; routing a
1418            // float vector here is a programming error (a `BinaryVector` column
1419            // is scored via `compute_distance_binary`).
1420            DistanceMetric::Hamming | DistanceMetric::Jaccard => {
1421                panic!("{self:?} is a binary-vector metric; use compute_distance_binary")
1422            }
1423        }
1424    }
1425
1426    /// Returns `true` if this metric operates on binary vectors (`&[u8]` lanes)
1427    /// rather than float vectors — i.e. [`DistanceMetric::Hamming`] or
1428    /// [`DistanceMetric::Jaccard`].
1429    ///
1430    /// Callers use this to route between [`DistanceMetric::compute_distance`]
1431    /// and [`DistanceMetric::compute_distance_binary`], and to decide that a
1432    /// column is scored exact/brute-force (binary metrics have no ANN backend).
1433    pub fn is_binary(&self) -> bool {
1434        matches!(self, DistanceMetric::Hamming | DistanceMetric::Jaccard)
1435    }
1436
1437    /// Computes the distance between two binary vectors (`u8` lanes) using this
1438    /// metric, following the LanceDB "lower is more similar" convention.
1439    ///
1440    /// - **Hamming**: number of differing bits, `Σ popcount(aᵢ ⊕ bᵢ)`.
1441    /// - **Jaccard**: `1 − |A ∩ B| / |A ∪ B|` computed bitwise; two all-zero
1442    ///   vectors are defined as distance `0`.
1443    ///
1444    /// # Panics
1445    ///
1446    /// Panics if `a` and `b` have different lengths, or if `self` is a
1447    /// float metric ([`DistanceMetric::L2`], `Cosine`, `Dot`, or `L1`) — those
1448    /// are computed via [`DistanceMetric::compute_distance`].
1449    pub fn compute_distance_binary(&self, a: &[u8], b: &[u8]) -> f32 {
1450        assert_eq!(a.len(), b.len(), "binary vector dimension mismatch");
1451        match self {
1452            DistanceMetric::Hamming => a
1453                .iter()
1454                .zip(b)
1455                .map(|(x, y)| (x ^ y).count_ones())
1456                .sum::<u32>() as f32,
1457            DistanceMetric::Jaccard => {
1458                let mut inter: u32 = 0;
1459                let mut union: u32 = 0;
1460                for (x, y) in a.iter().zip(b) {
1461                    inter += (x & y).count_ones();
1462                    union += (x | y).count_ones();
1463                }
1464                if union == 0 {
1465                    0.0
1466                } else {
1467                    1.0 - (inter as f32) / (union as f32)
1468                }
1469            }
1470            other => panic!("{other:?} is a float-vector metric; use compute_distance"),
1471        }
1472    }
1473}
1474
1475#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1476pub struct FullTextIndexConfig {
1477    pub name: String,
1478    pub label: String,
1479    pub properties: Vec<String>,
1480    pub tokenizer: TokenizerConfig,
1481    pub with_positions: bool,
1482    #[serde(default)]
1483    pub metadata: IndexMetadata,
1484}
1485
1486#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1487#[non_exhaustive]
1488pub enum TokenizerConfig {
1489    Standard,
1490    Whitespace,
1491    Ngram {
1492        min: u8,
1493        max: u8,
1494    },
1495    Custom {
1496        name: String,
1497    },
1498    /// Fully specified analyzer pipeline (base tokenizer + language + filters).
1499    ///
1500    /// Carries stemming, stop-word, lowercasing, ASCII-folding and token-length
1501    /// configuration so full-text indexes honor the requested analysis instead
1502    /// of falling back to the hardcoded standard tokenizer.
1503    Analyzer(AnalyzerConfig),
1504}
1505
1506/// Full analyzer pipeline configuration for a full-text index.
1507///
1508/// This is a backend-agnostic, plainly serializable description of the token
1509/// analysis chain. The `uni-store` Lance backend maps it onto the underlying
1510/// `InvertedIndexParams`; this crate stays free of any Lance dependency.
1511///
1512/// Every field carries `#[serde(default)]` so schemas persisted before a field
1513/// existed still deserialize.
1514#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1515pub struct AnalyzerConfig {
1516    /// Base tokenizer that splits raw text into tokens.
1517    #[serde(default)]
1518    pub base: BaseTokenizer,
1519    /// Language used for stemming and built-in stop-word lists.
1520    #[serde(default)]
1521    pub language: FtsLanguage,
1522    /// Whether to lowercase tokens.
1523    #[serde(default = "default_true")]
1524    pub lower_case: bool,
1525    /// Whether to apply language-specific stemming.
1526    #[serde(default = "default_true")]
1527    pub stem: bool,
1528    /// Whether to drop stop words (built-in list, or `custom_stop_words`).
1529    #[serde(default = "default_true")]
1530    pub remove_stop_words: bool,
1531    /// Explicit stop-word list overriding the language's built-in list.
1532    #[serde(default)]
1533    pub custom_stop_words: Option<Vec<String>>,
1534    /// Whether to fold accented characters to ASCII (é → e).
1535    #[serde(default = "default_true")]
1536    pub ascii_folding: bool,
1537    /// Drop tokens longer than this many bytes (`None` keeps the backend default).
1538    #[serde(default)]
1539    pub max_token_length: Option<u32>,
1540}
1541
1542impl Default for AnalyzerConfig {
1543    fn default() -> Self {
1544        Self {
1545            base: BaseTokenizer::default(),
1546            language: FtsLanguage::default(),
1547            lower_case: true,
1548            stem: true,
1549            remove_stop_words: true,
1550            custom_stop_words: None,
1551            ascii_folding: true,
1552            max_token_length: None,
1553        }
1554    }
1555}
1556
1557/// Base tokenizer that produces the initial token stream.
1558///
1559/// `Custom` is a passthrough for backend-native tokenizers such as
1560/// `"lindera/*"` or `"jieba/*"` (which require external dictionaries).
1561#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
1562#[non_exhaustive]
1563pub enum BaseTokenizer {
1564    /// Split on whitespace and punctuation (recommended default).
1565    #[default]
1566    Simple,
1567    /// Split on whitespace only.
1568    Whitespace,
1569    /// No tokenization; the whole field is one token.
1570    Raw,
1571    /// Character N-gram tokenizer with inclusive `[min, max]` gram lengths.
1572    Ngram {
1573        /// Minimum gram length (must be `>= 1` and `<= max`).
1574        min: u32,
1575        /// Maximum gram length.
1576        max: u32,
1577    },
1578    /// Backend-native tokenizer name, passed through verbatim (e.g. `"jieba/default"`).
1579    Custom(String),
1580}
1581
1582/// Language used for stemming and built-in stop-word removal.
1583///
1584/// Mirrors the 18 languages the Lance tokenizer supports. Note that not every
1585/// language ships a built-in stop-word list; the backend mapper handles those
1586/// cases (see `uni-store`).
1587#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
1588#[non_exhaustive]
1589pub enum FtsLanguage {
1590    /// Arabic.
1591    Arabic,
1592    /// Danish.
1593    Danish,
1594    /// Dutch.
1595    Dutch,
1596    /// English (default).
1597    #[default]
1598    English,
1599    /// Finnish.
1600    Finnish,
1601    /// French.
1602    French,
1603    /// German.
1604    German,
1605    /// Greek.
1606    Greek,
1607    /// Hungarian.
1608    Hungarian,
1609    /// Italian.
1610    Italian,
1611    /// Norwegian.
1612    Norwegian,
1613    /// Portuguese.
1614    Portuguese,
1615    /// Romanian.
1616    Romanian,
1617    /// Russian.
1618    Russian,
1619    /// Spanish.
1620    Spanish,
1621    /// Swedish.
1622    Swedish,
1623    /// Tamil.
1624    Tamil,
1625    /// Turkish.
1626    Turkish,
1627}
1628
1629/// Serde default helper: boolean fields that default to `true`.
1630fn default_true() -> bool {
1631    true
1632}
1633
1634#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1635pub struct JsonFtsIndexConfig {
1636    pub name: String,
1637    pub label: String,
1638    pub column: String,
1639    #[serde(default)]
1640    pub paths: Vec<String>,
1641    #[serde(default)]
1642    pub with_positions: bool,
1643    #[serde(default)]
1644    pub metadata: IndexMetadata,
1645}
1646
1647#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1648pub struct ScalarIndexConfig {
1649    pub name: String,
1650    pub label: String,
1651    pub properties: Vec<String>,
1652    pub index_type: ScalarIndexType,
1653    pub where_clause: Option<String>,
1654    #[serde(default)]
1655    pub metadata: IndexMetadata,
1656}
1657
1658#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
1659#[non_exhaustive]
1660pub enum ScalarIndexType {
1661    BTree,
1662    Hash,
1663    Bitmap,
1664    LabelList,
1665}
1666
1667pub struct SchemaManager {
1668    store: Arc<dyn ObjectStore>,
1669    path: ObjectStorePath,
1670    schema: RwLock<Arc<Schema>>,
1671}
1672
1673impl SchemaManager {
1674    pub async fn load(path: impl AsRef<Path>) -> Result<Self> {
1675        let path = path.as_ref();
1676        let parent = path
1677            .parent()
1678            .ok_or_else(|| anyhow!("Invalid schema path"))?;
1679        let filename = path
1680            .file_name()
1681            .ok_or_else(|| anyhow!("Invalid schema filename"))?
1682            .to_str()
1683            .ok_or_else(|| anyhow!("Invalid utf8 filename"))?;
1684
1685        let store = Arc::new(LocalFileSystem::new_with_prefix(parent)?);
1686        let obj_path = ObjectStorePath::from(filename);
1687
1688        Self::load_from_store(store, &obj_path).await
1689    }
1690
1691    pub async fn load_from_store(
1692        store: Arc<dyn ObjectStore>,
1693        path: &ObjectStorePath,
1694    ) -> Result<Self> {
1695        match store.get(path).await {
1696            Ok(result) => {
1697                let bytes = result.bytes().await?;
1698                let content = String::from_utf8(bytes.to_vec())?;
1699                let mut schema: Schema = serde_json::from_str(&content)?;
1700                // Self-heal catalogs that grew super-linearly under the
1701                // pre-fix `add_index` (issue rustic-ai/uni-db#63). Collapse
1702                // duplicate index entries by name, keeping the *last*
1703                // occurrence — matches the upsert semantics in `add_index`
1704                // and preserves whatever metadata the most recent rebuild
1705                // wrote. The dedup persists on the next mutation that
1706                // calls `save()`.
1707                let original_len = schema.indexes.len();
1708                if original_len > 0 {
1709                    let mut seen: std::collections::HashSet<String> =
1710                        std::collections::HashSet::with_capacity(original_len);
1711                    let mut dedup: Vec<IndexDefinition> = schema
1712                        .indexes
1713                        .iter()
1714                        .rev()
1715                        .filter(|idx| seen.insert(idx.name().to_string()))
1716                        .cloned()
1717                        .collect();
1718                    dedup.reverse();
1719                    if dedup.len() != original_len {
1720                        tracing::warn!(
1721                            collapsed = original_len - dedup.len(),
1722                            kept = dedup.len(),
1723                            "schema.indexes: collapsed duplicate entries on load (issue #63)"
1724                        );
1725                        schema.indexes = dedup;
1726                    }
1727                }
1728                Ok(Self {
1729                    store,
1730                    path: path.clone(),
1731                    schema: RwLock::new(Arc::new(schema)),
1732                })
1733            }
1734            Err(object_store::Error::NotFound { .. }) => Ok(Self {
1735                store,
1736                path: path.clone(),
1737                schema: RwLock::new(Arc::new(Schema::default())),
1738            }),
1739            Err(e) => Err(anyhow::Error::from(e)),
1740        }
1741    }
1742
1743    pub async fn save(&self) -> Result<()> {
1744        let content = {
1745            let schema_guard = acquire_read(&self.schema, "schema")?;
1746            serde_json::to_string_pretty(&**schema_guard)?
1747        };
1748        self.store
1749            .put(&self.path, content.into())
1750            .await
1751            .map_err(anyhow::Error::from)?;
1752        Ok(())
1753    }
1754
1755    pub fn path(&self) -> &ObjectStorePath {
1756        &self.path
1757    }
1758
1759    pub fn schema(&self) -> Arc<Schema> {
1760        self.schema
1761            .read()
1762            .expect("Schema lock poisoned - a thread panicked while holding it")
1763            .clone()
1764    }
1765
1766    /// Normalize function names in an expression to uppercase for case-insensitive matching.
1767    /// Examples: "lower(email)" -> "LOWER(email)", "trim(name)" -> "TRIM(name)"
1768    fn normalize_function_names(expr: &str) -> String {
1769        let mut result = String::with_capacity(expr.len());
1770        let mut chars = expr.chars().peekable();
1771
1772        while let Some(ch) = chars.next() {
1773            if ch.is_alphabetic() {
1774                // Collect identifier
1775                let mut ident = String::new();
1776                ident.push(ch);
1777
1778                while let Some(&next) = chars.peek() {
1779                    if next.is_alphanumeric() || next == '_' {
1780                        ident.push(chars.next().unwrap());
1781                    } else {
1782                        break;
1783                    }
1784                }
1785
1786                // If followed by '(', it's a function call - uppercase it
1787                if chars.peek() == Some(&'(') {
1788                    result.push_str(&ident.to_uppercase());
1789                } else {
1790                    result.push_str(&ident); // Keep property names as-is
1791                }
1792            } else {
1793                result.push(ch);
1794            }
1795        }
1796
1797        result
1798    }
1799
1800    /// Generate a consistent internal column name for an expression index.
1801    /// Uses a hash suffix to ensure uniqueness for different expressions that
1802    /// might sanitize to the same string (e.g., "a+b" and "a-b" both become "a_b").
1803    ///
1804    /// IMPORTANT: Uses FNV-1a hash which is stable across Rust versions and platforms.
1805    /// DefaultHasher is not guaranteed to be stable and could break persistent data
1806    /// if the hash changes after a compiler upgrade.
1807    pub fn generated_column_name(expr: &str) -> String {
1808        // Normalize function names to uppercase for case-insensitive matching
1809        let normalized = Self::normalize_function_names(expr);
1810
1811        let sanitized = normalized
1812            .replace(|c: char| !c.is_alphanumeric(), "_")
1813            .trim_matches('_')
1814            .to_string();
1815
1816        // FNV-1a 64-bit hash - stable across Rust versions and platforms
1817        const FNV_OFFSET_BASIS: u64 = 14695981039346656037;
1818        const FNV_PRIME: u64 = 1099511628211;
1819
1820        let mut hash = FNV_OFFSET_BASIS;
1821        for byte in normalized.as_bytes() {
1822            hash ^= *byte as u64;
1823            hash = hash.wrapping_mul(FNV_PRIME);
1824        }
1825
1826        format!("_gen_{}_{:x}", sanitized, hash)
1827    }
1828
1829    pub fn replace_schema(&self, new_schema: Schema) {
1830        let mut schema = self
1831            .schema
1832            .write()
1833            .expect("Schema lock poisoned - a thread panicked while holding it");
1834        *schema = Arc::new(new_schema);
1835    }
1836
1837    /// Build a fork-scoped manager whose schema is `primary ⊕ overlay`.
1838    ///
1839    /// Used by `UniInner::at_fork` to give a forked session a schema view
1840    /// that includes any labels/edge-types/properties the fork has
1841    /// introduced on top of primary. The returned manager owns its own
1842    /// in-memory `Arc<Schema>` — mutations to it never reach primary's
1843    /// schema file. The returned manager is *not* intended for `.save()`;
1844    /// fork-overlay persistence is owned by the registry layer
1845    /// (`catalog/fork_schemas/{fork_id}.json`).
1846    ///
1847    /// In Phase 1 the delta is always empty, so the merge is a clone.
1848    /// Phase 2 starts populating it when on-the-fly label creation lands.
1849    #[must_use]
1850    pub fn with_overlay(&self, overlay: &crate::core::fork::SchemaDelta) -> Arc<Self> {
1851        let primary = self.schema();
1852        let merged = if overlay.is_empty() {
1853            (*primary).clone()
1854        } else {
1855            let mut merged = (*primary).clone();
1856            for (name, label) in &overlay.added_labels {
1857                merged.labels.insert(name.clone(), label.clone());
1858            }
1859            for (name, edge_type) in &overlay.added_edge_types {
1860                merged.edge_types.insert(name.clone(), edge_type.clone());
1861            }
1862            for addition in &overlay.added_properties {
1863                let props = merged.properties.entry(addition.owner.clone()).or_default();
1864                props.insert(
1865                    addition.property.clone(),
1866                    PropertyMeta {
1867                        r#type: addition.data_type.clone(),
1868                        nullable: addition.nullable,
1869                        added_in: merged.schema_version,
1870                        state: SchemaElementState::Active,
1871                        generation_expression: None,
1872                        description: None,
1873                    },
1874                );
1875            }
1876            merged
1877        };
1878
1879        Arc::new(Self {
1880            store: self.store.clone(),
1881            path: self.path.clone(),
1882            schema: RwLock::new(Arc::new(merged)),
1883        })
1884    }
1885
1886    pub fn next_label_id(&self) -> u16 {
1887        self.schema()
1888            .labels
1889            .values()
1890            .map(|l| l.id)
1891            .max()
1892            .unwrap_or(0)
1893            + 1
1894    }
1895
1896    pub fn next_type_id(&self) -> u32 {
1897        let max_schema_id = self
1898            .schema()
1899            .edge_types
1900            .values()
1901            .map(|t| t.id)
1902            .max()
1903            .unwrap_or(0);
1904
1905        // Ensure we stay in schema'd ID space (bit 31 = 0)
1906        if max_schema_id >= MAX_SCHEMA_TYPE_ID {
1907            panic!("Schema edge type ID exhaustion");
1908        }
1909
1910        max_schema_id + 1
1911    }
1912
1913    /// Validate a label or edge-type name at definition time. (L6)
1914    ///
1915    /// Names flow into on-disk dataset paths (`vertices_{name}.lance`) and
1916    /// Lance branch names (`fork_{id}_{…}`); a name with a path separator,
1917    /// whitespace, or a control character corrupts those paths and breaks
1918    /// fork creation. Such names were never actually usable, so they are
1919    /// rejected up front rather than failing later. `.` is allowed
1920    /// (path-safe and common in qualified names).
1921    ///
1922    /// Public so the fork-create path can apply the same rule as a backstop
1923    /// over names that entered the schema through an infallible interning
1924    /// path (e.g. schemaless `get_or_assign_edge_type_id`).
1925    ///
1926    /// # Errors
1927    /// Returns an error if `name` is empty/all-whitespace, exceeds
1928    /// `MAX_SCHEMA_NAME_LEN` bytes, or contains a control, whitespace,
1929    /// `/`, or `\` character.
1930    pub fn validate_schema_element_name(kind: &str, name: &str) -> Result<()> {
1931        if name.is_empty() || name.chars().all(char::is_whitespace) {
1932            return Err(anyhow!(
1933                "{kind} name must be non-empty and not all whitespace"
1934            ));
1935        }
1936        if name.len() > MAX_SCHEMA_NAME_LEN {
1937            return Err(anyhow!("{kind} name exceeds {MAX_SCHEMA_NAME_LEN} bytes"));
1938        }
1939        if let Some(c) = name
1940            .chars()
1941            .find(|c| c.is_control() || c.is_whitespace() || matches!(c, '/' | '\\'))
1942        {
1943            return Err(anyhow!(
1944                "{kind} name '{name}' contains an unsafe character ({c:?})"
1945            ));
1946        }
1947        Ok(())
1948    }
1949
1950    pub fn add_label(&self, name: &str) -> Result<u16> {
1951        self.add_label_with_desc(name, None)
1952    }
1953
1954    pub fn add_label_with_desc(&self, name: &str, description: Option<String>) -> Result<u16> {
1955        Self::validate_schema_element_name("Label", name)?;
1956        let mut guard = acquire_write(&self.schema, "schema")?;
1957        let schema = Arc::make_mut(&mut *guard);
1958        if schema.labels.contains_key(name) {
1959            return Err(anyhow!("Label '{}' already exists", name));
1960        }
1961
1962        let id = schema.labels.values().map(|l| l.id).max().unwrap_or(0) + 1;
1963        if id >= VIRTUAL_LABEL_ID_START {
1964            return Err(anyhow!(
1965                "Native label space exhausted (next id {id:#x} would enter the \
1966                 virtual range {VIRTUAL_LABEL_ID_START:#x}..{VIRTUAL_LABEL_ID_SENTINEL:#x} \
1967                 reserved for catalog-resolved labels)"
1968            ));
1969        }
1970        schema.labels.insert(
1971            name.to_string(),
1972            LabelMeta {
1973                id,
1974                created_at: Utc::now(),
1975                state: SchemaElementState::Active,
1976                description,
1977            },
1978        );
1979        schema.bump_version();
1980        Ok(id)
1981    }
1982
1983    pub fn add_edge_type(
1984        &self,
1985        name: &str,
1986        src_labels: Vec<String>,
1987        dst_labels: Vec<String>,
1988    ) -> Result<u32> {
1989        self.add_edge_type_with_desc(name, src_labels, dst_labels, None)
1990    }
1991
1992    pub fn add_edge_type_with_desc(
1993        &self,
1994        name: &str,
1995        src_labels: Vec<String>,
1996        dst_labels: Vec<String>,
1997        description: Option<String>,
1998    ) -> Result<u32> {
1999        Self::validate_schema_element_name("Edge type", name)?;
2000        let mut guard = acquire_write(&self.schema, "schema")?;
2001        let schema = Arc::make_mut(&mut *guard);
2002        if schema.edge_types.contains_key(name) {
2003            return Err(anyhow!("Edge type '{}' already exists", name));
2004        }
2005
2006        let id = schema.edge_types.values().map(|t| t.id).max().unwrap_or(0) + 1;
2007
2008        // Stay in the schema-defined sub-range (bit 31 = 0, and below the
2009        // virtual reservation `VIRTUAL_EDGE_TYPE_ID_START`) — same bound as
2010        // `add_edge_type`, so the two entry points cannot disagree on the
2011        // legal ceiling.
2012        if id >= VIRTUAL_EDGE_TYPE_ID_START {
2013            return Err(anyhow!(
2014                "Native edge type space exhausted (next id {id:#x} would enter the \
2015                 virtual range {VIRTUAL_EDGE_TYPE_ID_START:#x}..{VIRTUAL_EDGE_TYPE_ID_SENTINEL:#x} \
2016                 reserved for catalog-resolved edge types)"
2017            ));
2018        }
2019
2020        schema.edge_types.insert(
2021            name.to_string(),
2022            EdgeTypeMeta {
2023                id,
2024                src_labels,
2025                dst_labels,
2026                state: SchemaElementState::Active,
2027                description,
2028            },
2029        );
2030        schema.bump_version();
2031        Ok(id)
2032    }
2033
2034    /// Delegates to [`Schema::get_or_assign_edge_type_id`].
2035    ///
2036    /// Read-lock fast path: the type name is almost always already known
2037    /// (it is constant per statement but resolved per row by the CREATE
2038    /// executor), and the slow path's write lock + `Arc::make_mut` deep-clones
2039    /// the whole `Schema` whenever the Arc is shared — which under SSI it
2040    /// always is. Double-checked: on a miss, `Schema::get_or_assign_edge_type_id`
2041    /// re-checks under the write lock, so two racing assigners converge on one id.
2042    pub fn get_or_assign_edge_type_id(&self, type_name: &str) -> u32 {
2043        {
2044            let guard = acquire_read(&self.schema, "schema")
2045                .expect("Schema lock poisoned - a thread panicked while holding it");
2046            if let Some(id) = guard.edge_type_id_unified(type_name) {
2047                return id;
2048            }
2049        }
2050        let mut guard = acquire_write(&self.schema, "schema")
2051            .expect("Schema lock poisoned - a thread panicked while holding it");
2052        let schema = Arc::make_mut(&mut *guard);
2053        schema.get_or_assign_edge_type_id(type_name)
2054    }
2055
2056    /// Delegates to [`Schema::edge_type_name_by_id_unified`].
2057    pub fn edge_type_name_by_id_unified(&self, type_id: u32) -> Option<String> {
2058        let schema = acquire_read(&self.schema, "schema")
2059            .expect("Schema lock poisoned - a thread panicked while holding it");
2060        schema.edge_type_name_by_id_unified(type_id)
2061    }
2062
2063    pub fn add_property(
2064        &self,
2065        label_or_type: &str,
2066        prop_name: &str,
2067        data_type: DataType,
2068        nullable: bool,
2069    ) -> Result<()> {
2070        self.add_property_with_desc(label_or_type, prop_name, data_type, nullable, None)
2071    }
2072
2073    pub fn add_property_with_desc(
2074        &self,
2075        label_or_type: &str,
2076        prop_name: &str,
2077        data_type: DataType,
2078        nullable: bool,
2079        description: Option<String>,
2080    ) -> Result<()> {
2081        validate_property_name(prop_name)?;
2082        let mut guard = acquire_write(&self.schema, "schema")?;
2083        let schema = Arc::make_mut(&mut *guard);
2084        let version = schema.schema_version;
2085        let props = schema
2086            .properties
2087            .entry(label_or_type.to_string())
2088            .or_default();
2089
2090        if props.contains_key(prop_name) {
2091            return Err(anyhow!(
2092                "Property '{}' already exists for '{}'",
2093                prop_name,
2094                label_or_type
2095            ));
2096        }
2097
2098        props.insert(
2099            prop_name.to_string(),
2100            PropertyMeta {
2101                r#type: data_type,
2102                nullable,
2103                added_in: version,
2104                state: SchemaElementState::Active,
2105                generation_expression: None,
2106                description,
2107            },
2108        );
2109        // Bump after stamping `added_in` with the pre-bump `version`.
2110        schema.bump_version();
2111        Ok(())
2112    }
2113
2114    /// Declares a property, idempotent when an identical declaration already exists.
2115    ///
2116    /// The schema-builder counterpart to [`Self::add_property_with_desc`] (which
2117    /// hard-errors on *any* re-add, as DDL `ALTER` semantics require). Re-applying a
2118    /// schema is common — every `apply()` re-declares — so an existing property with
2119    /// the same `data_type` and `nullable` is a no-op (`Ok(false)`; a differing
2120    /// `description` is ignored as docs-only). A differing type or nullability is a
2121    /// hard conflict: silently swallowing it let `VECTOR(4)` be "re-declared" as
2122    /// `VECTOR(8)` while the column stayed 4-dimensional (issue #137).
2123    ///
2124    /// Returns `true` if this call newly inserted the property.
2125    ///
2126    /// # Errors
2127    /// Returns an error when the property exists with a different type or nullability
2128    /// (the message deliberately does not contain "already exists", which callers
2129    /// historically string-matched to ignore benign re-adds), or when the property
2130    /// name is invalid.
2131    pub fn declare_property(
2132        &self,
2133        label_or_type: &str,
2134        prop_name: &str,
2135        data_type: DataType,
2136        nullable: bool,
2137        description: Option<String>,
2138    ) -> Result<bool> {
2139        validate_property_name(prop_name)?;
2140        let mut guard = acquire_write(&self.schema, "schema")?;
2141        let schema = Arc::make_mut(&mut *guard);
2142        let version = schema.schema_version;
2143        let props = schema
2144            .properties
2145            .entry(label_or_type.to_string())
2146            .or_default();
2147
2148        if let Some(existing) = props.get(prop_name) {
2149            if existing.r#type == data_type && existing.nullable == nullable {
2150                return Ok(false); // identical re-declaration (idempotent)
2151            }
2152            return Err(anyhow!(
2153                "Property '{}' on '{}' is declared as {:?} (nullable: {}); cannot re-declare \
2154                 as {:?} (nullable: {}). Property types are immutable — use a new property \
2155                 name or migrate the data",
2156                prop_name,
2157                label_or_type,
2158                existing.r#type,
2159                existing.nullable,
2160                data_type,
2161                nullable
2162            ));
2163        }
2164
2165        props.insert(
2166            prop_name.to_string(),
2167            PropertyMeta {
2168                r#type: data_type,
2169                nullable,
2170                added_in: version,
2171                state: SchemaElementState::Active,
2172                generation_expression: None,
2173                description,
2174            },
2175        );
2176        // Bump after stamping `added_in` with the pre-bump `version`.
2177        schema.bump_version();
2178        Ok(true)
2179    }
2180
2181    /// Register an INTERNAL property (underscore-prefixed name allowed) that is
2182    /// materialised by the storage layer, not written by the user — e.g. the MUVERA
2183    /// `__fde_*` derived column. Bypasses the user-facing underscore-prefix rule but
2184    /// still rejects storage-layer name collisions. Idempotent: a no-op if the property
2185    /// already exists with the same type (so re-creating an index is safe).
2186    ///
2187    /// Returns `true` if this call newly inserted the property, `false` if it already
2188    /// existed (idempotent). The check-and-insert is atomic under the schema write lock,
2189    /// so for concurrent callers exactly one observes `true` — letting callers gate
2190    /// expensive one-time work (e.g. the MUVERA backfill) on the winner.
2191    pub fn add_internal_property(
2192        &self,
2193        label_or_type: &str,
2194        prop_name: &str,
2195        data_type: DataType,
2196        nullable: bool,
2197    ) -> Result<bool> {
2198        validate_reserved_property_name(prop_name)?;
2199        let mut guard = acquire_write(&self.schema, "schema")?;
2200        let schema = Arc::make_mut(&mut *guard);
2201        let version = schema.schema_version;
2202        let props = schema
2203            .properties
2204            .entry(label_or_type.to_string())
2205            .or_default();
2206
2207        if let Some(existing) = props.get(prop_name) {
2208            if existing.r#type == data_type {
2209                return Ok(false); // already present (idempotent re-registration)
2210            }
2211            return Err(anyhow!(
2212                "Internal property '{}' already exists for '{}' with a different type",
2213                prop_name,
2214                label_or_type
2215            ));
2216        }
2217
2218        props.insert(
2219            prop_name.to_string(),
2220            PropertyMeta {
2221                r#type: data_type,
2222                nullable,
2223                added_in: version,
2224                state: SchemaElementState::Active,
2225                generation_expression: None,
2226                description: None,
2227            },
2228        );
2229        schema.bump_version();
2230        Ok(true)
2231    }
2232
2233    pub fn add_generated_property(
2234        &self,
2235        label_or_type: &str,
2236        prop_name: &str,
2237        data_type: DataType,
2238        expr: String,
2239    ) -> Result<()> {
2240        // System-generated `_gen_*` columns bypass the underscore-prefix rule
2241        // but must still avoid storage-layer column-name collisions.
2242        validate_reserved_property_name(prop_name)?;
2243        let mut guard = acquire_write(&self.schema, "schema")?;
2244        let schema = Arc::make_mut(&mut *guard);
2245        let version = schema.schema_version;
2246        let props = schema
2247            .properties
2248            .entry(label_or_type.to_string())
2249            .or_default();
2250
2251        if props.contains_key(prop_name) {
2252            return Err(anyhow!("Property '{}' already exists", prop_name));
2253        }
2254
2255        props.insert(
2256            prop_name.to_string(),
2257            PropertyMeta {
2258                r#type: data_type,
2259                nullable: true,
2260                added_in: version,
2261                state: SchemaElementState::Active,
2262                generation_expression: Some(expr),
2263                description: None,
2264            },
2265        );
2266        // Bump after stamping `added_in` with the pre-bump `version`.
2267        schema.bump_version();
2268        Ok(())
2269    }
2270
2271    pub fn set_label_description(&self, name: &str, description: Option<String>) -> Result<()> {
2272        let mut guard = acquire_write(&self.schema, "schema")?;
2273        let schema = Arc::make_mut(&mut *guard);
2274        let meta = schema
2275            .labels
2276            .get_mut(name)
2277            .ok_or_else(|| anyhow!("Label '{}' does not exist", name))?;
2278        meta.description = description;
2279        Ok(())
2280    }
2281
2282    pub fn set_edge_type_description(&self, name: &str, description: Option<String>) -> Result<()> {
2283        let mut guard = acquire_write(&self.schema, "schema")?;
2284        let schema = Arc::make_mut(&mut *guard);
2285        let meta = schema
2286            .edge_types
2287            .get_mut(name)
2288            .ok_or_else(|| anyhow!("Edge type '{}' does not exist", name))?;
2289        meta.description = description;
2290        Ok(())
2291    }
2292
2293    pub fn set_property_description(
2294        &self,
2295        entity: &str,
2296        prop_name: &str,
2297        description: Option<String>,
2298    ) -> Result<()> {
2299        let mut guard = acquire_write(&self.schema, "schema")?;
2300        let schema = Arc::make_mut(&mut *guard);
2301        let props = schema
2302            .properties
2303            .get_mut(entity)
2304            .ok_or_else(|| anyhow!("Entity '{}' does not exist", entity))?;
2305        let meta = props
2306            .get_mut(prop_name)
2307            .ok_or_else(|| anyhow!("Property '{}' does not exist on '{}'", prop_name, entity))?;
2308        meta.description = description;
2309        Ok(())
2310    }
2311
2312    /// Register an index definition on the schema, **upsert by name**.
2313    ///
2314    /// If an index with the same `IndexDefinition::name()` already exists, it
2315    /// is replaced in place; otherwise the def is appended. Idempotent under
2316    /// repeat invocation, which makes `SchemaBuilder::apply()` re-applicable
2317    /// without bloating `schema.indexes` and lets the rebuild epilogue inside
2318    /// every `IndexManager::create_*_index` re-record metadata updates without
2319    /// duplicating entries (issue rustic-ai/uni-db#63).
2320    pub fn add_index(&self, index_def: IndexDefinition) -> Result<()> {
2321        let mut guard = acquire_write(&self.schema, "schema")?;
2322        let schema = Arc::make_mut(&mut *guard);
2323        if let Some(existing) = schema
2324            .indexes
2325            .iter_mut()
2326            .find(|i| i.name() == index_def.name())
2327        {
2328            *existing = index_def;
2329        } else {
2330            schema.indexes.push(index_def);
2331        }
2332        schema.bump_version();
2333        Ok(())
2334    }
2335
2336    pub fn get_index(&self, name: &str) -> Option<IndexDefinition> {
2337        let schema = self.schema.read().expect("Schema lock poisoned");
2338        schema.indexes.iter().find(|i| i.name() == name).cloned()
2339    }
2340
2341    /// Updates the lifecycle metadata for an index by name.
2342    ///
2343    /// The closure receives a mutable reference to the index's `IndexMetadata`,
2344    /// allowing callers to update status, timestamps, etc.
2345    pub fn update_index_metadata(
2346        &self,
2347        index_name: &str,
2348        f: impl FnOnce(&mut IndexMetadata),
2349    ) -> Result<()> {
2350        let mut guard = acquire_write(&self.schema, "schema")?;
2351        let schema = Arc::make_mut(&mut *guard);
2352        let idx = schema
2353            .indexes
2354            .iter_mut()
2355            .find(|i| i.name() == index_name)
2356            .ok_or_else(|| anyhow!("Index '{}' not found", index_name))?;
2357        f(idx.metadata_mut());
2358        Ok(())
2359    }
2360
2361    pub fn remove_index(&self, name: &str) -> Result<()> {
2362        let mut guard = acquire_write(&self.schema, "schema")?;
2363        let schema = Arc::make_mut(&mut *guard);
2364        if let Some(pos) = schema.indexes.iter().position(|i| i.name() == name) {
2365            schema.indexes.remove(pos);
2366            schema.bump_version();
2367            Ok(())
2368        } else {
2369            Err(anyhow!("Index '{}' not found", name))
2370        }
2371    }
2372
2373    pub fn add_constraint(&self, constraint: Constraint) -> Result<()> {
2374        let mut guard = acquire_write(&self.schema, "schema")?;
2375        let schema = Arc::make_mut(&mut *guard);
2376        if schema.constraints.iter().any(|c| c.name == constraint.name) {
2377            return Err(anyhow!("Constraint '{}' already exists", constraint.name));
2378        }
2379        schema.constraints.push(constraint);
2380        schema.bump_version();
2381        Ok(())
2382    }
2383
2384    pub fn drop_constraint(&self, name: &str, if_exists: bool) -> Result<()> {
2385        let mut guard = acquire_write(&self.schema, "schema")?;
2386        let schema = Arc::make_mut(&mut *guard);
2387        if let Some(pos) = schema.constraints.iter().position(|c| c.name == name) {
2388            schema.constraints.remove(pos);
2389            schema.bump_version();
2390            Ok(())
2391        } else if if_exists {
2392            Ok(())
2393        } else {
2394            Err(anyhow!("Constraint '{}' not found", name))
2395        }
2396    }
2397
2398    pub fn drop_property(&self, label_or_type: &str, prop_name: &str) -> Result<()> {
2399        let mut guard = acquire_write(&self.schema, "schema")?;
2400        let schema = Arc::make_mut(&mut *guard);
2401        let Some(props) = schema.properties.get_mut(label_or_type) else {
2402            return Err(anyhow!("Label or Edge Type '{}' not found", label_or_type));
2403        };
2404        if props.remove(prop_name).is_none() {
2405            return Err(anyhow!(
2406                "Property '{}' not found for '{}'",
2407                prop_name,
2408                label_or_type
2409            ));
2410        }
2411        schema.bump_version();
2412        Ok(())
2413    }
2414
2415    pub fn rename_property(
2416        &self,
2417        label_or_type: &str,
2418        old_name: &str,
2419        new_name: &str,
2420    ) -> Result<()> {
2421        // Validate the new name like declare_property/add_property do — otherwise
2422        // a rename bypasses the reserved-storage-column guard and the leading-
2423        // underscore rule, letting a user property collide with an internal Arrow
2424        // column (e.g. `_vid`, `src_vid`, `overflow_json`).
2425        validate_property_name(new_name)?;
2426        let mut guard = acquire_write(&self.schema, "schema")?;
2427        let schema = Arc::make_mut(&mut *guard);
2428        let Some(props) = schema.properties.get_mut(label_or_type) else {
2429            return Err(anyhow!("Label or Edge Type '{}' not found", label_or_type));
2430        };
2431        let Some(meta) = props.remove(old_name) else {
2432            return Err(anyhow!(
2433                "Property '{}' not found for '{}'",
2434                old_name,
2435                label_or_type
2436            ));
2437        };
2438        if props.contains_key(new_name) {
2439            // Rollback removal? Or just error.
2440            props.insert(old_name.to_string(), meta); // Restore
2441            return Err(anyhow!("Property '{}' already exists", new_name));
2442        }
2443        props.insert(new_name.to_string(), meta);
2444        schema.bump_version();
2445        Ok(())
2446    }
2447
2448    pub fn drop_label(&self, name: &str, if_exists: bool) -> Result<()> {
2449        let mut guard = acquire_write(&self.schema, "schema")?;
2450        let schema = Arc::make_mut(&mut *guard);
2451        if let Some(label_meta) = schema.labels.get_mut(name) {
2452            label_meta.state = SchemaElementState::Tombstone { since: Utc::now() };
2453            // Do not remove properties; they are implicitly tombstoned by the label
2454            schema.bump_version();
2455            Ok(())
2456        } else if if_exists {
2457            Ok(())
2458        } else {
2459            Err(anyhow!("Label '{}' not found", name))
2460        }
2461    }
2462
2463    pub fn drop_edge_type(&self, name: &str, if_exists: bool) -> Result<()> {
2464        let mut guard = acquire_write(&self.schema, "schema")?;
2465        let schema = Arc::make_mut(&mut *guard);
2466        if let Some(edge_meta) = schema.edge_types.get_mut(name) {
2467            edge_meta.state = SchemaElementState::Tombstone { since: Utc::now() };
2468            // Do not remove properties; they are implicitly tombstoned by the edge type
2469            schema.bump_version();
2470            Ok(())
2471        } else if if_exists {
2472            Ok(())
2473        } else {
2474            Err(anyhow!("Edge Type '{}' not found", name))
2475        }
2476    }
2477}
2478
2479/// Validate identifier names to prevent injection and ensure compatibility.
2480pub fn validate_identifier(name: &str) -> Result<()> {
2481    // Length check
2482    if name.is_empty() || name.len() > 64 {
2483        return Err(anyhow!("Identifier '{}' must be 1-64 characters", name));
2484    }
2485
2486    // First character must be letter or underscore
2487    let first = name.chars().next().unwrap();
2488    if !first.is_alphabetic() && first != '_' {
2489        return Err(anyhow!(
2490            "Identifier '{}' must start with letter or underscore",
2491            name
2492        ));
2493    }
2494
2495    // Remaining characters: alphanumeric or underscore
2496    if !name.chars().all(|c| c.is_alphanumeric() || c == '_') {
2497        return Err(anyhow!(
2498            "Identifier '{}' must contain only alphanumeric and underscore",
2499            name
2500        ));
2501    }
2502
2503    // Reserved words
2504    const RESERVED: &[&str] = &[
2505        "MATCH", "CREATE", "DELETE", "SET", "RETURN", "WHERE", "MERGE", "CALL", "YIELD", "WITH",
2506        "UNION", "ORDER", "LIMIT",
2507    ];
2508    if RESERVED.contains(&name.to_uppercase().as_str()) {
2509        return Err(anyhow!("Identifier '{}' cannot be a reserved word", name));
2510    }
2511
2512    Ok(())
2513}
2514
2515/// Reject user-declared property names that collide with internal Arrow column
2516/// names used by the storage layer.
2517///
2518/// Without this, declaring a property named e.g. `ext_id` produces an Arrow
2519/// schema with two `ext_id` fields at flush time, which Lance rejects with
2520/// "Duplicate field name" — silently losing all in-session writes on shutdown.
2521pub fn validate_property_name(name: &str) -> Result<()> {
2522    if name.starts_with('_') {
2523        return Err(anyhow!(
2524            "Property name '{}' is reserved: names starting with '_' are reserved by the storage layer",
2525            name
2526        ));
2527    }
2528    validate_reserved_property_name(name)
2529}
2530
2531/// Reject names that collide with storage-layer Arrow column names.
2532///
2533/// Used both by `validate_property_name` (user-facing path) and directly by
2534/// `add_generated_property` (system-generated `_gen_*` path) — the latter
2535/// needs to bypass the underscore-prefix rule but must still reject the
2536/// fixed-name collisions below.
2537fn validate_reserved_property_name(name: &str) -> Result<()> {
2538    // Unprefixed names that get appended alongside user properties in the
2539    // per-label vertex (`storage/vertex.rs`), per-edge-type edge
2540    // (`storage/edge.rs`), or per-edge-type delta (`storage/delta.rs`)
2541    // Arrow schemas — declaring one of these as a user property produces a
2542    // duplicate Arrow field and a Lance "Duplicate field name" error at
2543    // flush time. Fixed-schema-only columns (`type`, `props_json`,
2544    // `labels` in the main tables) are NOT listed: those tables don't
2545    // append user properties, so no collision can occur.
2546    const RESERVED_PROPS: &[&str] = &[
2547        "ext_id",
2548        "overflow_json",
2549        "eid",
2550        "src_vid",
2551        "dst_vid",
2552        "op",
2553        // Internal planner sentinel: a column-name marker used by
2554        // `mark_set_item_variables` (uni-query::query::planner) to request
2555        // narrow structural projection without full-schema expansion.
2556        // Reserved here defensively so an internal `add_generated_property`
2557        // path can't accidentally create a colliding user-facing column.
2558        // The user-facing `validate_property_name` already rejects this
2559        // via the underscore-prefix rule, so this is belt-and-suspenders.
2560        "__set_struct__",
2561    ];
2562    if RESERVED_PROPS.contains(&name) {
2563        return Err(anyhow!(
2564            "Property name '{}' is reserved by the storage layer; please choose a different name",
2565            name
2566        ));
2567    }
2568    Ok(())
2569}
2570
2571#[cfg(test)]
2572mod tests {
2573    use super::*;
2574    use crate::value::{TemporalValue, Value};
2575    use object_store::local::LocalFileSystem;
2576    use tempfile::tempdir;
2577
2578    #[test]
2579    fn binary_vector_metrics_exact() {
2580        // Hamming = number of differing bits. 0x00 vs 0xFF = 8 bits; 0xA5 vs 0xA5
2581        // = 0; 0x0F vs 0x00 = 4 bits.
2582        assert_eq!(
2583            DistanceMetric::Hamming.compute_distance_binary(&[0x00], &[0xFF]),
2584            8.0
2585        );
2586        assert_eq!(
2587            DistanceMetric::Hamming.compute_distance_binary(&[0xA5, 0x0F], &[0xA5, 0x00]),
2588            4.0
2589        );
2590        assert_eq!(
2591            DistanceMetric::Hamming.compute_distance_binary(&[0xA5], &[0xA5]),
2592            0.0
2593        );
2594
2595        // Jaccard = 1 − |A∩B|/|A∪B|. 0b1100 & 0b1010 = 0b1000 (1 bit);
2596        // 0b1100 | 0b1010 = 0b1110 (3 bits) → 1 − 1/3 = 2/3.
2597        let j = DistanceMetric::Jaccard.compute_distance_binary(&[0b1100], &[0b1010]);
2598        assert!((j - (2.0 / 3.0)).abs() < 1e-6, "got {j}");
2599        // Identical vectors → distance 0.
2600        assert_eq!(
2601            DistanceMetric::Jaccard.compute_distance_binary(&[0xFF], &[0xFF]),
2602            0.0
2603        );
2604        // Two all-zero vectors are defined as distance 0 (empty union).
2605        assert_eq!(
2606            DistanceMetric::Jaccard.compute_distance_binary(&[0x00, 0x00], &[0x00, 0x00]),
2607            0.0
2608        );
2609    }
2610
2611    #[test]
2612    fn binary_metrics_are_binary_and_route_correctly() {
2613        assert!(DistanceMetric::Hamming.is_binary());
2614        assert!(DistanceMetric::Jaccard.is_binary());
2615        assert!(!DistanceMetric::L2.is_binary());
2616        assert!(!DistanceMetric::L1.is_binary());
2617    }
2618
2619    #[test]
2620    #[should_panic(expected = "binary-vector metric")]
2621    fn float_compute_distance_rejects_binary_metric() {
2622        DistanceMetric::Hamming.compute_distance(&[1.0], &[0.0]);
2623    }
2624
2625    #[test]
2626    fn check_binary_vector_value_guards() {
2627        let ty = DataType::BinaryVector { dimensions: 3 };
2628        assert!(
2629            ty.check_vector_dims(&Value::BinaryVector(vec![1, 2, 3]))
2630                .is_ok()
2631        );
2632        assert!(ty.check_vector_dims(&Value::Null).is_ok());
2633        // Wrong lane count.
2634        assert!(
2635            ty.check_vector_dims(&Value::BinaryVector(vec![1, 2]))
2636                .is_err()
2637        );
2638        // List of byte-ints is the literal form.
2639        assert!(
2640            ty.check_vector_dims(&Value::List(vec![
2641                Value::Int(0),
2642                Value::Int(255),
2643                Value::Int(128)
2644            ]))
2645            .is_ok()
2646        );
2647        // Out-of-byte-range element.
2648        assert!(
2649            ty.check_vector_dims(&Value::List(vec![
2650                Value::Int(0),
2651                Value::Int(256),
2652                Value::Int(1)
2653            ]))
2654            .is_err()
2655        );
2656    }
2657
2658    #[test]
2659    fn test_datatype_accepts_matrix() {
2660        let dt = || TemporalValue::DateTime {
2661            nanos_since_epoch: 0,
2662            offset_seconds: 0,
2663            timezone_name: None,
2664        };
2665
2666        // Null is accepted by every type (nullability checked separately).
2667        for ty in [
2668            DataType::String,
2669            DataType::Int64,
2670            DataType::Bool,
2671            DataType::DateTime,
2672            DataType::Float64,
2673        ] {
2674            assert!(ty.accepts(&Value::Null), "{ty:?} must accept Null");
2675        }
2676
2677        // Exact-type matches.
2678        assert!(DataType::String.accepts(&Value::String("x".into())));
2679        assert!(DataType::Int64.accepts(&Value::Int(1)));
2680        assert!(DataType::Bool.accepts(&Value::Bool(true)));
2681        assert!(DataType::DateTime.accepts(&Value::Temporal(dt())));
2682
2683        // Intentional lossless widenings remain allowed.
2684        assert!(
2685            DataType::Float64.accepts(&Value::Int(3)),
2686            "Int widens to Float"
2687        );
2688        assert!(DataType::Int32.accepts(&Value::Int(3)), "Int fits Int32");
2689        assert!(DataType::Timestamp.accepts(&Value::Temporal(dt())));
2690        assert!(
2691            DataType::Timestamp.accepts(&Value::String("2026-01-01T00:00:00Z".into())),
2692            "storage parses strings for non-struct Timestamp columns"
2693        );
2694
2695        // The #68 data-loss cases must be rejected (coercion handles strings separately).
2696        assert!(
2697            !DataType::DateTime.accepts(&Value::String("2026-01-01T00:00:00Z".into())),
2698            "String into a DateTime struct column nulls silently — reject here"
2699        );
2700        assert!(!DataType::Bool.accepts(&Value::Int(1)));
2701        assert!(!DataType::Int64.accepts(&Value::Bool(true)));
2702        assert!(!DataType::Int64.accepts(&Value::Float(1.5)));
2703        assert!(
2704            !DataType::String.accepts(&Value::Int(10)),
2705            "no implicit stringification"
2706        );
2707        assert!(!DataType::Duration.accepts(&Value::String("P1D".into())));
2708
2709        // Opaque columns accept anything.
2710        assert!(DataType::CypherValue.accepts(&Value::Map(Default::default())));
2711    }
2712
2713    #[test]
2714    fn test_check_vector_dims_matrix() {
2715        let vec3 = DataType::Vector { dimensions: 3 };
2716        let multi2 = DataType::List(Box::new(DataType::Vector { dimensions: 2 }));
2717        let flist = |vals: &[f64]| Value::List(vals.iter().map(|f| Value::Float(*f)).collect());
2718
2719        // Null is accepted everywhere (nullability enforced separately).
2720        assert!(vec3.check_vector_dims(&Value::Null).is_ok());
2721        assert!(multi2.check_vector_dims(&Value::Null).is_ok());
2722
2723        // Correct-dimension values pass; Int elements are numeric.
2724        assert!(
2725            vec3.check_vector_dims(&Value::Vector(vec![1.0, 2.0, 3.0]))
2726                .is_ok()
2727        );
2728        assert!(vec3.check_vector_dims(&flist(&[1.0, 2.0, 3.0])).is_ok());
2729        assert!(
2730            vec3.check_vector_dims(&Value::List(vec![
2731                Value::Int(1),
2732                Value::Float(2.0),
2733                Value::Int(3)
2734            ]))
2735            .is_ok()
2736        );
2737
2738        // The #137 cases: wrong length, empty list, non-numeric element, wrong shape.
2739        assert_eq!(
2740            vec3.check_vector_dims(&Value::Vector(vec![1.0, 2.0])),
2741            Err(VectorDimError::WrongLength {
2742                expected: 3,
2743                actual: 2
2744            })
2745        );
2746        assert_eq!(
2747            vec3.check_vector_dims(&flist(&[1.0, 2.0, 3.0, 4.0, 5.0])),
2748            Err(VectorDimError::WrongLength {
2749                expected: 3,
2750                actual: 5
2751            })
2752        );
2753        assert_eq!(
2754            vec3.check_vector_dims(&Value::List(vec![])),
2755            Err(VectorDimError::WrongLength {
2756                expected: 3,
2757                actual: 0
2758            })
2759        );
2760        assert_eq!(
2761            vec3.check_vector_dims(&Value::List(vec![
2762                Value::Float(1.0),
2763                Value::String("x".into()),
2764                Value::Float(3.0),
2765            ])),
2766            Err(VectorDimError::NonNumericElement { index: 1 })
2767        );
2768        assert_eq!(
2769            vec3.check_vector_dims(&Value::List(vec![
2770                Value::Float(1.0),
2771                Value::Null,
2772                Value::Float(3.0)
2773            ])),
2774            Err(VectorDimError::NonNumericElement { index: 1 })
2775        );
2776        assert_eq!(
2777            vec3.check_vector_dims(&Value::String("not a vector".into())),
2778            Err(VectorDimError::NotAVector { actual: "String" })
2779        );
2780
2781        // Multi-vector: empty token list is a legal empty multi-vector; each
2782        // token must match the declared per-token dimensions.
2783        assert!(multi2.check_vector_dims(&Value::List(vec![])).is_ok());
2784        assert!(
2785            multi2
2786                .check_vector_dims(&Value::List(vec![flist(&[1.0, 2.0]), flist(&[3.0, 4.0])]))
2787                .is_ok()
2788        );
2789        assert_eq!(
2790            multi2.check_vector_dims(&Value::List(vec![
2791                flist(&[1.0, 2.0]),
2792                flist(&[9.0, 9.0, 9.0])
2793            ])),
2794            Err(VectorDimError::TokenWrongLength {
2795                token: 1,
2796                expected: 2,
2797                actual: 3
2798            })
2799        );
2800        assert_eq!(
2801            multi2.check_vector_dims(&Value::List(vec![Value::String("tok".into())])),
2802            Err(VectorDimError::TokenNotAVector {
2803                token: 0,
2804                actual: "String"
2805            })
2806        );
2807        assert_eq!(
2808            multi2.check_vector_dims(&Value::Vector(vec![1.0, 2.0])),
2809            Err(VectorDimError::NotATokenList { actual: "Vector" })
2810        );
2811
2812        // Non-vector declared types never object, so callers may check unconditionally.
2813        assert!(
2814            DataType::Int64
2815                .check_vector_dims(&Value::String("x".into()))
2816                .is_ok()
2817        );
2818        assert!(
2819            DataType::List(Box::new(DataType::Float64))
2820                .check_vector_dims(&Value::List(vec![Value::String("x".into())]))
2821                .is_ok()
2822        );
2823        assert!(
2824            DataType::SparseVector { dimensions: 8 }
2825                .check_vector_dims(&Value::Map(Default::default()))
2826                .is_ok()
2827        );
2828
2829        // Error rendering carries both lengths so write errors are actionable.
2830        let msg = VectorDimError::WrongLength {
2831            expected: 4,
2832            actual: 5,
2833        }
2834        .to_string();
2835        assert!(msg.contains('4') && msg.contains('5'), "message: {msg}");
2836    }
2837
2838    #[tokio::test]
2839    async fn test_declare_property_idempotent_and_conflicting() -> Result<()> {
2840        let dir = tempdir()?;
2841        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2842        let path = ObjectStorePath::from("schema.json");
2843        let manager = SchemaManager::load_from_store(store.clone(), &path).await?;
2844
2845        manager.add_label("Doc")?;
2846        let vec4 = DataType::Vector { dimensions: 4 };
2847
2848        // First declaration inserts.
2849        assert!(manager.declare_property("Doc", "embedding", vec4.clone(), true, None)?);
2850
2851        // Identical re-declaration is an idempotent no-op — the register-on-every-open
2852        // pattern; a differing description is docs-only and also ignored.
2853        assert!(!manager.declare_property("Doc", "embedding", vec4.clone(), true, None)?);
2854        assert!(!manager.declare_property(
2855            "Doc",
2856            "embedding",
2857            vec4.clone(),
2858            true,
2859            Some("new docs".into())
2860        )?);
2861
2862        // A dimension change is a conflict (#137 case c), and the message must not
2863        // contain "already exists" (historically string-matched and swallowed).
2864        let err = manager
2865            .declare_property(
2866                "Doc",
2867                "embedding",
2868                DataType::Vector { dimensions: 8 },
2869                true,
2870                None,
2871            )
2872            .unwrap_err()
2873            .to_string();
2874        assert!(err.contains('4') && err.contains('8'), "message: {err}");
2875        assert!(!err.contains("already exists"), "message: {err}");
2876
2877        // Nullability flips are conflicts too — they change NOT NULL enforcement.
2878        assert!(
2879            manager
2880                .declare_property("Doc", "embedding", vec4.clone(), false, None)
2881                .is_err()
2882        );
2883
2884        // The schema still holds the original declaration.
2885        let schema = manager.schema();
2886        let meta = &schema.properties["Doc"]["embedding"];
2887        assert_eq!(meta.r#type, vec4);
2888        assert!(meta.nullable);
2889        Ok(())
2890    }
2891
2892    #[tokio::test]
2893    async fn test_schema_management() -> Result<()> {
2894        let dir = tempdir()?;
2895        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2896        let path = ObjectStorePath::from("schema.json");
2897        let manager = SchemaManager::load_from_store(store.clone(), &path).await?;
2898
2899        // Labels
2900        let lid = manager.add_label("Person")?;
2901        assert_eq!(lid, 1);
2902        assert!(manager.add_label("Person").is_err());
2903
2904        // Properties
2905        manager.add_property("Person", "name", DataType::String, false)?;
2906        assert!(
2907            manager
2908                .add_property("Person", "name", DataType::String, false)
2909                .is_err()
2910        );
2911
2912        // Edge types
2913        let tid = manager.add_edge_type("knows", vec!["Person".into()], vec!["Person".into()])?;
2914        assert_eq!(tid, 1);
2915
2916        manager.save().await?;
2917        // Check file exists
2918        assert!(store.get(&path).await.is_ok());
2919
2920        let manager2 = SchemaManager::load_from_store(store, &path).await?;
2921        assert!(manager2.schema().labels.contains_key("Person"));
2922        assert!(
2923            manager2
2924                .schema()
2925                .properties
2926                .get("Person")
2927                .unwrap()
2928                .contains_key("name")
2929        );
2930
2931        Ok(())
2932    }
2933
2934    #[tokio::test]
2935    async fn test_reserved_property_names_rejected() -> Result<()> {
2936        let dir = tempdir()?;
2937        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
2938        let path = ObjectStorePath::from("schema.json");
2939        let manager = SchemaManager::load_from_store(store, &path).await?;
2940
2941        manager.add_label("Tiny")?;
2942
2943        // Unprefixed reserved names — these collide with internal Arrow
2944        // columns in storage tables and previously caused Lance
2945        // "Duplicate field name" errors at flush time.
2946        for reserved in &["ext_id", "overflow_json", "eid", "src_vid", "dst_vid", "op"] {
2947            let err = manager
2948                .add_property("Tiny", reserved, DataType::String, true)
2949                .expect_err(&format!("expected '{reserved}' to be rejected"));
2950            assert!(
2951                err.to_string().contains("reserved"),
2952                "error for '{reserved}' should mention 'reserved', got: {err}"
2953            );
2954        }
2955
2956        // Planner sentinel — reserved in RESERVED_PROPS (belt-and-suspenders
2957        // alongside the underscore-prefix rule). Confirms an internal
2958        // `add_generated_property` path cannot accidentally create a column
2959        // that collides with the SET-target structural-projection marker.
2960        let err = manager
2961            .add_property("Tiny", "__set_struct__", DataType::String, true)
2962            .expect_err("expected '__set_struct__' to be rejected");
2963        assert!(
2964            err.to_string().contains("reserved"),
2965            "__set_struct__ rejection should mention 'reserved', got: {err}"
2966        );
2967
2968        // Leading-underscore pattern rule.
2969        for reserved in &["_vid", "_uid", "_eid", "_version", "_created_at"] {
2970            assert!(
2971                manager
2972                    .add_property("Tiny", reserved, DataType::String, true)
2973                    .is_err(),
2974                "expected '{reserved}' to be rejected"
2975            );
2976        }
2977
2978        // Names that merely contain a reserved substring should still be
2979        // accepted.
2980        manager.add_property("Tiny", "ext_id_foo", DataType::String, true)?;
2981        manager.add_property("Tiny", "user_op", DataType::String, true)?;
2982        manager.add_property("Tiny", "type_name", DataType::String, true)?;
2983
2984        // Same check applies to edge-type properties (single dispatch).
2985        manager.add_edge_type("knows", vec!["Tiny".into()], vec!["Tiny".into()])?;
2986        assert!(
2987            manager
2988                .add_property("knows", "src_vid", DataType::Int64, true)
2989                .is_err()
2990        );
2991
2992        // And to generated properties.
2993        assert!(
2994            manager
2995                .add_generated_property(
2996                    "Tiny",
2997                    "ext_id",
2998                    DataType::String,
2999                    "concat('x', name)".into()
3000                )
3001                .is_err()
3002        );
3003
3004        Ok(())
3005    }
3006
3007    #[test]
3008    fn test_normalize_function_names() {
3009        assert_eq!(
3010            SchemaManager::normalize_function_names("lower(email)"),
3011            "LOWER(email)"
3012        );
3013        assert_eq!(
3014            SchemaManager::normalize_function_names("LOWER(email)"),
3015            "LOWER(email)"
3016        );
3017        assert_eq!(
3018            SchemaManager::normalize_function_names("Lower(email)"),
3019            "LOWER(email)"
3020        );
3021        assert_eq!(
3022            SchemaManager::normalize_function_names("trim(lower(email))"),
3023            "TRIM(LOWER(email))"
3024        );
3025    }
3026
3027    #[test]
3028    fn test_generated_column_name_case_insensitive() {
3029        let col1 = SchemaManager::generated_column_name("lower(email)");
3030        let col2 = SchemaManager::generated_column_name("LOWER(email)");
3031        let col3 = SchemaManager::generated_column_name("Lower(email)");
3032        assert_eq!(col1, col2);
3033        assert_eq!(col2, col3);
3034        assert!(col1.starts_with("_gen_LOWER_email_"));
3035    }
3036
3037    #[test]
3038    fn test_index_metadata_serde_backward_compat() {
3039        // Simulate old JSON without metadata field
3040        let json = r#"{
3041            "type": "Scalar",
3042            "name": "idx_person_name",
3043            "label": "Person",
3044            "properties": ["name"],
3045            "index_type": "BTree",
3046            "where_clause": null
3047        }"#;
3048        let def: IndexDefinition = serde_json::from_str(json).unwrap();
3049        let meta = def.metadata();
3050        assert_eq!(meta.status, IndexStatus::Online);
3051        assert!(meta.last_built_at.is_none());
3052        assert!(meta.row_count_at_build.is_none());
3053    }
3054
3055    #[test]
3056    fn test_index_metadata_serde_roundtrip() {
3057        let now = Utc::now();
3058        let def = IndexDefinition::Scalar(ScalarIndexConfig {
3059            name: "idx_test".to_string(),
3060            label: "Test".to_string(),
3061            properties: vec!["prop".to_string()],
3062            index_type: ScalarIndexType::BTree,
3063            where_clause: None,
3064            metadata: IndexMetadata {
3065                status: IndexStatus::Building,
3066                last_built_at: Some(now),
3067                row_count_at_build: Some(42),
3068            },
3069        });
3070
3071        let json = serde_json::to_string(&def).unwrap();
3072        let parsed: IndexDefinition = serde_json::from_str(&json).unwrap();
3073        assert_eq!(parsed.metadata().status, IndexStatus::Building);
3074        assert_eq!(parsed.metadata().row_count_at_build, Some(42));
3075        assert!(parsed.metadata().last_built_at.is_some());
3076    }
3077
3078    #[tokio::test]
3079    async fn test_update_index_metadata() -> Result<()> {
3080        let dir = tempdir()?;
3081        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3082        let path = ObjectStorePath::from("schema.json");
3083        let manager = SchemaManager::load_from_store(store, &path).await?;
3084
3085        manager.add_label("Person")?;
3086        let idx = IndexDefinition::Scalar(ScalarIndexConfig {
3087            name: "idx_test".to_string(),
3088            label: "Person".to_string(),
3089            properties: vec!["name".to_string()],
3090            index_type: ScalarIndexType::BTree,
3091            where_clause: None,
3092            metadata: Default::default(),
3093        });
3094        manager.add_index(idx)?;
3095
3096        // Verify initial status is Online
3097        let initial = manager.get_index("idx_test").unwrap();
3098        assert_eq!(initial.metadata().status, IndexStatus::Online);
3099
3100        // Update to Building
3101        manager.update_index_metadata("idx_test", |m| {
3102            m.status = IndexStatus::Building;
3103            m.row_count_at_build = Some(100);
3104        })?;
3105
3106        let updated = manager.get_index("idx_test").unwrap();
3107        assert_eq!(updated.metadata().status, IndexStatus::Building);
3108        assert_eq!(updated.metadata().row_count_at_build, Some(100));
3109
3110        // Non-existent index should error
3111        assert!(manager.update_index_metadata("nope", |_| {}).is_err());
3112
3113        Ok(())
3114    }
3115
3116    /// `add_internal_property` reports whether THIS call inserted the property: `true` on
3117    /// first insert, `false` on idempotent re-registration, `Err` on a type conflict. The
3118    /// MUVERA backfill gates on this (only the inserter backfills), so two concurrent
3119    /// creates of the same index can't both run the full-table rewrite (issue #107).
3120    #[tokio::test]
3121    async fn add_internal_property_reports_newly_added() -> Result<()> {
3122        let dir = tempdir()?;
3123        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3124        let path = ObjectStorePath::from("schema.json");
3125        let manager = SchemaManager::load_from_store(store, &path).await?;
3126        manager.add_label("Doc")?;
3127
3128        let dt = DataType::Vector { dimensions: 16 };
3129        // First registration: newly added.
3130        assert!(manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
3131        // Idempotent re-registration with the same type: NOT newly added.
3132        assert!(!manager.add_internal_property("Doc", "__fde_x", dt.clone(), true)?);
3133        // Same name, conflicting type: hard error (no silent divergence).
3134        assert!(
3135            manager
3136                .add_internal_property("Doc", "__fde_x", DataType::Vector { dimensions: 8 }, true)
3137                .is_err()
3138        );
3139        Ok(())
3140    }
3141
3142    /// `add_index` is upsert-by-name (issue rustic-ai/uni-db#63). Repeat
3143    /// invocations with the same `IndexDefinition::name()` must replace
3144    /// the entry in place rather than appending. Subsequent `add_index`
3145    /// calls also reflect metadata updates from the new definition.
3146    #[tokio::test]
3147    async fn test_add_index_is_upsert_by_name() -> Result<()> {
3148        let dir = tempdir()?;
3149        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3150        let path = ObjectStorePath::from("schema.json");
3151        let manager = SchemaManager::load_from_store(store, &path).await?;
3152        manager.add_label("Person")?;
3153
3154        let initial = IndexDefinition::Scalar(ScalarIndexConfig {
3155            name: "idx_test".to_string(),
3156            label: "Person".to_string(),
3157            properties: vec!["name".to_string()],
3158            index_type: ScalarIndexType::BTree,
3159            where_clause: None,
3160            metadata: IndexMetadata {
3161                status: IndexStatus::Building,
3162                ..Default::default()
3163            },
3164        });
3165        manager.add_index(initial.clone())?;
3166        assert_eq!(manager.schema().indexes.len(), 1);
3167
3168        // Re-add the identical def — must remain a single entry.
3169        manager.add_index(initial.clone())?;
3170        assert_eq!(
3171            manager.schema().indexes.len(),
3172            1,
3173            "duplicate add_index by name must not append"
3174        );
3175
3176        // Re-add with updated metadata — must replace in place, len unchanged.
3177        let mut updated_cfg = match initial {
3178            IndexDefinition::Scalar(c) => c,
3179            _ => unreachable!(),
3180        };
3181        updated_cfg.metadata.status = IndexStatus::Online;
3182        updated_cfg.metadata.row_count_at_build = Some(42);
3183        manager.add_index(IndexDefinition::Scalar(updated_cfg))?;
3184        assert_eq!(manager.schema().indexes.len(), 1);
3185        let stored = manager.get_index("idx_test").unwrap();
3186        assert_eq!(stored.metadata().status, IndexStatus::Online);
3187        assert_eq!(stored.metadata().row_count_at_build, Some(42));
3188
3189        // A *different* name appends as a new entry.
3190        let other = IndexDefinition::Scalar(ScalarIndexConfig {
3191            name: "idx_other".to_string(),
3192            label: "Person".to_string(),
3193            properties: vec!["age".to_string()],
3194            index_type: ScalarIndexType::BTree,
3195            where_clause: None,
3196            metadata: IndexMetadata::default(),
3197        });
3198        manager.add_index(other)?;
3199        assert_eq!(manager.schema().indexes.len(), 2);
3200
3201        Ok(())
3202    }
3203
3204    /// `load_from_store` self-heals catalogs that were bloated by the
3205    /// pre-fix `add_index` (kept the *last* def per name).
3206    #[tokio::test]
3207    async fn test_load_dedups_bloated_indexes() -> Result<()> {
3208        let dir = tempdir()?;
3209        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3210        let path = ObjectStorePath::from("schema.json");
3211
3212        // Seed disk with a hand-crafted bloated schema: 50 entries, all
3213        // sharing the same name. The last entry has distinct metadata so
3214        // we can assert "last writer wins" semantics.
3215        let mut schema = Schema::default();
3216        schema.labels.insert(
3217            "Person".to_string(),
3218            LabelMeta {
3219                id: 1,
3220                created_at: chrono::Utc::now(),
3221                state: SchemaElementState::Active,
3222                description: None,
3223            },
3224        );
3225        let make = |status: IndexStatus, count: Option<u64>| {
3226            IndexDefinition::Scalar(ScalarIndexConfig {
3227                name: "idx_dup".to_string(),
3228                label: "Person".to_string(),
3229                properties: vec!["name".to_string()],
3230                index_type: ScalarIndexType::BTree,
3231                where_clause: None,
3232                metadata: IndexMetadata {
3233                    status,
3234                    row_count_at_build: count,
3235                    ..Default::default()
3236                },
3237            })
3238        };
3239        for _ in 0..49 {
3240            schema.indexes.push(make(IndexStatus::Building, None));
3241        }
3242        schema.indexes.push(make(IndexStatus::Online, Some(123)));
3243        let json = serde_json::to_string_pretty(&schema)?;
3244        store.put(&path, json.into()).await?;
3245
3246        let manager = SchemaManager::load_from_store(store, &path).await?;
3247        let schema = manager.schema();
3248        assert_eq!(
3249            schema.indexes.len(),
3250            1,
3251            "load() must collapse 50 duplicates by name to 1"
3252        );
3253        // Last-writer-wins: the kept entry is the final push (Online, 123).
3254        assert_eq!(schema.indexes[0].metadata().status, IndexStatus::Online);
3255        assert_eq!(schema.indexes[0].metadata().row_count_at_build, Some(123));
3256
3257        Ok(())
3258    }
3259
3260    #[test]
3261    fn test_vector_index_for_property_skips_non_online() {
3262        let mut schema = Schema::default();
3263        schema.labels.insert(
3264            "Document".to_string(),
3265            LabelMeta {
3266                id: 1,
3267                created_at: chrono::Utc::now(),
3268                state: SchemaElementState::Active,
3269                description: None,
3270            },
3271        );
3272
3273        // Add a vector index with Stale status
3274        schema
3275            .indexes
3276            .push(IndexDefinition::Vector(VectorIndexConfig {
3277                name: "vec_doc_embedding".to_string(),
3278                label: "Document".to_string(),
3279                property: "embedding".to_string(),
3280                index_type: VectorIndexType::Flat,
3281                metric: DistanceMetric::Cosine,
3282                embedding_config: None,
3283                metadata: IndexMetadata {
3284                    status: IndexStatus::Stale,
3285                    ..Default::default()
3286                },
3287            }));
3288
3289        // Stale index should NOT be returned
3290        assert!(
3291            schema
3292                .vector_index_for_property("Document", "embedding")
3293                .is_none()
3294        );
3295
3296        // Set to Online — should now be returned
3297        if let IndexDefinition::Vector(cfg) = &mut schema.indexes[0] {
3298            cfg.metadata.status = IndexStatus::Online;
3299        }
3300        let result = schema.vector_index_for_property("Document", "embedding");
3301        assert!(result.is_some());
3302        assert_eq!(result.unwrap().metric, DistanceMetric::Cosine);
3303    }
3304
3305    #[tokio::test]
3306    async fn with_overlay_empty_clones_primary_in_isolation() -> Result<()> {
3307        use crate::core::fork::SchemaDelta;
3308
3309        let dir = tempdir()?;
3310        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3311        let path = ObjectStorePath::from("schema.json");
3312        let primary = SchemaManager::load_from_store(store, &path).await?;
3313        primary.add_label("Person")?;
3314
3315        let overlay = primary.with_overlay(&SchemaDelta::empty());
3316        assert_eq!(overlay.schema().labels.len(), 1);
3317
3318        // Phase 1 invariant: mutating the overlay manager must not bleed
3319        // into primary's schema.
3320        overlay.add_label("Forked")?;
3321        assert!(overlay.schema().labels.contains_key("Forked"));
3322        assert!(!primary.schema().labels.contains_key("Forked"));
3323
3324        Ok(())
3325    }
3326
3327    #[tokio::test]
3328    async fn with_overlay_merges_added_labels_and_edge_types() -> Result<()> {
3329        use crate::core::fork::SchemaDelta;
3330        use chrono::Utc;
3331
3332        let dir = tempdir()?;
3333        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3334        let path = ObjectStorePath::from("schema.json");
3335        let primary = SchemaManager::load_from_store(store, &path).await?;
3336        primary.add_label("Existing")?;
3337
3338        let label_meta = LabelMeta {
3339            id: 99,
3340            created_at: Utc::now(),
3341            state: SchemaElementState::Active,
3342            description: None,
3343        };
3344        let edge_meta = EdgeTypeMeta {
3345            id: 99,
3346            src_labels: vec!["NewLabel".into()],
3347            dst_labels: vec!["NewLabel".into()],
3348            state: SchemaElementState::Active,
3349            description: None,
3350        };
3351        let delta = SchemaDelta {
3352            added_labels: vec![("NewLabel".to_string(), label_meta)],
3353            added_edge_types: vec![("NewEdge".to_string(), edge_meta)],
3354            added_properties: vec![],
3355        };
3356
3357        let overlay = primary.with_overlay(&delta);
3358        let merged = overlay.schema();
3359        assert!(merged.labels.contains_key("Existing"));
3360        assert!(merged.labels.contains_key("NewLabel"));
3361        assert!(merged.edge_types.contains_key("NewEdge"));
3362
3363        // Primary unchanged.
3364        assert!(!primary.schema().labels.contains_key("NewLabel"));
3365        Ok(())
3366    }
3367
3368    /// N threads racing `get_or_assign_edge_type_id` for the same new name
3369    /// must converge on a single id (the read-lock fast path double-checks
3370    /// under the write lock); a schema-defined type must win over the
3371    /// schemaless registry.
3372    #[tokio::test]
3373    async fn test_get_or_assign_edge_type_id_concurrent() -> Result<()> {
3374        let dir = tempdir()?;
3375        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?);
3376        let path = ObjectStorePath::from("schema.json");
3377        let manager = Arc::new(SchemaManager::load_from_store(store, &path).await?);
3378
3379        let mut handles = Vec::new();
3380        for _ in 0..16 {
3381            let m = manager.clone();
3382            handles.push(std::thread::spawn(move || {
3383                m.get_or_assign_edge_type_id("RACED")
3384            }));
3385        }
3386        let ids: Vec<u32> = handles.into_iter().map(|h| h.join().unwrap()).collect();
3387        assert!(
3388            ids.iter().all(|&id| id == ids[0]),
3389            "all racers must observe one id, got {ids:?}"
3390        );
3391        // Fast path returns the same id afterwards.
3392        assert_eq!(manager.get_or_assign_edge_type_id("RACED"), ids[0]);
3393
3394        // Schema-defined type wins over the schemaless registry.
3395        manager.add_label("A")?;
3396        let declared = manager.add_edge_type("DECLARED", vec!["A".into()], vec!["A".into()])?;
3397        assert_eq!(manager.get_or_assign_edge_type_id("DECLARED"), declared);
3398        Ok(())
3399    }
3400
3401    /// Minting a brand-new schemaless edge type must bump `schema_version`
3402    /// (the plan cache keys on it; untyped traversals bake `all_edge_type_ids()`
3403    /// into the plan, so a stale plan would silently drop edges of the new
3404    /// type). Re-resolving an existing type must NOT bump. (review C5)
3405    #[test]
3406    fn test_new_schemaless_edge_type_bumps_schema_version() {
3407        let mut schema = Schema::default();
3408        let v0 = schema.schema_version;
3409
3410        let id1 = schema.get_or_assign_edge_type_id("FRESH");
3411        assert_eq!(
3412            schema.schema_version,
3413            v0.wrapping_add(1),
3414            "minting a new edge type must bump schema_version"
3415        );
3416
3417        // Re-resolving the same type is a no-op — no further bump.
3418        let id1_again = schema.get_or_assign_edge_type_id("FRESH");
3419        assert_eq!(id1, id1_again);
3420        assert_eq!(
3421            schema.schema_version,
3422            v0.wrapping_add(1),
3423            "resolving an existing edge type must not bump schema_version"
3424        );
3425
3426        // A second distinct new type bumps again.
3427        let _id2 = schema.get_or_assign_edge_type_id("OTHER");
3428        assert_eq!(
3429            schema.schema_version,
3430            v0.wrapping_add(2),
3431            "a second new edge type must bump schema_version again"
3432        );
3433    }
3434
3435    /// L6: label/edge-type names with path separators, whitespace, or
3436    /// control chars are rejected at definition; benign names (incl. `.`)
3437    /// are accepted.
3438    #[test]
3439    fn validate_schema_element_name_rejects_unsafe() {
3440        for bad in ["", "   ", "a/b", "a b", "a\nb", "a\\b", "x\0y"] {
3441            assert!(
3442                SchemaManager::validate_schema_element_name("Label", bad).is_err(),
3443                "expected {bad:?} to be rejected"
3444            );
3445        }
3446        for good in ["Person", "My.Label", "edge_2", "KNOWS"] {
3447            assert!(
3448                SchemaManager::validate_schema_element_name("Label", good).is_ok(),
3449                "expected {good:?} to be accepted"
3450            );
3451        }
3452        // Over-length is rejected.
3453        let long = "x".repeat(MAX_SCHEMA_NAME_LEN + 1);
3454        assert!(SchemaManager::validate_schema_element_name("Label", &long).is_err());
3455    }
3456}