Scan an INSERT ... SELECT source in its own engine

nodedb/src/control/insert_select/expand_staged.rs

@@ -19,7 +19,7 @@ use nodedb_types::{DatabaseId, Surrogate, TenantId};
 
 use crate::bridge::envelope::PhysicalPlan;
 use crate::control::insert_select::copy_rows::{assign_page_rows, resolve_copy_spec};
-use crate::control::maintenance::clone_materializer::scan_source_page;
+use crate::control::maintenance::clone_materializer::scan_source_page_auto;
 use crate::control::state::SharedState;
 use crate::types::{TxnId, VShardId};
 use nodedb_physical::physical_plan::DocumentOp;
@@ -174,7 +174,7 @@ async fn materialize_copy(
     let mut rows: Vec<(String, Vec<u8>, Surrogate)> = Vec::new();
 
     while remaining > 0 {
-        let (entries, next_cursor) = scan_source_page(
+        let (entries, next_cursor) = scan_source_page_auto(
             state,
             tenant_id,
             database_id,

nodedb/src/control/insert_select/orchestrator.rs

@@ -19,7 +19,7 @@ use nodedb_types::{DatabaseId, Lsn, Surrogate, TenantId};
 
 use crate::bridge::envelope::{Payload, PhysicalPlan, Response, Status};
 use crate::control::insert_select::copy_rows::{assign_page_rows, resolve_copy_spec};
-use crate::control::maintenance::clone_materializer::{dispatch_local, scan_source_page};
+use crate::control::maintenance::clone_materializer::{dispatch_local, scan_source_page_auto};
 use crate::control::state::SharedState;
 use nodedb_physical::physical_plan::DocumentOp;
 
@@ -99,7 +99,7 @@ pub(crate) async fn run_insert_select(
 
     while remaining > 0 {
         // Phase 1: scan one source page (point-in-time snapshot).
-        let (entries, next_cursor) = scan_source_page(
+        let (entries, next_cursor) = scan_source_page_auto(
             state,
             tenant_id,
             database_id,

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

@@ -0,0 +1,90 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Engine-aware source scan for `INSERT ... SELECT`.
+//!
+//! The document materializer reads the document store; a kv collection keeps
+//! nothing there, so scanning a kv source with it materializes zero rows and
+//! the statement reports `INSERT 0 0`. Route on the source collection's own
+//! engine and normalize every page to the document entry shape
+//! `(doc_id, source_surrogate, value_bytes)`: the copy pipeline ignores the
+//! ids and shapes the body through its column map, and for kv the body is the
+//! stored msgpack row, so expression cells evaluate exactly as they do over a
+//! document source.
+
+use nodedb_sql::types::EngineType;
+
+use super::{document, kv};
+
+pub(crate) async fn scan_source_page(
+    state: &crate::control::state::SharedState,
+    tenant_id: nodedb_types::TenantId,
+    database_id: nodedb_types::DatabaseId,
+    source_qualified: &str,
+    cursor: &[u8],
+    system_as_of_ms: Option<i64>,
+    txn_id: Option<crate::types::TxnId>,
+) -> crate::Result<(Vec<(String, u32, Vec<u8>)>, Vec<u8>)> {
+    let catalog = state.credentials.catalog();
+    let stored = catalog
+        .get_collection(
+            database_id,
+            tenant_id.as_u64(),
+            &crate::control::target_identity::bare_collection_name(database_id, source_qualified),
+        )?
+        .ok_or_else(|| crate::Error::CollectionNotFound {
+            tenant_id,
+            collection: source_qualified.to_string(),
+        })?;
+    let (engine, _, _) =
+        crate::control::planner::catalog_adapter::type_convert::convert_collection_type(&stored);
+
+    match engine {
+        EngineType::DocumentSchemaless | EngineType::DocumentStrict => {
+            document::scan_source_page(
+                state,
+                tenant_id,
+                database_id,
+                source_qualified,
+                cursor,
+                system_as_of_ms,
+                txn_id,
+            )
+            .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?;
+            // 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)| {
+                    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))
+        }
+        // Refused by name: these engines have no INSERT ... SELECT source
+        // materializer, and a silent copy would read nothing.
+        EngineType::Columnar | EngineType::Timeseries | EngineType::Spatial | EngineType::Array => {
+            Err(crate::Error::PlanError {
+                detail: format!("INSERT ... SELECT from {engine:?} sources is not supported yet"),
+            })
+        }
+    }
+}

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

@@ -189,7 +189,7 @@ fn checkpoint_progress(
 /// Run one source-side `MaterializeScan` round-trip. Returns the entries in
 /// this page (raw `(key, value)` byte pairs) plus the next-cursor; the
 /// cursor is empty when the scan is complete.
-async fn scan_source_page(
+pub(crate) async fn scan_source_page(
     state: &SharedState,
     tenant_id: TenantId,
     source_db_id: DatabaseId,

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

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: BUSL-1.1
 
+mod auto_source;
 mod columnar;
 mod dispatch;
 mod document;
@@ -17,5 +18,6 @@ pub use walker::{
 
 // Shared with the `INSERT ... SELECT` orchestrator, which reuses the same
 // local-dispatch primitive and source-scan cursor decode.
+pub(crate) use auto_source::scan_source_page as scan_source_page_auto;
 pub(crate) use dispatch::{dispatch_local, dispatch_local_on_vshard};
-pub(crate) use document::{read_all_source_rows, scan_source_page};
+pub(crate) use document::read_all_source_rows;

nodedb/src/control/planner/catalog_adapter/mod.rs

@@ -33,6 +33,6 @@
 mod adapter;
 mod sequence_access;
 mod sql_catalog_impl;
-mod type_convert;
+pub(crate) mod type_convert;
 
 pub use adapter::OriginCatalog;

nodedb/src/control/planner/catalog_adapter/type_convert.rs

@@ -6,7 +6,7 @@ use nodedb_sql::types::{ColumnInfo, EngineType, SqlDataType};
 use nodedb_types::columnar::{FloatWidth, IntWidth};
 
 /// Convert a StoredCollection to engine type, columns, and primary key.
-pub(super) fn convert_collection_type(
+pub(crate) fn convert_collection_type(
     stored: &crate::control::security::catalog::StoredCollection,
 ) -> (EngineType, Vec<ColumnInfo>, Option<String>) {
     use nodedb_types::CollectionType;

nodedb/tests/wire/cases/insert_select_cross_engine.rs

@@ -307,3 +307,101 @@ 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:?}"
+    );
+}
+
+/// 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();
+}