diff --git a/crates/omnigraph-compiler/src/schema/parser.rs b/crates/omnigraph-compiler/src/schema/parser.rs index 6e34e53c..78d5c91b 100644 --- a/crates/omnigraph-compiler/src/schema/parser.rs +++ b/crates/omnigraph-compiler/src/schema/parser.rs @@ -947,11 +947,28 @@ fn validate_type_constraints( if is_edge && (col == "src" || col == "dst") { continue; } - if !prop_names.contains_key(col.as_str()) { + let Some(prop) = prop_names.get(col.as_str()) else { return Err(CompilerError::Parse(format!( "@unique on {} references unknown property '{}'", type_name, col ))); + }; + // Uniqueness needs a type that reduces to a scalar key: a + // Blob cannot be compared or scanned by projection and a + // Vector has no meaningful equality — mirror the @key + // guards so the class fails at compile time, not at + // validation/scan time. + if matches!(prop.prop_type.scalar, ScalarType::Blob) { + return Err(CompilerError::Parse(format!( + "@unique is not supported on blob property {}.{}", + type_name, col + ))); + } + if matches!(prop.prop_type.scalar, ScalarType::Vector(_)) { + return Err(CompilerError::Parse(format!( + "@unique is not supported on vector property {}.{}", + type_name, col + ))); } } } diff --git a/crates/omnigraph-compiler/src/schema/parser_tests.rs b/crates/omnigraph-compiler/src/schema/parser_tests.rs index 9a2e1ba1..b18c4025 100644 --- a/crates/omnigraph-compiler/src/schema/parser_tests.rs +++ b/crates/omnigraph-compiler/src/schema/parser_tests.rs @@ -1013,3 +1013,37 @@ fn test_parse_error_diagnostic_has_span() { let err = parse_schema_diagnostic(input).unwrap_err(); assert!(err.span.is_some()); } + +#[test] +fn test_reject_unique_on_blob_property() { + let input = r#" +node Doc { +name: String @key +data: Blob? +@unique(name, data) +} +"#; + let err = parse_schema(input).unwrap_err(); + assert!( + err.to_string() + .contains("@unique is not supported on blob property Doc.data"), + "got: {err}" + ); +} + +#[test] +fn test_reject_unique_on_vector_property() { + let input = r#" +node Doc { +name: String @key +embedding: Vector(4) +@unique(embedding) +} +"#; + let err = parse_schema(input).unwrap_err(); + assert!( + err.to_string() + .contains("@unique is not supported on vector property Doc.embedding"), + "got: {err}" + ); +} diff --git a/crates/omnigraph/src/exec/mutation.rs b/crates/omnigraph/src/exec/mutation.rs index ae820d08..48a9d5ac 100644 --- a/crates/omnigraph/src/exec/mutation.rs +++ b/crates/omnigraph/src/exec/mutation.rs @@ -382,6 +382,124 @@ fn predicate_to_sql( /// Replace specific columns in a RecordBatch with new literal values. /// +/// Field-level update: the column plan for a partial-schema update. +struct PartialUpdatePlan { + /// Accumulated-batch schema: (id + assigned + completion), in catalog + /// order — the validation change-set's view. + output_schema: SchemaRef, + /// Completion columns that ride ONLY the validation change-set and are + /// projected OUT of the staged merge source (staging them would patch + /// unassigned columns and prune their indexes). The staged set is derived + /// at stage time as (batch schema − validation_only), so it stays + /// consistent with `apply_assignments`' own omissions (e.g. a Blob + /// assigned a non-String value is omitted from the batch and therefore + /// never listed for staging). + validation_only: Vec, + /// Scan schema: (id + completion-minus-assigned), in catalog order — + /// the only columns whose OLD values the update needs. + scan_schema: SchemaRef, + /// `scan_schema`'s column names (the scan projection). + scan_cols: Vec, +} + +/// Compute the partial-update column plan: the staged source carries the merge +/// key (`id`) and the assigned columns; the validation change-set additionally +/// carries the constraint-completion columns so the end-of-query evaluator can +/// validate whole tuples (an update assigning only `room` of +/// `@unique(room, hour)` must still detect a (room, hour) collision). +/// +/// Completion is the FIXED-POINT CLOSURE over every `@unique` group AND the +/// `@key` group (the evaluator registers `@key` as a unique constraint too): +/// overlapping groups chain — assigning `room` pulls in `hour` via +/// `(room, hour)`, which pulls in `day` via `(hour, day)` — and the evaluator +/// hard-errors on a partially present group, so any group intersecting the +/// needed set must be completed in full. Blob columns are excluded from +/// completion/scan end-to-end (they cannot be projected through the scanner; +/// the schema layer also rejects `@unique` on Blob/Vector). +/// +/// Columns outside the plan are left physically untouched by the matched-only +/// merge, and every index over them keeps its coverage (Lance prunes only +/// `fields_modified`). +fn partial_update_plan( + node_type: &omnigraph_compiler::catalog::NodeType, + full_schema: &SchemaRef, + resolved: &HashMap, +) -> PartialUpdatePlan { + // A blob assigned a non-String value is a no-op on every path (the + // whole-row path copies the committed value through; historically the + // column was omitted) — exclude it from the plan so the partial batch + // neither requires a materializing scan nor patches the blob. + let assigned: HashSet<&str> = resolved + .iter() + .filter(|(name, value)| { + !(node_type.blob_properties.contains(name.as_str()) + && !matches!(value, Literal::String(_))) + }) + .map(|(name, _)| name.as_str()) + .collect(); + + let mut groups: Vec<&[String]> = node_type + .unique_constraints + .iter() + .map(|g| g.as_slice()) + .collect(); + if let Some(key) = &node_type.key { + groups.push(key.as_slice()); + } + let mut needed: HashSet<&str> = assigned.clone(); + loop { + let mut grew = false; + for group in &groups { + if group.iter().any(|c| needed.contains(c.as_str())) + && !group.iter().all(|c| needed.contains(c.as_str())) + { + for c in group.iter() { + needed.insert(c.as_str()); + } + grew = true; + } + } + if !grew { + break; + } + } + + let mut output_fields = Vec::new(); + let mut scan_fields = Vec::new(); + let mut validation_only = Vec::new(); + for field in full_schema.fields() { + let name = field.name().as_str(); + let is_assigned = assigned.contains(name); + let is_completion = !is_assigned && name != "id" && needed.contains(name); + let is_blob = node_type.blob_properties.contains(name); + if is_completion && is_blob { + // Defense in depth for pre-guard schemas: a Blob can neither be + // scanned by projection nor compared by the evaluator's key + // encoding; the parser now rejects @unique on Blob outright. + continue; + } + if name == "id" || is_assigned || is_completion { + output_fields.push(field.clone()); + } + if is_completion { + validation_only.push(name.to_string()); + scan_fields.push(field.clone()); + } else if name == "id" { + scan_fields.insert(0, field.clone()); + } + } + let scan_cols = scan_fields + .iter() + .map(|f| f.name().clone()) + .collect::>(); + PartialUpdatePlan { + output_schema: Arc::new(Schema::new(output_fields)), + validation_only, + scan_schema: Arc::new(Schema::new(scan_fields)), + scan_cols, + } +} + /// Blob-bearing updates always arrive with the full logical schema. Committed /// blob payloads were materialized by the caller and rebuilt as logical /// `Struct` arrays; pending batches already have that shape. An @@ -902,6 +1020,31 @@ impl Omnigraph { assignments, predicate, } => { + // Field-level-update eligibility: partial staging only when this + // update is the table's ONLY op in the query (the count + // includes this op itself, so == 1 means "no other op + // touches the type"). Every other combination falls back + // to whole-row staging: partial and full batches cannot + // share one uniform-schema merge source (a present + // column's null cell means "set NULL", so widening a + // partial batch would null-overwrite), a second same-table + // op's read-your-writes scan must see full rows, and one + // table commits at most one version per query + // (invariant 4). Op lists are tiny; the inline count + // replaces a per-query census map. + let partial_ok = ir + .ops + .iter() + .filter(|op| { + let t = match op { + MutationOpIR::Insert { type_name, .. } + | MutationOpIR::Update { type_name, .. } + | MutationOpIR::Delete { type_name, .. } => type_name, + }; + t == type_name + }) + .count() + == 1; self.execute_update( type_name, assignments, @@ -910,6 +1053,7 @@ impl Omnigraph { branch, staging, txn, + partial_ok, ) .await? } @@ -1044,6 +1188,7 @@ impl Omnigraph { branch: Option<&str>, staging: &mut MutationStaging, txn: &crate::db::WriteTxn, + partial_ok: bool, ) -> Result { let catalog = &txn.catalog; // Defense in depth: ensure this is a node type @@ -1054,12 +1199,13 @@ impl Omnigraph { ))); } - // Reject updates to @key properties — identity is immutable - if let Some(key_prop) = catalog.node_types[type_name].key_property() { - if assignments.iter().any(|a| a.property == key_prop) { + // Reject updates to @key properties — identity is immutable. Checks + // EVERY key column (composite keys included), not just the first. + if let Some(key_cols) = &catalog.node_types[type_name].key { + if let Some(a) = assignments.iter().find(|a| key_cols.contains(&a.property)) { return Err(OmniError::manifest(format!( "cannot update @key property '{}' — delete and re-insert instead", - key_prop + a.property ))); } } @@ -1068,6 +1214,34 @@ impl Omnigraph { let schema = catalog.node_types[type_name].arrow_schema.clone(); let blob_props = catalog.node_types[type_name].blob_properties.clone(); + let mut resolved: HashMap = HashMap::new(); + for a in assignments { + resolved.insert(a.property.clone(), resolve_expr_value(&a.value, params)?); + } + + // Field-level update: when this update is the table's only op in the query, + // stage a PARTIAL source — (id + assigned + constraint-completion) + // columns — instead of whole rows. `partial_plan` is `None` on the + // whole-row fallback path. Takes RESOLVED assignments so no-op blob + // assignments (non-String values) are excluded from the plan. + // + // Blob-bearing tables always take the whole-row path: Lance's + // partial-source column patch trips the blob writer on a blob-bearing + // fragment even when no blob column is in the source ("Blob struct + // missing `data` field", lance blob.rs) — pinned by + // `end_to_end::update_with_blob_mid_schema_does_not_panic`. Revisit + // when Lance's blob merge accepts partial sources. + let partial_ok = partial_ok && blob_props.is_empty(); + let partial_plan = if partial_ok { + Some(partial_update_plan( + &catalog.node_types[type_name], + &schema, + &resolved, + )) + } else { + None + }; + let table_key = format!("node:{}", type_name); let (handle, _full_path, _table_branch) = open_table_for_mutation( self, @@ -1100,7 +1274,25 @@ impl Omnigraph { // only those payloads into the logical blob schema before unioning // pending rows. This keeps correctness independent of whether an id // index happens to steer Lance onto its legacy partial-column plan. - let batches = if blob_props.is_empty() { + let batches = if let Some(plan) = partial_plan.as_ref() { + // Field-level update: scan only (id + completion) — assigned + // columns' old values are never needed (assignments are literal + // values) and the WHERE predicate evaluates by pushdown without + // being projected. Blobs are never in this projection (the plan + // excludes them), so the blob-materialization arm below is + // unnecessary here; pending is empty by the eligibility rule. + let scan_cols: Vec<&str> = plan.scan_cols.iter().map(|c| c.as_str()).collect(); + self.storage() + .scan_with_pending( + &ds, + pending_batches, + pending_schema, + Some(scan_cols.as_slice()), + Some(&pred_sql), + Some("id"), + ) + .await? + } else if blob_props.is_empty() { self.storage() .scan_with_pending( &ds, @@ -1131,26 +1323,35 @@ impl Omnigraph { } // Concat the matched batches (committed + pending) into one. The - // helper binds both sides to the catalog's full logical schema. Any - // divergence here is an internal scan/staging contract violation. - let matched = concat_match_batches_to_schema(&schema, batches)?; + // helper binds both sides to the catalog's full logical schema (partial + // path: to the plan's SCAN schema — id + completion). Any divergence + // here is an internal scan/staging contract violation. + let (concat_schema, output_schema) = match partial_plan.as_ref() { + Some(plan) => (plan.scan_schema.clone(), plan.output_schema.clone()), + None => (schema.clone(), schema.clone()), + }; + let matched = concat_match_batches_to_schema(&concat_schema, batches)?; let affected_count = matched.num_rows(); - let mut resolved: HashMap = HashMap::new(); - for a in assignments { - resolved.insert(a.property.clone(), resolve_expr_value(&a.value, params)?); - } - let updated = apply_assignments(&schema, &matched, &resolved, &blob_props)?; + let updated = apply_assignments(&output_schema, &matched, &resolved, &blob_props)?; // Validation (value/enum/unique) runs end-of-query via the evaluator. - // Accumulate the updated batch into the Merge-mode pending stream. - // The accumulator may now contain entries with the same id as a - // prior insert or update on this table; `MutationStaging::finalize` - // dedupes by id (last-occurrence wins) before issuing the single - // `stage_merge_insert` call at end-of-query. + // Accumulate the updated batch. Whole-row: the Merge-mode pending + // stream (may coalesce with prior same-table ops; finalize dedupes by + // id, last wins). Partial: a dedicated partial-update entry + // that stages as a matched-only merge. let updated_schema = updated.schema(); - staging.append_batch(&table_key, updated_schema, PendingMode::Merge, updated)?; + if let Some(plan) = partial_plan.as_ref() { + staging.append_partial_update_batch( + &table_key, + updated_schema, + updated, + plan.validation_only.clone(), + )?; + } else { + staging.append_batch(&table_key, updated_schema, PendingMode::Merge, updated)?; + } Ok(MutationResult { affected_nodes: affected_count, diff --git a/crates/omnigraph/src/exec/staging.rs b/crates/omnigraph/src/exec/staging.rs index 5e7fe811..7d030ba9 100644 --- a/crates/omnigraph/src/exec/staging.rs +++ b/crates/omnigraph/src/exec/staging.rs @@ -44,6 +44,14 @@ use crate::error::{OmniError, Result}; pub(crate) enum PendingMode { Append, Merge, + /// Field-level update: a Merge whose staged source is the batch + /// schema MINUS the table's `partial_validation_only` columns, staged + /// matched-only (`WhenNotMatched::DoNothing`). A distinct variant — not a + /// side-flag on `Merge` — so every `match` on the mode is forced to + /// disposition partial batches explicitly: treating one as a full-row + /// Merge (a future coalescing change, a retry path) would null-overwrite + /// unassigned columns, and the compiler now surfaces that decision. + PartialUpdate, Overwrite, } @@ -69,6 +77,22 @@ pub(crate) struct PendingTable { pub(crate) schema: SchemaRef, pub(crate) mode: PendingMode, pub(crate) batches: Vec, + /// Field-level update: `Some(validation_only)` marks the + /// accumulated batches as PARTIAL — they carry (merge key + assigned + + /// constraint-completion) columns for the validation change-set, and the + /// STAGED merge source is the batch schema MINUS these validation-only + /// completion columns, staged matched-only (`WhenNotMatched::DoNothing`). + /// Deriving the staged set by subtraction (rather than listing it) keeps + /// it consistent with `apply_assignments`' own omissions — e.g. a Blob + /// assigned a non-String value is omitted from the batch entirely. + /// Completion columns are validation inputs, never merge inputs: Lance + /// counts every source column as modified, so staging an unassigned + /// `@unique`-group member would patch it and prune its index for no + /// semantic reason. Set only by + /// [`MutationStaging::append_partial_update_batch`], whose eligibility + /// rule (the table's ONLY op in this query is a single `update`) + /// guarantees no full-schema batch ever lands on the same table. + pub(crate) partial_validation_only: Option>, } impl PendingTable { @@ -77,6 +101,7 @@ impl PendingTable { schema, mode, batches: Vec::new(), + partial_validation_only: None, } } @@ -208,6 +233,13 @@ impl MutationStaging { // caller a clearer point of failure attached to the specific // op that introduced the drift. if let Some(existing) = self.pending.get(table_key) { + if existing.mode == PendingMode::PartialUpdate { + return Err(OmniError::manifest_internal(format!( + "append_batch: table '{}' holds a partial-update batch — the \ + eligibility rule should have prevented a second op on this table", + table_key + ))); + } if existing.mode == PendingMode::Overwrite || mode == PendingMode::Overwrite { if existing.mode != mode { return Err(OmniError::manifest_internal(format!( @@ -245,6 +277,38 @@ impl MutationStaging { Ok(()) } + /// Field-level update: stage a PARTIAL-schema update batch — (key + assigned + + /// completion) columns — for a table whose only op in this query is one + /// `update`. The caller (`execute_update`) enforces that eligibility rule + /// from the lowered IR, so an existing accumulator entry here is an + /// internal invariant breach, not a user error: partial and full batches + /// cannot share one uniform-schema merge source, and silently widening + /// (or narrowing) would corrupt the staged shape. + pub(crate) fn append_partial_update_batch( + &mut self, + table_key: &str, + schema: SchemaRef, + batch: RecordBatch, + validation_only: Vec, + ) -> Result<()> { + if batch.num_rows() == 0 { + return Ok(()); + } + if self.pending.contains_key(table_key) { + return Err(OmniError::manifest_internal(format!( + "append_partial_update_batch: table '{}' already has pending batches — \ + partial-update staging requires the update to be the table's only op \ + (the eligibility rule in execute_named_mutation)", + table_key + ))); + } + let mut entry = PendingTable::new(schema, PendingMode::PartialUpdate); + entry.partial_validation_only = Some(validation_only); + entry.batches.push(batch); + self.pending.insert(table_key.to_string(), entry); + Ok(()) + } + /// Record a delete predicate for `table_key`. The caller must have already /// called `ensure_path` (via `open_table_for_mutation`) so the table's /// path/version/op-kind are captured. D₂ guarantees a delete-touched table @@ -468,7 +532,10 @@ async fn stage_pending_table( // pin; its files stage on the target handle after sidecar + fork. let stage_kind = match table.mode { PendingMode::Append => crate::db::MutationOpKind::Insert, - PendingMode::Merge => crate::db::MutationOpKind::Merge, + // PartialUpdate reopens with the same non-strict policy as Merge; + // strictness for the commit-time drift check comes from `op_kinds` + // (the update op registered MutationOpKind::Update there). + PendingMode::Merge | PendingMode::PartialUpdate => crate::db::MutationOpKind::Merge, PendingMode::Overwrite => crate::db::MutationOpKind::SchemaRewrite, }; let ds = match path.deferred_fork.as_ref() { @@ -494,7 +561,9 @@ async fn stage_pending_table( } let combined = match table.mode { - PendingMode::Merge => dedupe_merge_batches_by_id(&table.schema, table.batches)?, + PendingMode::Merge | PendingMode::PartialUpdate => { + dedupe_merge_batches_by_id(&table.schema, table.batches)? + } PendingMode::Append | PendingMode::Overwrite => { if table.batches.len() == 1 { table.batches.into_iter().next().unwrap() @@ -509,6 +578,19 @@ async fn stage_pending_table( // transaction is staged on the new target ref after the sidecar is // durable, then bound to this UUID. Its read version remains Lance's // independently-derived value and is checked by the binder. + // Field-level update: project the staged set here so the deferred + // stage receives exactly what a non-deferred stage would. + let combined = match (&table.mode, &table.partial_validation_only) { + (PendingMode::PartialUpdate, Some(validation_only)) => { + project_out_validation_only(combined, validation_only)? + } + (PendingMode::PartialUpdate, None) => { + return Err(OmniError::manifest_internal( + "PartialUpdate entry without validation-only columns".to_string(), + )); + } + _ => combined, + }; let planned_transaction = pre_minted_transaction_identity(expected); return Ok(Some(StagedTableEntry { table_key, @@ -539,6 +621,32 @@ async fn stage_pending_table( ) .await? } + PendingMode::PartialUpdate => { + // Field-level update: staged source = batch schema − validation-only columns + // (completion columns ride the validation change-set only; staging + // them would patch unassigned columns and prune their indexes). A + // subtraction cannot reference a column the batch lacks, so + // apply-time omissions (unassigned blobs) are immune. Matched-only: + // Lance patches the provided columns in place and never inserts — + // a subset schema could not satisfy non-null target columns anyway, + // and the update executor only stages rows it matched. + let validation_only = + table.partial_validation_only.as_deref().ok_or_else(|| { + OmniError::manifest_internal( + "PartialUpdate entry without validation-only columns".to_string(), + ) + })?; + let projected = project_out_validation_only(combined, validation_only)?; + db.storage() + .stage_merge_insert( + ds.clone(), + projected, + vec!["id".to_string()], + lance::dataset::WhenMatched::UpdateAll, + lance::dataset::WhenNotMatched::DoNothing, + ) + .await? + } PendingMode::Overwrite => db.storage().stage_overwrite(&ds, combined).await?, }; let planned_transaction = staged.transaction_identity(); @@ -665,6 +773,28 @@ fn pre_minted_transaction_identity( } } +/// Field-level update: staged source = batch schema − validation-only +/// completion columns (they ride the validation change-set only; staging them +/// would patch unassigned columns and prune their indexes). Subtraction from +/// the batch's actual schema keeps the projection immune to apply-time +/// omissions (e.g. a Blob assigned a non-String value is absent entirely). +fn project_out_validation_only( + batch: RecordBatch, + validation_only: &[String], +) -> Result { + let schema = batch.schema(); + let indices: Vec = schema + .fields() + .iter() + .enumerate() + .filter(|(_, f)| !validation_only.contains(f.name())) + .map(|(i, _)| i) + .collect(); + batch + .project(&indices) + .map_err(|e| OmniError::Lance(e.to_string())) +} + async fn stage_deferred_plan( db: &crate::db::Omnigraph, target: SnapshotHandle, @@ -685,6 +815,20 @@ async fn stage_deferred_plan( ) .await? } + // Field-level update: the batch was projected down to the staged + // set (key + assigned) at pack time; matched-only so unmatched + // partial rows can never insert. + PendingMode::PartialUpdate => { + db.storage() + .stage_merge_insert( + target, + batch, + vec!["id".to_string()], + lance::dataset::WhenMatched::UpdateAll, + lance::dataset::WhenNotMatched::DoNothing, + ) + .await? + } PendingMode::Overwrite => db.storage().stage_overwrite(&target, batch).await?, }, DeferredStagePlan::Delete { predicate } => db diff --git a/crates/omnigraph/src/instrumentation.rs b/crates/omnigraph/src/instrumentation.rs index d235038e..fafb38f6 100644 --- a/crates/omnigraph/src/instrumentation.rs +++ b/crates/omnigraph/src/instrumentation.rs @@ -175,6 +175,14 @@ pub struct MergeWriteProbes { pub stage_append_rows: Arc, pub stage_merge_insert_calls: Arc, pub stage_merge_insert_rows: Arc, + /// Per-call shape of each `stage_merge_insert`: the source batch's column + /// names (schema order) and whether unmatched source rows insert + /// (`WhenNotMatched::InsertAll`) or drop (`DoNothing`). Lets a fitness test + /// assert a partial-schema matched-only update stages exactly + /// (key + assigned) columns — the field-level-update staging shape. + /// (`@unique`-completion columns ride only the validation change-set and + /// are projected out before staging, so they never appear here.) + pub merge_shapes: Arc>>, /// Inline vector-index (IVF) builds. The fast-forward adopt path defers /// index coverage to the reconciler, so an adopt merge must do 0 of these. pub create_vector_index_calls: Arc, @@ -184,10 +192,22 @@ pub struct MergeWriteProbes { pub scan_staged_combined_calls: Arc, } +/// One `stage_merge_insert` call's observable shape (see +/// [`MergeWriteProbes::merge_shapes`]). +#[derive(Clone, Debug)] +pub struct MergeShape { + pub source_columns: Vec, + pub inserts_unmatched: bool, +} + impl MergeWriteProbes { pub fn stage_append_calls(&self) -> u64 { self.stage_append_calls.load(Ordering::Relaxed) } + /// Snapshot of the recorded per-call merge shapes, in call order. + pub fn merge_shapes(&self) -> Vec { + self.merge_shapes.lock().unwrap().clone() + } pub fn stage_append_rows(&self) -> u64 { self.stage_append_rows.load(Ordering::Relaxed) } @@ -227,12 +247,28 @@ pub(crate) fn record_stage_append(rows: u64) { }); } -/// Record one `stage_merge_insert` of `rows` rows against the active probes. -/// No-op in production (no probes installed). -pub(crate) fn record_stage_merge_insert(rows: u64) { +/// Record one `stage_merge_insert` of `rows` rows against the active probes, +/// with its observable shape (source columns + unmatched-row disposition). +/// No-op in production (no probes installed): the schema arrives as a cheap +/// `&SchemaRef` borrow and the column-name strings materialize only inside +/// `try_with` — the same classify-inside-the-closure pattern as +/// `record_open`, so production pays zero allocations. +pub(crate) fn record_stage_merge_insert( + rows: u64, + source_schema: &arrow_schema::SchemaRef, + inserts_unmatched: bool, +) { let _ = MERGE_WRITE_PROBES.try_with(|p| { p.stage_merge_insert_calls.fetch_add(1, Ordering::Relaxed); p.stage_merge_insert_rows.fetch_add(rows, Ordering::Relaxed); + p.merge_shapes.lock().unwrap().push(MergeShape { + source_columns: source_schema + .fields() + .iter() + .map(|f| f.name().clone()) + .collect(), + inserts_unmatched, + }); }); } diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs index 528de07b..e8a1960b 100644 --- a/crates/omnigraph/src/table_store.rs +++ b/crates/omnigraph/src/table_store.rs @@ -897,7 +897,7 @@ impl TableStore { return self.table_state(dataset_uri, ds).await; } let schema = batch.schema(); - let reader = arrow_array::RecordBatchIterator::new(vec![Ok(batch)], schema); + let reader = arrow_array::RecordBatchIterator::new(vec![Ok(batch)], schema.clone()); let params = WriteParams { mode: WriteMode::Append, allow_external_blob_outside_bases: true, @@ -1218,6 +1218,7 @@ impl TableStore { )); } let merged_rows = batch.num_rows() as u64; + let inserts_unmatched = matches!(when_not_matched, WhenNotMatched::InsertAll); // Precondition for the FirstSeen workaround below: every call path that // reaches stage_merge_insert (load, MutationStaging::finalize, @@ -1252,7 +1253,7 @@ impl TableStore { .try_build() .map_err(|e| OmniError::Lance(e.to_string()))?; let schema = batch.schema(); - let reader = arrow_array::RecordBatchIterator::new(vec![Ok(batch)], schema); + let reader = arrow_array::RecordBatchIterator::new(vec![Ok(batch)], schema.clone()); let stream = lance_datafusion::utils::reader_to_stream(Box::new(reader)); let uncommitted = job .execute_uncommitted(stream) @@ -1260,7 +1261,11 @@ impl TableStore { .map_err(|e| OmniError::Lance(e.to_string()))?; // Record only after the staging write succeeds, so a failed write does // not inflate the probe (matches `stage_append`/`stage_append_stream`). - crate::instrumentation::record_stage_merge_insert(merged_rows); + crate::instrumentation::record_stage_merge_insert( + merged_rows, + &schema, + inserts_unmatched, + ); // Operation::Update { removed_fragment_ids, updated_fragments, new_fragments, .. } — // `new_fragments` are the freshly inserted rows; `updated_fragments` // are rewrites of existing fragments that include both retained and diff --git a/crates/omnigraph/src/validate.rs b/crates/omnigraph/src/validate.rs index 1d0dc0d7..4527822a 100644 --- a/crates/omnigraph/src/validate.rs +++ b/crates/omnigraph/src/validate.rs @@ -682,6 +682,18 @@ async fn evaluate_unique( // became null removes the id (it no longer holds a unique key). let mut final_by_id: HashMap, Vec)> = HashMap::new(); for batch in change.value_batches() { + // Field-level-update partial batches carry only (id + assigned + + // completion) columns. A group with NO column in the batch is + // untouched by the write — its committed values still satisfy the + // constraint — so skip it. A PARTIALLY present group would mean the + // completion-column projection failed; keep that loud (the error + // below fires on the first missing member). + if columns + .iter() + .all(|name| batch.column_by_name(name).is_none()) + { + continue; + } let group_columns = columns .iter() .map(|name| { diff --git a/crates/omnigraph/tests/lance_surface_guards.rs b/crates/omnigraph/tests/lance_surface_guards.rs index a969ffb5..4472afe8 100644 --- a/crates/omnigraph/tests/lance_surface_guards.rs +++ b/crates/omnigraph/tests/lance_surface_guards.rs @@ -1314,3 +1314,149 @@ async fn filtered_scan_tolerates_merge_update_row_id_overlap() { assert_eq!(rows, expected, "filtered read for {slug}"); } } + +// --- Guard: partial-schema merge patches columns in place, pruning only +// --- fields_modified (the RFC-022 substrate contract) ----------------------- +// +// A merge-insert whose source carries a SUBSET of the target schema must: +// (a) patch the provided columns for matched rows IN PLACE — same fragment +// id, no new fragments (`update_mode: RewriteColumns`); +// (b) leave missing columns' values untouched; +// (c) prune ONLY indexes covering a source column from the touched +// fragment's bitmap (`prune_updated_fields_from_indices` keyed on +// `fields_modified`) — an index on an untouched column keeps coverage. +// Current residual pinned here: the join key rides the source, so the KEY +// column's index is pruned too even though matched keys are equal by +// definition. If (d) below goes red, upstream started excluding ON columns +// from column patches — tighten omnigraph's coverage test in +// scalar_indexes.rs and drop this arm. +#[tokio::test] +async fn partial_schema_merge_patches_in_place_and_prunes_only_modified_fields() { + use arrow_array::Int64Array; + use futures::TryStreamExt; + + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().join("guard_partial_merge.lance"); + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, false), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["x", "y", "z"])), + Arc::new(Int64Array::from(vec![1, 2, 3])), + Arc::new(Int64Array::from(vec![10, 20, 30])), + ], + ) + .unwrap(); + let reader = RecordBatchIterator::new(vec![Ok(batch)], schema.clone()); + let params = WriteParams { + mode: WriteMode::Create, + enable_stable_row_ids: true, + data_storage_version: Some(LanceFileVersion::V2_2), + ..Default::default() + }; + let mut ds = Dataset::write(reader, uri.to_str().unwrap(), Some(params)) + .await + .unwrap(); + ds.create_index_builder(&["id"], IndexType::BTree, &ScalarIndexParams::default()) + .await + .unwrap(); + ds.create_index_builder(&["b"], IndexType::BTree, &ScalarIndexParams::default()) + .await + .unwrap(); + ds.checkout_latest().await.unwrap(); + let frags_before: Vec = ds.fragments().iter().map(|f| f.id).collect(); + + // Partial source: (id, a) — patch a=200 for id="y". No `b` column. + let partial_schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("a", DataType::Int64, false), + ])); + let patch = RecordBatch::try_new( + partial_schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["y"])), + Arc::new(Int64Array::from(vec![200i64])), + ], + ) + .unwrap(); + let job = MergeInsertBuilder::try_new(Arc::new(ds), vec!["id".to_string()]) + .unwrap() + .when_matched(WhenMatched::UpdateAll) + .when_not_matched(WhenNotMatched::DoNothing) + .try_build() + .unwrap(); + let source = RecordBatchIterator::new(vec![Ok(patch)], partial_schema); + let (ds, _stats) = job.execute_reader(source).await.unwrap(); + + // (a) in place: same fragment set. + let frags_after: Vec = ds.fragments().iter().map(|f| f.id).collect(); + assert_eq!(frags_before, frags_after, "partial merge must patch in place"); + + // (b) patched + retained values. + let rows = ds + .scan() + .try_into_stream() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let all = arrow_select::concat::concat_batches(&rows[0].schema(), &rows).unwrap(); + let ids = all + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let a = all + .column_by_name("a") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let b = all + .column_by_name("b") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + for i in 0..all.num_rows() { + match ids.value(i) { + "y" => { + assert_eq!(a.value(i), 200, "assigned column patched"); + assert_eq!(b.value(i), 20, "missing column retained"); + } + "x" => assert_eq!((a.value(i), b.value(i)), (1, 10)), + "z" => assert_eq!((a.value(i), b.value(i)), (3, 30)), + other => panic!("unexpected id {other}"), + } + } + + // (c) index on untouched column `b` keeps its fragment coverage. + let indices = ds.load_indices().await.unwrap(); + let bitmap_of = |col_field: &str| { + let field_id = ds.schema().field(col_field).unwrap().id; + indices + .iter() + .find(|i| i.fields == vec![field_id]) + .and_then(|i| i.fragment_bitmap.as_ref()) + .map(|bm| bm.iter().collect::>()) + }; + assert_eq!( + bitmap_of("b"), + Some(frags_before.iter().map(|f| *f as u32).collect::>()), + "index on an untouched column must keep the patched fragment" + ); + // (d) residual: the join key's index IS pruned (source carries `id`). + assert_eq!( + bitmap_of("id"), + Some(vec![]), + "join-key index currently loses the patched fragment; if this went red \ + with a non-empty bitmap, upstream now excludes ON columns from patches \ + — tighten scalar_indexes.rs and update RFC-022's residual note" + ); +} diff --git a/crates/omnigraph/tests/scalar_indexes.rs b/crates/omnigraph/tests/scalar_indexes.rs index bfa31054..d7683357 100644 --- a/crates/omnigraph/tests/scalar_indexes.rs +++ b/crates/omnigraph/tests/scalar_indexes.rs @@ -73,3 +73,113 @@ async fn node_scalar_and_enum_index_columns_get_btree() { "un-annotated column should have no scalar index, got {note_cov:?}" ); } + +const TOUCH_NOTE: &str = r#" +query touch_note($slug: String, $note: String) { + update Item set { note: $note } where slug = $slug +} +"#; + +// Field-level-update acceptance cell (red → green): updating an UN-indexed property must +// not degrade index coverage on any OTHER column. The whole-row update merge +// marks every column modified, so Lance prunes the touched fragments from every +// index (`prune_updated_fields_from_indices`) and the rewritten rows land +// outside coverage; the partial-schema update (fields_modified = assigned +// columns only, in-place column patch on the same fragment) leaves the id/enum/ +// scalar BTREEs intact. This is the erosion class measured in the field-level- +// updates investigation: keyed operations degrade ~+1 read/fragment after +// whole-row updates until an optimize. +#[tokio::test] +async fn update_of_unindexed_property_preserves_other_index_coverage() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let mut db = Omnigraph::init(uri, SCHEMA).await.unwrap(); + load_jsonl(&mut db, DATA, LoadMode::Overwrite).await.unwrap(); + db.ensure_indices().await.unwrap(); + + db.mutate( + "main", + TOUCH_NOTE, + "touch_note", + ¶ms(&[("$slug", "a"), ("$note", "touched")]), + ) + .await + .unwrap(); + + let snap = snapshot_main(&db).await.unwrap(); + let ds = snap.open("node:Item").await.unwrap(); + for col in ["status", "published", "rank"] { + let cov = TableStore::key_column_index_coverage(&ds, col).await.unwrap(); + assert_eq!( + cov, + IndexCoverage::Indexed, + "updating un-indexed 'note' must not degrade the index on '{col}', got {cov:?}" + ); + } + + // Known residual (tripwire, not the goal state): the partial merge source + // must carry the join key (`id`) for matching, and Lance's column patcher + // counts every source column as modified — including the ON column whose + // values are equal by definition of the match — so the id BTREE alone + // loses the patched fragment until the reconciler folds it back. Parity + // with the whole-row path for `id`, strictly better for every other index. + // When upstream excludes ON columns from column patches this assertion + // goes red — flip it to `Indexed` and delete this comment. + let id_cov = TableStore::key_column_index_coverage(&ds, "id").await.unwrap(); + assert!( + matches!(id_cov, IndexCoverage::Degraded { .. }), + "id-BTREE currently loses patched fragments (join key rides the source); \ + if this is now Indexed, upstream fixed ON-column patching — tighten this test" + ); +} + +const COMPOSITE_SCHEMA: &str = r#" +node Slot { + name: String @key + room: String? + hour: I32 @index + @unique(room, hour) +} +"#; + +const COMPOSITE_DATA: &str = r#"{"type":"Slot","data":{"name":"s1","room":"roomA","hour":9}} +{"type":"Slot","data":{"name":"s2","room":"roomB","hour":9}}"#; + +const MOVE_SLOT: &str = r#" +query move_slot($name: String, $room: String) { + update Slot set { room: $room } where name = $name +} +"#; + +// A `@unique`-group completion column is a VALIDATION input, not a merge +// input: an update assigning only `room` must not patch `hour` (its value is +// unchanged), so `hour`'s index keeps the fragment. The staged merge source is +// (id + assigned); the completion column rides only the validation change-set. +#[tokio::test] +async fn completion_column_index_survives_partial_update() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let mut db = Omnigraph::init(uri, COMPOSITE_SCHEMA).await.unwrap(); + load_jsonl(&mut db, COMPOSITE_DATA, LoadMode::Overwrite) + .await + .unwrap(); + db.ensure_indices().await.unwrap(); + + db.mutate( + "main", + MOVE_SLOT, + "move_slot", + ¶ms(&[("$name", "s2"), ("$room", "roomC")]), + ) + .await + .unwrap(); + + let snap = snapshot_main(&db).await.unwrap(); + let ds = snap.open("node:Slot").await.unwrap(); + let cov = TableStore::key_column_index_coverage(&ds, "hour").await.unwrap(); + assert_eq!( + cov, + IndexCoverage::Indexed, + "completion column 'hour' was not assigned — its index must keep coverage, got {cov:?}" + ); +} diff --git a/crates/omnigraph/tests/validators.rs b/crates/omnigraph/tests/validators.rs index 226968f5..7a331318 100644 --- a/crates/omnigraph/tests/validators.rs +++ b/crates/omnigraph/tests/validators.rs @@ -8,7 +8,7 @@ mod helpers; use omnigraph::db::Omnigraph; use omnigraph::loader::{LoadMode, load_jsonl}; -use helpers::{count_rows, mutate_main, params}; +use helpers::{count_rows, mixed_params, mutate_main, params}; const ENUM_SCHEMA: &str = r#" node Person { @@ -708,3 +708,142 @@ async fn cardinality_rejected_on_jsonl_load() { err ); } + +// ─── RFC-022: composite-unique completion under partial updates ───────────── + +const COMPOSITE_UNIQUE_SCHEMA: &str = r#" +node Slot { + name: String @key + room: String? + hour: I32? + @unique(room, hour) +} +"#; + +const COMPOSITE_MUTATIONS: &str = r#" +query insert_slot($name: String, $room: String, $hour: I32) { + insert Slot { name: $name, room: $room, hour: $hour } +} +query move_slot($name: String, $room: String) { + update Slot set { room: $room } where name = $name +} +"#; + +/// An update assigning only PART of a composite `@unique` group must still +/// validate the whole group — the unassigned member's committed value completes +/// the tuple. Guards RFC-022's completion-column projection rule: a partial +/// update that assigns `room` must also carry `hour` (old value) into the +/// change-set, or the (room, hour) collision below would go undetected. +#[tokio::test] +async fn partial_update_completes_composite_unique_group() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let db = omnigraph::db::Omnigraph::init(uri, COMPOSITE_UNIQUE_SCHEMA) + .await + .unwrap(); + + for (name, room, hour) in [("s1", "roomA", 9i64), ("s2", "roomB", 9)] { + db.mutate( + "main", + COMPOSITE_MUTATIONS, + "insert_slot", + &mixed_params(&[("$name", name), ("$room", room)], &[("$hour", hour)]), + ) + .await + .unwrap(); + } + + // Moving s2 into roomA collides with s1's (roomA, 9). + let err = db + .mutate( + "main", + COMPOSITE_MUTATIONS, + "move_slot", + ¶ms(&[("$name", "s2"), ("$room", "roomA")]), + ) + .await + .unwrap_err(); + assert!( + err.to_string().contains("@unique"), + "moving s2 to (roomA, 9) must be a composite-unique violation, got: {err}" + ); + + // A non-colliding move succeeds. + db.mutate( + "main", + COMPOSITE_MUTATIONS, + "move_slot", + ¶ms(&[("$name", "s2"), ("$room", "roomC")]), + ) + .await + .unwrap(); +} + +const OVERLAP_UNIQUE_SCHEMA: &str = r#" +node Booking { + name: String @key + room: String? + hour: I32? + day: String? + @unique(room, hour) + @unique(hour, day) +} +"#; + +const OVERLAP_MUTATIONS: &str = r#" +query insert_booking($name: String, $room: String, $hour: I32, $day: String) { + insert Booking { name: $name, room: $room, hour: $hour, day: $day } +} +query move_room($name: String, $room: String) { + update Booking set { room: $room } where name = $name +} +"#; + +/// Overlapping `@unique` groups: a sole update assigning `room` pulls in +/// `hour` (its group partner), and `hour`'s OTHER group `(hour, day)` must +/// then be completed transitively — the evaluator hard-errors on a partially +/// present group, so without the transitive closure a perfectly legal update +/// fails with "missing unique column 'day'". Both directions pinned: the +/// legal move succeeds, and a genuine (room, hour) collision is still caught. +#[tokio::test] +async fn overlapping_unique_groups_allow_sole_partial_update() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let db = Omnigraph::init(uri, OVERLAP_UNIQUE_SCHEMA).await.unwrap(); + + for (name, room, hour, day) in [("b1", "roomA", 9i64, "mon"), ("b2", "roomB", 9, "tue")] { + db.mutate( + "main", + OVERLAP_MUTATIONS, + "insert_booking", + &mixed_params(&[("$name", name), ("$room", room), ("$day", day)], &[("$hour", hour)]), + ) + .await + .unwrap(); + } + + // Legal sole update — touches neither `day` nor collides. + db.mutate( + "main", + OVERLAP_MUTATIONS, + "move_room", + ¶ms(&[("$name", "b2"), ("$room", "roomC")]), + ) + .await + .unwrap(); + + // Genuine collision on (room, hour) still rejected. + let err = db + .mutate( + "main", + OVERLAP_MUTATIONS, + "move_room", + ¶ms(&[("$name", "b2"), ("$room", "roomA")]), + ) + .await + .unwrap_err(); + assert!( + err.to_string().contains("@unique"), + "moving b2 to (roomA, 9) must collide with b1, got: {err}" + ); +} diff --git a/crates/omnigraph/tests/writes.rs b/crates/omnigraph/tests/writes.rs index 2181fd9d..fae8d2bb 100644 --- a/crates/omnigraph/tests/writes.rs +++ b/crates/omnigraph/tests/writes.rs @@ -693,8 +693,10 @@ async fn mixed_insert_and_update_on_same_person_coalesces_to_one_merge() { let pre_version = version_main(&db).await.unwrap(); - let result = db - .mutate( + let probes = omnigraph::instrumentation::MergeWriteProbes::default(); + let result = omnigraph::instrumentation::with_merge_write_probes( + probes.clone(), + db.mutate( "main", STAGED_QUERIES, "insert_then_update_same_person", @@ -702,11 +704,28 @@ async fn mixed_insert_and_update_on_same_person_coalesces_to_one_merge() { &[("$name", "Yves")], &[("$insert_age", 10), ("$update_age", 99)], ), - ) - .await - .unwrap(); + ), + ) + .await + .unwrap(); assert_eq!(result.affected_nodes, 2, "1 insert + 1 update reported"); + // Field-level-update fallback pin: mixing insert + update on one table stages FULL + // rows as an upsert — partial and full batches cannot share one + // uniform-schema merge source, and one table commits at most one version + // per query (invariant 4). + let shapes = probes.merge_shapes(); + assert_eq!(shapes.len(), 1, "insert+update coalesce into one staged merge"); + assert!( + shapes[0].source_columns.contains(&"name".to_string()), + "mixed insert+update stages full rows, got {:?}", + shapes[0].source_columns + ); + assert!( + shapes[0].inserts_unmatched, + "mixed insert+update keeps upsert semantics (InsertAll)" + ); + // The end-state row carries the update value (last-write-wins via // end-of-query dedupe), proving the staged merge_insert ran with the // correct source dedupe. Read the underlying Person table directly @@ -1099,22 +1118,38 @@ async fn chained_updates_with_overlapping_predicate_respects_intermediate_value( let pre_version = version_main(&db).await.unwrap(); - db.mutate( - "main", - STAGED_QUERIES, - "update_then_filter_by_old_value", - &mixed_params( - &[("$first_name", "Alice")], - &[ - ("$first_new_age", 99), - ("$second_threshold", 50), - ("$second_new_age", 10), - ], + let probes = omnigraph::instrumentation::MergeWriteProbes::default(); + omnigraph::instrumentation::with_merge_write_probes( + probes.clone(), + db.mutate( + "main", + STAGED_QUERIES, + "update_then_filter_by_old_value", + &mixed_params( + &[("$first_name", "Alice")], + &[ + ("$first_new_age", 99), + ("$second_threshold", 50), + ("$second_new_age", 10), + ], + ), ), ) .await .unwrap(); + // Field-level-update fallback pin: two updates on one table stage FULL rows — a + // later update's read-your-writes scan must see the earlier update's + // full effect, and two partial batches with different assigned sets + // cannot share one uniform-schema merge source. + let shapes = probes.merge_shapes(); + assert_eq!(shapes.len(), 1, "chained updates coalesce into one staged merge"); + assert!( + shapes[0].source_columns.contains(&"name".to_string()), + "chained updates stage full rows (fallback), got {:?}", + shapes[0].source_columns + ); + // After op-1: Alice = 99. After op-2 (where age > 50): Alice // matches (99 > 50) → set to 10. End state: Alice = 10. let batches = read_table(&db, "node:Person").await; @@ -2094,3 +2129,117 @@ async fn filtered_read_after_append_and_delete_is_consistent() { assert_eq!(got, expected, "filtered read for {name}"); } } + +// ─── Field-level update staging shape ────────────────────────────── + +/// Field-level-update acceptance (red → green): a query whose ONLY op on a table is one +/// `update` stages a PARTIAL source — exactly (merge key + assigned columns) — +/// as a matched-only merge (`WhenNotMatched::DoNothing`). The whole-row path +/// stages every column with `InsertAll`. +#[tokio::test] +async fn single_update_stages_partial_matched_only_source() { + use omnigraph::instrumentation::{MergeWriteProbes, with_merge_write_probes}; + let dir = tempfile::tempdir().unwrap(); + let db = init_and_load(&dir).await; + + let probes = MergeWriteProbes::default(); + with_merge_write_probes( + probes.clone(), + db.mutate( + "main", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "Alice")], &[("$age", 41)]), + ), + ) + .await + .unwrap(); + + let shapes = probes.merge_shapes(); + assert_eq!(shapes.len(), 1, "one staged merge for the touched table"); + let mut cols = shapes[0].source_columns.clone(); + cols.sort(); + assert_eq!( + cols, + vec!["age".to_string(), "id".to_string()], + "a sole update stages only (key + assigned) columns, got {:?}", + shapes[0].source_columns + ); + assert!( + !shapes[0].inserts_unmatched, + "a sole update stages a matched-only merge (WhenNotMatched::DoNothing)" + ); +} + +/// An update matching zero rows stages nothing and commits nothing — +/// unchanged by field-level updates (the early return precedes staging). +#[tokio::test] +async fn empty_match_update_stages_no_merge() { + use omnigraph::instrumentation::{MergeWriteProbes, with_merge_write_probes}; + let dir = tempfile::tempdir().unwrap(); + let db = init_and_load(&dir).await; + + let probes = MergeWriteProbes::default(); + let result = with_merge_write_probes( + probes.clone(), + db.mutate( + "main", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "nobody-here")], &[("$age", 1)]), + ), + ) + .await + .unwrap(); + + assert_eq!(result.affected_nodes, 0); + assert!(probes.merge_shapes().is_empty(), "no merge staged for an empty match"); +} + +const BLOB_DOC_SCHEMA: &str = r#" +node Doc { + name: String @key + content: Blob? +} +"#; + +const BLOB_DOC_MUTATIONS: &str = r#" +query insert_doc($name: String) { + insert Doc { name: $name } +} +query set_content($name: String, $c: Blob?) { + update Doc set { content: $c } where name = $name +} +"#; + +/// A sole update assigning `null` to an optional Blob is the historical no-op +/// (the blob column is omitted from the staged batch and the merge leaves it +/// untouched). The partial-update plan must mirror that omission: listing the +/// blob in the stage projection while `apply_assignments` omits it fails the +/// whole mutation with an internal error — and made the outcome depend on +/// whether unrelated statements shared the query (whole-row fallback +/// succeeded). Pinned: the sole-update path succeeds like the fallback does. +#[tokio::test] +async fn null_blob_assignment_in_sole_update_is_a_no_op() { + use omnigraph_compiler::query::ast::Literal; + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let db = Omnigraph::init(uri, BLOB_DOC_SCHEMA).await.unwrap(); + + db.mutate( + "main", + BLOB_DOC_MUTATIONS, + "insert_doc", + ¶ms(&[("$name", "d1")]), + ) + .await + .unwrap(); + + let mut p = params(&[("$name", "d1")]); + p.insert("c".to_string(), Literal::Null); + let result = db + .mutate("main", BLOB_DOC_MUTATIONS, "set_content", &p) + .await + .unwrap(); + assert_eq!(result.affected_nodes, 1); +} diff --git a/docs/dev/writes.md b/docs/dev/writes.md index 21227191..1cfa1467 100644 --- a/docs/dev/writes.md +++ b/docs/dev/writes.md @@ -229,6 +229,36 @@ shared by both `mutate_as` and the bulk loader: (below) still prevents inserts/updates from coexisting with deletes in one query. +- **Field-level updates (partial-schema staging).** A node table whose ONLY op + in the query is a single `update` stages a PARTIAL merge source — exactly + (id + assigned) columns — as a **matched-only** merge + (`WhenNotMatched::DoNothing`). The accumulated (pending) batch additionally + carries the constraint-completion columns (every member of a `@unique` group + intersecting the assigned set) so the end-of-query evaluator can validate the + whole tuple — but those are **validation inputs only** and are projected out + before staging: Lance counts every source column as modified, so staging an + unassigned group member would patch it and prune its index for no semantic + reason. Lance's partial-schema path patches the + provided columns in place on the same fragment (`update_mode: + RewriteColumns`), so the staged source is exactly (row key + assigned): + completion columns are read as validation-only inputs but never rewritten, + every other unassigned column is neither read nor rewritten, and indexes + over all unassigned columns keep fragment coverage (Lance prunes only + `fields_modified`). Every other shape (insert+update on one table, multiple + updates, etc.) falls back to whole-row staging: partial and full batches + cannot share one uniform-schema merge source (a present column's null cell + means "set NULL", so widening would null-overwrite), a later same-table op's + read-your-writes scan needs full rows, and one table commits at most one + version per query; blob-bearing tables also always take the whole-row path + (Lance's partial-source column patch trips the blob writer on blob-bearing + fragments — revisit when upstream accepts partial sources there). The unique + evaluator skips a `@unique` group with no column present in a batch + (untouched by the write) and stays loud on a partially-present group. Known residual (pinned by + `lance_surface_guards::partial_schema_merge_patches_in_place_and_prunes_only_modified_fields` + and the `scalar_indexes` coverage cell): the join key rides the source, so + the **id BTREE** alone still loses patched fragments (parity with the + whole-row path) until upstream excludes ON columns from column patches. + This upholds the manifest-atomic mutation and read-your-writes invariants tracked in [docs/dev/invariants.md](invariants.md). diff --git a/docs/user/mutations/index.md b/docs/user/mutations/index.md index afc02c35..f00011b0 100644 --- a/docs/user/mutations/index.md +++ b/docs/user/mutations/index.md @@ -35,6 +35,27 @@ failure leaves the graph untouched. See [transactions](../branching/transactions for the per-query atomicity contract and [branches](../branching/index.md) for multi-query workflows. +## Updates write only the columns they assign + +When an `update` is the only statement touching its type in a query, the engine +patches just the row key plus the assigned columns in place. Columns that +complete a `@unique` group the assignment touches may be **read** as +validation-only inputs (so the whole tuple is checked), but they are never +rewritten. All other unassigned columns — including large `Vector` embedding +columns — are neither read nor rewritten, and indexes over every unassigned +column keep their coverage. Semantics are identical to a whole-row update; the +difference is cost: update latency and write volume scale with what you assign, +not with row width. Types with a +`Blob` property currently always use the whole-row path (including the blob +materialization described above). + +Queries that combine an update with other statements on the same type fall back +to whole-row staging (same semantics, previous cost). Note that updating a +property that feeds an `@embed` column does **not** recompute the embedding — +re-run the embedding pass after changing embed-source text. + +## Concurrency + Concurrent changes use optimistic concurrency over the whole target branch. Insert/Merge/Append operations whose branch changed before physical effects are discarded and fully revalidated with a bounded internal retry. Strict