feat(executor): version-aware binary tuple decoding for strict collections

farhan-syah · farhan-syah · commit 93139e22eb50 · 2026-04-17T03:59:53.000+08:00
`binary_tuple_to_value` now detects when a stored tuple's schema version
is behind the current catalog version and decodes using
`schema_for_version` to match the physical column layout at write time.
Columns added after the tuple's version are filled with their DEFAULT
value (or NULL) rather than causing an index-out-of-bounds or
returning corrupt data.

`binary_tuple_to_json` is refactored to delegate to
`binary_tuple_to_value` so the version-aware path is shared across both
read modes without duplication.

Remaining `StrictSchema` construction sites in the executor initialise
`dropped_columns` to keep all call sites consistent.

Add integration tests covering the full lifecycle of schema-altering DDL
on a strict collection:
- Pre-ALTER rows return correct values for existing columns after ADD COLUMN
- Pre-ALTER rows return the column DEFAULT for newly added columns
- Updating a pre-ALTER row migrates it to the current schema
- DROP COLUMN leaves pre-drop rows readable for surviving columns
- Multiple ADD COLUMN operations in sequence remain readable
- RENAME COLUMN and ALTER COLUMN TYPE on pre-existing rows
diff --git a/nodedb/src/data/executor/handlers/convert.rs b/nodedb/src/data/executor/handlers/convert.rs
@@ -105,6 +105,7 @@ impl CoreLoop {
         let schema = StrictSchema {
             columns,
             version: 1,
+            dropped_columns: Vec::new(),
         };
 
         // Scan all existing documents.
diff --git a/nodedb/src/data/executor/handlers/document/read.rs b/nodedb/src/data/executor/handlers/document/read.rs
@@ -626,6 +626,7 @@ mod tests {
                 ColumnDef::nullable("age", ColumnType::Int64),
             ],
             version: 1,
+            dropped_columns: Vec::new(),
         };
         let mut map = std::collections::HashMap::new();
         map.insert("id".into(), Value::String("u1".into()));
diff --git a/nodedb/src/data/executor/strict_format.rs b/nodedb/src/data/executor/strict_format.rs
@@ -85,12 +85,37 @@ pub(super) fn binary_tuple_to_value(tuple_bytes: &[u8], schema: &StrictSchema) -
         return None;
     }
 
-    let values = decoder.extract_all(tuple_bytes).ok()?;
-
+    // Version-aware decoding: if the tuple was written with an older schema
+    // (fewer columns due to ADD COLUMN), build a sub-schema decoder matching
+    // the physical layout and fill defaults for new columns.
     let mut map = std::collections::HashMap::with_capacity(schema.columns.len());
-    for (i, col) in schema.columns.iter().enumerate() {
-        map.insert(col.name.clone(), values[i].clone());
+    if version < schema.version {
+        let old_schema = schema.schema_for_version(version);
+        let old_decoder = nodedb_strict::TupleDecoder::new(&old_schema);
+        let old_values = old_decoder.extract_all(tuple_bytes).ok()?;
+
+        // Map old columns by name.
+        for (i, col) in old_schema.columns.iter().enumerate() {
+            map.insert(col.name.clone(), old_values[i].clone());
+        }
+        // Fill defaults for columns added after this tuple's version.
+        for col in &schema.columns {
+            if col.added_at_version > version {
+                let default_val = col
+                    .default
+                    .as_deref()
+                    .map(StrictSchema::parse_default_literal)
+                    .unwrap_or(Value::Null);
+                map.insert(col.name.clone(), default_val);
+            }
+        }
+    } else {
+        let values = decoder.extract_all(tuple_bytes).ok()?;
+        for (i, col) in schema.columns.iter().enumerate() {
+            map.insert(col.name.clone(), values[i].clone());
+        }
     }
+
     Some(Value::Object(map))
 }
 
@@ -108,14 +133,20 @@ pub(super) fn binary_tuple_to_json(
     tuple_bytes: &[u8],
     schema: &StrictSchema,
 ) -> Option<serde_json::Value> {
-    let decoder = nodedb_strict::TupleDecoder::new(schema);
-    let values = decoder.extract_all(tuple_bytes).ok()?;
-
-    let mut obj = serde_json::Map::with_capacity(schema.columns.len());
-    for (i, col) in schema.columns.iter().enumerate() {
-        obj.insert(col.name.clone(), value_to_json(&values[i]));
+    // Delegate to binary_tuple_to_value (which handles version-aware decoding)
+    // then convert Value → JSON.
+    let val = binary_tuple_to_value(tuple_bytes, schema)?;
+    match val {
+        Value::Object(map) => {
+            let mut obj = serde_json::Map::with_capacity(map.len());
+            for col in &schema.columns {
+                let v = map.get(&col.name).unwrap_or(&Value::Null);
+                obj.insert(col.name.clone(), value_to_json(v));
+            }
+            Some(serde_json::Value::Object(obj))
+        }
+        _ => None,
     }
-    Some(serde_json::Value::Object(obj))
 }
 
 /// Coerce a `nodedb_types::Value` to match a column's declared type.
@@ -372,6 +403,7 @@ mod tests {
                 ColumnDef::nullable("age", ColumnType::Int64),
             ],
             version: 1,
+            dropped_columns: Vec::new(),
         }
     }
 
