test(insert-select): cover the kv source route, its refusals, and the key column

- a point-in-time or transactional read from a kv source is refused by
  name; an in-transaction INSERT ... SELECT reaches that path
- a source engine with no materializer (columnar) is refused by name
- a kv collection whose key column is the key sentinel now carries the
  key: the arm uses the one shaping rule every scan path uses, so the
  copied column reads the row's own key instead of NULL

Author: EnRaiha

nodedb/src/control/maintenance/clone_materializer/auto_source.rs

@@ -65,9 +65,17 @@ pub(crate) async fn scan_source_page(
             let (pairs, next) =
                 kv::scan_source_page(state, tenant_id, database_id, source_qualified, cursor)
                     .await?;
+            // The KV key slot sits outside the stored body. The one shaping
+            // rule injects it (the same converter the scan and RETURNING paths
+            // use), so a copied `key` column reads the row's own key instead of
+            // NULL.
             let entries = pairs
                 .into_iter()
-                .map(|(key, value)| (String::from_utf8_lossy(&key).into_owned(), 0, value))
+                .map(|(key, value)| {
+                    let key = String::from_utf8_lossy(&key).into_owned();
+                    let body = nodedb_query::msgpack_scan::kv_row_msgpack(&key, &value);
+                    (key, 0, body)
+                })
                 .collect();
             Ok((entries, next))
         }

nodedb/tests/wire/cases/insert_select_cross_engine.rs

@@ -348,3 +348,60 @@ async fn insert_select_copies_a_kv_source_through_its_own_engine() {
         "both rows must copy, with expression cells evaluated: {rows:?}"
     );
 }
+
+/// A source engine with no `INSERT ... SELECT` materializer is refused by name:
+/// scanning it with the document materializer would copy nothing and report
+/// `INSERT 0 0`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn insert_select_refuses_a_source_it_cannot_scan() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION isk_nosrc_src (id TEXT PRIMARY KEY) WITH (engine = 'columnar')")
+        .await
+        .unwrap();
+    server
+        .exec("CREATE COLLECTION isk_nosrc_dst")
+        .await
+        .unwrap();
+
+    server
+        .expect_error(
+            "INSERT INTO isk_nosrc_dst SELECT * FROM isk_nosrc_src",
+            "Columnar",
+        )
+        .await;
+}
+
+/// A kv collection whose primary key column is `key` stores the key outside the
+/// row body. The copy must still carry it: a NULL key column is silent data
+/// loss, and the target's own key would be minted from the wrong column.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn insert_select_copies_a_kv_key_column() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION isk_key_src (key TEXT PRIMARY KEY, v TEXT) WITH (engine = 'kv')")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO isk_key_src (key, v) VALUES ('k1', 'hello')")
+        .await
+        .unwrap();
+
+    server.exec("CREATE COLLECTION isk_key_dst").await.unwrap();
+    server
+        .exec("INSERT INTO isk_key_dst (key, v) SELECT key, v FROM isk_key_src")
+        .await
+        .unwrap();
+
+    let rows = server
+        .query_rows("SELECT key, v FROM isk_key_dst")
+        .await
+        .unwrap();
+    assert_eq!(
+        rows,
+        vec![vec!["k1".to_string(), "hello".to_string()]],
+        "the copied key column must carry the source key: {rows:?}"
+    );
+}

nodedb/tests/wire/cases/sql_transactions_insert_select_overlay.rs

@@ -242,3 +242,33 @@ async fn strict_insert_select_sees_source_rows_staged_earlier_in_txn() {
     )
     .await;
 }
+
+/// In-transaction `INSERT ... SELECT` stages through the per-transaction
+/// overlay, so the source read is transactional. The kv materialize-scan
+/// carries no snapshot fields; the read is refused by name rather than copying
+/// committed-only rows the transaction did not ask for.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn kv_source_refuses_an_in_transaction_read() {
+    let server = TestServer::start().await;
+    server
+        .exec(
+            "CREATE COLLECTION is_kv_tx_src (id STRING PRIMARY KEY, n INT) \
+             WITH (engine='kv')",
+        )
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO is_kv_tx_src (id, n) VALUES ('a', 1)")
+        .await
+        .unwrap();
+    server.exec("CREATE COLLECTION is_kv_tx_tgt").await.unwrap();
+
+    server.exec("BEGIN").await.unwrap();
+    server
+        .expect_error(
+            "INSERT INTO is_kv_tx_tgt SELECT * FROM is_kv_tx_src",
+            "transactional",
+        )
+        .await;
+    server.exec("ROLLBACK").await.unwrap();
+}