Method insert

nodedb-columnar/src/mutation/write.rs:30–72

Insert a row with upsert-on-duplicate semantics. Returns WAL records to persist. Validates schema. If the PK already exists, the prior row is tombstoned via the segment's delete bitmap (a single positional delete) before the new row is appended to the memtable. The PK index is rebound to the new row location. This matches the ClickHouse / Iceberg "sparse PK + positional delete" model and keeps `S

(&mut self, values: &[Value])

    /// want `ON CONFLICT DO NOTHING` semantics should use
    /// [`Self::insert_if_absent`].
    pub fn insert(&mut self, values: &[Value]) -> Result<MutationResult, ColumnarError> {
        let pk_bytes = self.extract_pk_bytes(values)?;
        let mut wal_records = Vec::with_capacity(2);

        // Bitemporal collections preserve every version of a PK: each
        // write appends a new row with a distinct `_ts_system` stamp and
        // the prior row stays visible to `AS OF` queries. Skipping the
        // upsert-tombstone here keeps compaction lossless without
        // needing a separate "version-aware" delete bitmap. The PK
        // index is still rebound below so current-state reads see the
        // latest version.
        let bitemporal = self.schema.is_bitemporal();

        // If a prior row exists for this PK, tombstone it in place so
        // subsequent scans skip the stale row. The PK index is rebound
        // below to the freshly-appended row.
        if !bitemporal && let Some(prior) = self.pk_index.get(&pk_bytes).copied() {
            let bitmap = self.delete_bitmaps.entry(prior.segment_id).or_default();
            bitmap.mark_deleted(prior.row_index);
            wal_records.push(ColumnarWalRecord::DeleteRows {
                collection: self.collection.clone(),
                segment_id: prior.segment_id,
                row_indices: vec![prior.row_index],
            });
        }

        let row_data = encode_row_for_wal(values)?;
        wal_records.push(ColumnarWalRecord::InsertRow {
            collection: self.collection.clone(),
            row_data,
        });

        self.memtable.append_row(values)?;
        let location = RowLocation {
            segment_id: self.memtable_segment_id,
            row_index: self.memtable_row_counter,
        };
        self.pk_index.upsert(pk_bytes, location);
        self.memtable_surrogates.push(None);
        self.memtable_row_counter += 1;

        Ok(MutationResult { wal_records })
    }

    /// Insert with a stable cross-engine surrogate identity.
    ///