diff --git a/nodedb/tests/executor_tests/test_cross_type_join.rs b/nodedb/tests/executor_tests/test_cross_type_join.rs
@@ -96,6 +96,7 @@ fn document_scan_preserves_kv_rows_when_collection_has_strict_config() {
                         ColumnDef::nullable("lang", ColumnType::String),
                     ],
                     version: 1,
+                    dropped_columns: Vec::new(),
                 },
             },
             enforcement: Box::new(EnforcementOptions::default()),
diff --git a/nodedb/tests/sql_transactions.rs b/nodedb/tests/sql_transactions.rs
@@ -226,3 +226,275 @@ async fn alter_collection_alter_column_type() {
         .unwrap();
     assert_eq!(rows.len(), 1);
 }
+
+// ── Pre-ALTER row survival tests ──────────────────────────────────────
+//
+// Every test below verifies that rows written BEFORE a schema-altering DDL
+// remain readable with correct values AFTER the DDL. The bug class is:
+// catalog schema mutated without row migration or read-time compat shim.
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn add_column_preserves_pre_alter_row_existing_columns() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION ac_preserve TYPE DOCUMENT STRICT (id TEXT PRIMARY KEY, name TEXT)")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO ac_preserve (id, name) VALUES ('a', 'alice')")
+        .await
+        .unwrap();
+
+    server
+        .exec("ALTER TABLE ac_preserve ADD COLUMN note TEXT DEFAULT 'n/a'")
+        .await
+        .unwrap();
+
+    // Pre-ALTER row must still return correct values for original columns.
+    let rows = server
+        .query_text("SELECT id, name FROM ac_preserve WHERE id = 'a'")
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 1, "pre-ALTER row must be visible");
+    assert!(
+        rows[0].contains("\"name\":\"alice\""),
+        "original column 'name' must retain its value, got {:?}",
+        rows[0]
+    );
+    // Regression guard: must NOT return null-everywhere.
+    assert!(
+        !rows[0].contains("\"name\":null"),
+        "pre-ALTER row must not have null-everywhere corruption, got {:?}",
+        rows[0]
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn add_column_returns_default_for_pre_alter_row() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION ac_default TYPE DOCUMENT STRICT (id TEXT PRIMARY KEY, name TEXT)")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO ac_default (id, name) VALUES ('a', 'alice')")
+        .await
+        .unwrap();
+
+    server
+        .exec("ALTER TABLE ac_default ADD COLUMN note TEXT DEFAULT 'n/a'")
+        .await
+        .unwrap();
+
+    // The new column should virtual-fill with its DEFAULT for pre-ALTER rows.
+    let rows = server
+        .query_text("SELECT id, name, note FROM ac_default WHERE id = 'a'")
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 1, "pre-ALTER row must be visible");
+    assert!(
+        rows[0].contains("\"note\":\"n/a\""),
+        "new column must return DEFAULT value for pre-ALTER rows, got {:?}",
+        rows[0]
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn add_column_then_update_pre_alter_row() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION ac_update TYPE DOCUMENT STRICT (id TEXT PRIMARY KEY, name TEXT)")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO ac_update (id, name) VALUES ('a', 'alice')")
+        .await
+        .unwrap();
+
+    server
+        .exec("ALTER TABLE ac_update ADD COLUMN note TEXT DEFAULT 'n/a'")
+        .await
+        .unwrap();
+
+    // Updating a pre-ALTER row must succeed and preserve all columns.
+    server
+        .exec("UPDATE ac_update SET note = 'updated' WHERE id = 'a'")
+        .await
+        .unwrap();
+
+    let rows = server
+        .query_text("SELECT id, name, note FROM ac_update WHERE id = 'a'")
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+    assert!(
+        rows[0].contains("\"name\":\"alice\""),
+        "original column must survive update, got {:?}",
+        rows[0]
+    );
+    assert!(
+        rows[0].contains("\"note\":\"updated\""),
+        "updated column must reflect new value, got {:?}",
+        rows[0]
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn multiple_add_columns_preserves_pre_alter_row() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION ac_multi TYPE DOCUMENT STRICT (id TEXT PRIMARY KEY, name TEXT)")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO ac_multi (id, name) VALUES ('a', 'alice')")
+        .await
+        .unwrap();
+
+    server
+        .exec("ALTER TABLE ac_multi ADD COLUMN col1 INT DEFAULT 0")
+        .await
+        .unwrap();
+    server
+        .exec("ALTER TABLE ac_multi ADD COLUMN col2 TEXT DEFAULT 'x'")
+        .await
+        .unwrap();
+
+    // Two sequential ADD COLUMNs compound the schema drift — pre-ALTER row
+    // must still be readable with correct values and defaults.
+    let rows = server
+        .query_text("SELECT id, name, col1, col2 FROM ac_multi WHERE id = 'a'")
+        .await
+        .unwrap();
+    assert_eq!(
+        rows.len(),
+        1,
+        "pre-ALTER row must be visible after two ADD COLUMNs"
+    );
+    assert!(
+        rows[0].contains("\"name\":\"alice\""),
+        "original column must retain value, got {:?}",
+        rows[0]
+    );
+    // Regression guard: null-everywhere means total schema-data offset corruption.
+    assert!(
+        !rows[0].contains("\"name\":null"),
+        "must not exhibit null-everywhere corruption, got {:?}",
+        rows[0]
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn drop_column_preserves_pre_alter_row_remaining_columns() {
+    let server = TestServer::start().await;
+
+    server
+        .exec(
+            "CREATE COLLECTION dc_preserve TYPE DOCUMENT STRICT (\
+                id TEXT PRIMARY KEY, \
+                name TEXT NOT NULL, \
+                scratch TEXT)",
+        )
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO dc_preserve (id, name, scratch) VALUES ('a', 'alice', 'temp')")
+        .await
+        .unwrap();
+
+    server
+        .exec("ALTER COLLECTION dc_preserve DROP COLUMN scratch")
+        .await
+        .unwrap();
+
+    // Remaining columns of the pre-ALTER row must read correctly.
+    let rows = server
+        .query_text("SELECT id, name FROM dc_preserve WHERE id = 'a'")
+        .await
+        .unwrap();
+    assert_eq!(
+        rows.len(),
+        1,
+        "pre-ALTER row must be visible after DROP COLUMN"
+    );
+    assert!(
+        rows[0].contains("\"name\":\"alice\""),
+        "remaining column 'name' must retain its value after DROP COLUMN, got {:?}",
+        rows[0]
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn rename_column_preserves_pre_alter_row_value() {
+    let server = TestServer::start().await;
+
+    server
+        .exec(
+            "CREATE COLLECTION rc_preserve TYPE DOCUMENT STRICT (\
+                id TEXT PRIMARY KEY, \
+                name TEXT NOT NULL, \
+                score INT DEFAULT 0)",
+        )
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO rc_preserve (id, name, score) VALUES ('a', 'alice', 42)")
+        .await
+        .unwrap();
+
+    server
+        .exec("ALTER COLLECTION rc_preserve RENAME COLUMN score TO points")
+        .await
+        .unwrap();
+
+    // Pre-ALTER row must be readable under the new column name with correct value.
+    let rows = server
+        .query_text("SELECT id, name, points FROM rc_preserve WHERE id = 'a'")
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+    assert!(
+        rows[0].contains("\"points\":42") || rows[0].contains("\"points\": 42"),
+        "renamed column must retain pre-ALTER value, got {:?}",
+        rows[0]
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn alter_column_type_preserves_pre_alter_row_value() {
+    let server = TestServer::start().await;
+
+    server
+        .exec(
+            "CREATE COLLECTION at_preserve TYPE DOCUMENT STRICT (\
+                id TEXT PRIMARY KEY, \
+                value INT NOT NULL)",
+        )
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO at_preserve (id, value) VALUES ('a', 42)")
+        .await
+        .unwrap();
+
+    server
+        .exec("ALTER COLLECTION at_preserve ALTER COLUMN value TYPE BIGINT")
+        .await
+        .unwrap();
+
+    // Pre-ALTER row must still read correctly after type widening.
+    let rows = server
+        .query_text("SELECT id, value FROM at_preserve WHERE id = 'a'")
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+    assert!(
+        rows[0].contains("\"value\":42") || rows[0].contains("\"value\": 42"),
+        "value must survive ALTER COLUMN TYPE, got {:?}",
+        rows[0]
+    );
+}
