test(insert-select): cover a kv-engine source through its own route

The test copies two kv rows into a document target with an expression cell
and asserts both the row count and the evaluated cells. The kv arm also
refuses a point-in-time or transactional read by name: the materialize-scan
carries no snapshot fields, so the read had nothing to honor.

Author: EnRaiha

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

@@ -52,6 +52,16 @@ pub(crate) async fn scan_source_page(
             .await
         }
         EngineType::KeyValue => {
+            // The kv materialize-scan carries no snapshot fields, so a
+            // point-in-time or transactional read has nothing to thread into.
+            // Refusing by name beats copying rows the caller did not ask for.
+            if system_as_of_ms.is_some() || txn_id.is_some() {
+                return Err(crate::Error::PlanError {
+                    detail: "a point-in-time or transactional read is not supported \
+                             for an INSERT ... SELECT kv source"
+                        .to_string(),
+                });
+            }
             let (pairs, next) =
                 kv::scan_source_page(state, tenant_id, database_id, source_qualified, cursor)
                     .await?;

nodedb/tests/wire/cases/insert_select_cross_engine.rs

@@ -307,3 +307,44 @@ async fn insert_select_from_strict_source_normalizes_and_resolves() {
         "vector search must resolve the copied strict-source 'alpha'; got {near_e1:?}"
     );
 }
+
+/// A kv-engine source keeps nothing in the document store, so the document
+/// materializer read zero rows and the statement reported `INSERT 0 0`. The
+/// route scans the kv source through its own engine, and the copied rows must
+/// carry expression cells with the same semantics as a document source.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn insert_select_copies_a_kv_source_through_its_own_engine() {
+    let server = TestServer::start().await;
+
+    server
+        .exec("CREATE COLLECTION isk_src (id BIGINT PRIMARY KEY, v TEXT) WITH (engine = 'kv')")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO isk_src (id, v) VALUES (1, 'hello')")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO isk_src (id, v) VALUES (2, 'world')")
+        .await
+        .unwrap();
+
+    server.exec("CREATE COLLECTION isk_dst").await.unwrap();
+    server
+        .exec("INSERT INTO isk_dst (id, v) SELECT id, upper(v) FROM isk_src")
+        .await
+        .unwrap();
+
+    let rows = server
+        .query_rows("SELECT id, v FROM isk_dst ORDER BY id")
+        .await
+        .unwrap();
+    assert_eq!(
+        rows,
+        vec![
+            vec!["1".to_string(), "HELLO".to_string()],
+            vec!["2".to_string(), "WORLD".to_string()],
+        ],
+        "both rows must copy, with expression cells evaluated: {rows:?}"
+    );
+}