diff --git a/nodedb/src/control/insert_select/expand_staged.rs b/nodedb/src/control/insert_select/expand_staged.rs index 583c42011..ea2f7cada 100644 --- a/nodedb/src/control/insert_select/expand_staged.rs +++ b/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, 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, diff --git a/nodedb/src/control/insert_select/orchestrator.rs b/nodedb/src/control/insert_select/orchestrator.rs index 0c2d5c442..7208d5145 100644 --- a/nodedb/src/control/insert_select/orchestrator.rs +++ b/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, diff --git a/nodedb/src/control/maintenance/clone_materializer/auto_source.rs b/nodedb/src/control/maintenance/clone_materializer/auto_source.rs new file mode 100644 index 000000000..ccb70f082 --- /dev/null +++ b/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, + txn_id: Option, +) -> crate::Result<(Vec<(String, u32, Vec)>, Vec)> { + 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"), + }) + } + } +} diff --git a/nodedb/src/control/maintenance/clone_materializer/kv.rs b/nodedb/src/control/maintenance/clone_materializer/kv.rs index abe438050..b930bc09d 100644 --- a/nodedb/src/control/maintenance/clone_materializer/kv.rs +++ b/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, diff --git a/nodedb/src/control/maintenance/clone_materializer/mod.rs b/nodedb/src/control/maintenance/clone_materializer/mod.rs index 23f0ef7ca..163bd07d9 100644 --- a/nodedb/src/control/maintenance/clone_materializer/mod.rs +++ b/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; diff --git a/nodedb/src/control/planner/catalog_adapter/mod.rs b/nodedb/src/control/planner/catalog_adapter/mod.rs index fe5c8beca..f0bb06dd1 100644 --- a/nodedb/src/control/planner/catalog_adapter/mod.rs +++ b/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; diff --git a/nodedb/src/control/planner/catalog_adapter/type_convert.rs b/nodedb/src/control/planner/catalog_adapter/type_convert.rs index 7438b2666..e13742daf 100644 --- a/nodedb/src/control/planner/catalog_adapter/type_convert.rs +++ b/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, Option) { use nodedb_types::CollectionType; diff --git a/nodedb/tests/wire/cases/insert_select_cross_engine.rs b/nodedb/tests/wire/cases/insert_select_cross_engine.rs index fb75a1378..feed42cd0 100644 --- a/nodedb/tests/wire/cases/insert_select_cross_engine.rs +++ b/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:?}" + ); +} diff --git a/nodedb/tests/wire/cases/sql_transactions_insert_select_overlay.rs b/nodedb/tests/wire/cases/sql_transactions_insert_select_overlay.rs index fef06e230..b6a9c3f18 100644 --- a/nodedb/tests/wire/cases/sql_transactions_insert_select_overlay.rs +++ b/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(); +}