rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-08-18 03:58:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: submit generated column creation jobs

Browse Source
This commit is contained in:
Xuanwo
2026-08-12 10:58:04 +08:00
parent ca1d04db25
commit 6ed1a25439
4 changed files with 896 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/lancedb/src/remote/job.rs
+55
View File
@@ -84,6 +84,7 @@ pub(super) fn job_state_to_client(state: &str) -> String {
/// Closed current no-result job_type vocabulary.
const KNOWN_NO_RESULT_JOB_TYPES: &[&str] = &[
"create_index",
"create_gen_column",
"reindex_ivf_pq",
"reindex_ivf_flat",
"reindex_ivf_rq",
@@ -550,6 +551,7 @@ mod tests {
/// Closed current no-result job_type vocabulary.
const KNOWN_NO_RESULT_JOB_TYPES: &[&str] = &[
"create_index",
"create_gen_column",
"reindex_ivf_pq",
"reindex_ivf_flat",
"reindex_ivf_rq",
@@ -767,6 +769,59 @@ mod tests {
wait_expect_http(&job).await;
}
/// create_gen_column DONE with omitted or explicit None projects JobResult::None.
#[tokio::test]
async fn remote_job_result_create_gen_column_done_omitted_or_explicit_none_projects_none() {
for result_field in [
JsonField::Absent,
JsonField::Present(job_result_none_wire()),
] {
let body = describe_body(
"job-create-gen-column",
"DONE",
JsonField::Present(Value::String("create_gen_column".into())),
result_field,
);
let job = remote_job_with_describe_body("job-create-gen-column", body);
wait_expect_none(&job).await;
}
}
/// create_gen_column DONE with Function or unknown result kind is protocol Http.
#[tokio::test]
async fn remote_job_result_create_gen_column_function_or_unknown_result_is_http() {
let function = sample_remote_function();
let mut unknown_kind = job_result_none_wire();
unknown_kind["kind"] = Value::String("artifact".into());
let cases = [
(
"function",
JsonField::Present(job_result_function_wire(&function)),
),
("unknown_kind", JsonField::Present(unknown_kind)),
];
let mut unexpected = Vec::new();
for (label, result_field) in cases {
let body = describe_body(
"job-create-gen-column-bad",
"DONE",
JsonField::Present(Value::String("create_gen_column".into())),
result_field,
);
let job = remote_job_with_describe_body("job-create-gen-column-bad", body);
match job.wait().await {
Err(Error::Http { .. }) => {}
Ok(value) => unexpected.push(format!("{label}: Ok({value:?})")),
Err(other) => unexpected.push(format!("{label}: Err({other:?})")),
}
}
assert!(
unexpected.is_empty(),
"create_gen_column Function/unknown result must be Error::Http: {unexpected:?}"
);
}
/// Unknown job_type fails wait; empty historical job_type succeeds only without an explicit result.
#[tokio::test]
async fn remote_job_result_unknown_or_empty_job_type_expectation() {
rust/lancedb/src/remote/table.rs
+684 -1
View File
@@ -12,7 +12,7 @@ use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE};
use crate::blob::BlobFile;
use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions};
use crate::expr::expr_to_sql_string;
use crate::function::GeneratedColumnBindingSnapshot;
use crate::function::{CreateGeneratedColumnJobSpec, GeneratedColumnBindingSnapshot};
use crate::index::Index;
use crate::index::IndexStatistics;
use crate::index::waiter::wait_for_index;
@@ -1937,6 +1937,72 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
)
}
async fn submit_create_generated_column(
&self,
source_table_version: u64,
spec: CreateGeneratedColumnJobSpec,
) -> Result<Job> {
// Fixed-version handles fail before any submit request.
self.check_mutable().await?;
// Envelope keys are exactly source_table_version + nested FF-009 spec.
// Branch identity (when present) is additive table identity, not part of
// the operation payload. table_ref is derived server-side.
let mut body = serde_json::json!({
"source_table_version": source_table_version,
"spec": spec,
});
self.apply_branch_body(&mut body);
let request = self
.client
.post(&format!(
"/v1/table/{}/generated_columns/create/",
self.identifier
))
.json(&body);
// Typed literals may carry business data: reuse sensitive retry/check.
let (request_id, response) = self
.client
.send_sensitive_with_retry(request, None, true)
.await?;
let response = self
.client
.check_sensitive_response(&request_id, response)
.await?;
// Payload-free protocol failure: never fold response bytes into Error::Http.
let bytes = response.bytes().await.err_to_http(request_id.clone())?;
let value: serde_json::Value = match serde_json::from_slice(&bytes) {
Ok(value) => value,
Err(_) => {
return Err(Error::Http {
source: "create generated column response is not valid JSON".into(),
request_id,
status_code: None,
});
}
};
let job_id = match value.get("job_id") {
Some(serde_json::Value::String(job_id)) if !job_id.is_empty() => job_id.clone(),
_ => {
return Err(Error::Http {
source: "create generated column response missing or invalid job_id".into(),
request_id,
status_code: None,
});
}
};
// Submit acceptance is not table mutation: do not touch schema cache or
// read/write freshness watermarks.
Ok(Job::new(Box::new(RemoteJob::new(
self.client.clone(),
job_id,
))))
}
async fn create_branch(
&self,
name: &str,
@@ -8108,6 +8174,623 @@ mod tests {
}
}
// -------------------------------------------------------------------------
// CreateGeneratedColumnJobSpec remote table submit transport (FF-031)
// -------------------------------------------------------------------------
const CREATE_GEN_COL_LITERAL_MARKER: &str = "SENSITIVE_CREATE_GEN_COL_LITERAL_MARKER";
const CREATE_GEN_COL_RESPONSE_MARKER: &str = "SENSITIVE_CREATE_GEN_COL_RESPONSE_MARKER";
fn sample_create_generated_column_job_spec() -> CreateGeneratedColumnJobSpec {
use crate::function::{
Function, FunctionArgument, FunctionCall, FunctionId, FunctionOutput,
FunctionParameter, FunctionSignature,
};
use arrow_array::StringArray;
let function = Function::new(
FunctionId::try_new("fn.exact.create-submit.remote").expect("valid FunctionId"),
FunctionSignature::try_new(
vec![
FunctionParameter::new("x", DataType::Int32),
FunctionParameter::new("label", DataType::Utf8),
],
FunctionOutput::new(DataType::Int32, true),
)
.expect("valid FunctionSignature"),
);
let call = FunctionCall::try_new(
&function,
vec![
(
"x".to_string(),
FunctionArgument::try_field(1, DataType::Int32).expect("field arg"),
),
(
"label".to_string(),
FunctionArgument::try_literal(Arc::new(StringArray::from(vec![Some(
CREATE_GEN_COL_LITERAL_MARKER,
)])) as _)
.expect("literal arg"),
),
],
)
.expect("valid FunctionCall");
CreateGeneratedColumnJobSpec::try_new("normalized", &function, call).expect("valid spec")
}
fn create_gen_col_error_chain_text(err: &Error) -> String {
let mut text = format!("{err}\n{err:?}");
let mut current: Option<&(dyn std::error::Error + 'static)> = Some(err);
while let Some(e) = current {
text.push('\n');
text.push_str(&e.to_string());
text.push('\n');
text.push_str(&format!("{e:?}"));
current = e.source();
}
text
}
fn assert_create_gen_col_markers_absent(err: &Error) {
let text = create_gen_col_error_chain_text(err);
assert!(
!text.contains(CREATE_GEN_COL_LITERAL_MARKER),
"request literal marker must be absent from error/debug/source chain: {text}"
);
assert!(
!text.contains(CREATE_GEN_COL_RESPONSE_MARKER),
"response marker must be absent from error/debug/source chain: {text}"
);
}
fn assert_create_gen_col_envelope(
body: &serde_json::Value,
expected_source_version: u64,
expected_spec: &CreateGeneratedColumnJobSpec,
expected_branch: Option<&str>,
) {
let object = body
.as_object()
.expect("create generated column body must be a JSON object");
let mut expected_keys = vec!["source_table_version", "spec"];
if expected_branch.is_some() {
expected_keys.push("branch");
}
expected_keys.sort_unstable();
let mut actual_keys: Vec<&str> = object.keys().map(|k| k.as_str()).collect();
actual_keys.sort_unstable();
assert_eq!(
actual_keys, expected_keys,
"envelope keys must be exact; got {actual_keys:?}"
);
assert_eq!(
body["source_table_version"],
serde_json::json!(expected_source_version),
"source_table_version must be preserved exactly"
);
assert!(
body.get("table_ref").is_none(),
"table_ref must not be sent; server derives it"
);
match expected_branch {
Some(branch) => assert_eq!(body["branch"], branch),
None => assert!(body.get("branch").is_none()),
}
let nested = body.get("spec").expect("nested spec");
let nested_object = nested.as_object().expect("spec must be object");
let mut nested_keys: Vec<&str> = nested_object.keys().map(|k| k.as_str()).collect();
nested_keys.sort_unstable();
assert_eq!(
nested_keys,
vec!["column_name", "format_version", "function_call"],
"nested FF-009 spec keys must be exact"
);
let expected_spec_json =
serde_json::to_value(expected_spec).expect("serialize CreateGeneratedColumnJobSpec");
assert_eq!(
nested, &expected_spec_json,
"nested spec must be the exact FF-009 CreateGeneratedColumnJobSpec wire"
);
for forbidden in [
"table_ref",
"source_table_version",
"table",
"version",
"name",
"handle",
"output",
"output_field_id",
"dependency_epoch",
"materialized_epoch",
"artifact",
"job_id",
"idempotency_key",
"retry",
] {
assert!(
nested_object.get(forbidden).is_none(),
"nested spec must not contain forbidden key `{forbidden}`"
);
}
}
/// Remote main posts exact route and two-key envelope; returns opaque Job ID.
#[tokio::test]
async fn test_submit_create_generated_column_main_exact_route_envelope_and_job_id() {
let spec = sample_create_generated_column_job_spec();
let expected_spec = sample_create_generated_column_job_spec();
let paths = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let paths_ref = paths.clone();
let table = Table::new_with_handler("my_table", move |request| {
let path = request.url().path().to_string();
paths_ref.lock().unwrap().push(path.clone());
assert_eq!(request.method(), reqwest::Method::POST);
assert_eq!(path, "/v1/table/my_table/generated_columns/create/");
let body = request_body_json(&request);
assert_create_gen_col_envelope(&body, 42, &expected_spec, None);
http::Response::builder()
.status(200)
.body(r#"{"job_id":"job-create-gen-col-1","server_extra":{"ok":true}}"#)
.unwrap()
});
let job = table
.submit_create_generated_column(42, spec)
.await
.expect("main submit must succeed");
assert_eq!(job.id(), Some("job-create-gen-col-1"));
assert_eq!(
*paths.lock().unwrap(),
vec!["/v1/table/my_table/generated_columns/create/".to_string()],
"submit must issue exactly one create request and no describe/read"
);
}
/// Branch identity adds exact `branch` without changing source version or nested spec.
#[tokio::test]
async fn test_submit_create_generated_column_branch_adds_exact_branch() {
use lance::dataset::refs::Ref;
let spec = sample_create_generated_column_job_spec();
let expected_spec = sample_create_generated_column_job_spec();
let table =
Table::new_with_handler("my_table", move |request| match request.url().path() {
"/v1/table/my_table/branches/create/" => http::Response::builder()
.status(200)
.body("{}".to_string())
.unwrap(),
"/v1/table/my_table/generated_columns/create/" => {
let body = request_body_json(&request);
assert_create_gen_col_envelope(&body, 7, &expected_spec, Some("exp"));
http::Response::builder()
.status(200)
.body(r#"{"job_id":"job-create-gen-col-branch"}"#.to_string())
.unwrap()
}
path => panic!("unexpected path: {path}"),
});
let branch = table
.create_branch("exp", Ref::Version(None, None))
.await
.expect("create branch");
let job = branch
.submit_create_generated_column(7, spec)
.await
.expect("branch submit must succeed");
assert_eq!(job.id(), Some("job-create-gen-col-branch"));
}
/// Explicit-version handles fail mutation preflight before any submit request.
#[tokio::test]
async fn test_submit_create_generated_column_fixed_version_fails_before_submit() {
let describe_body =
describe_response(&Schema::new(vec![Field::new("a", DataType::Int32, false)]));
let submit_hits = Arc::new(AtomicUsize::new(0));
let submit_hits_ref = submit_hits.clone();
let table =
Table::new_with_handler("my_table", move |request| match request.url().path() {
"/v1/table/my_table/describe/" => http::Response::builder()
.status(200)
.body(describe_body.clone())
.unwrap(),
"/v1/table/my_table/generated_columns/create/" => {
submit_hits_ref.fetch_add(1, Ordering::SeqCst);
panic!("fixed-version handle must not reach submit");
}
path => panic!("unexpected path: {path}"),
});
table.checkout(42).await.expect("checkout fixed version");
let err = table
.submit_create_generated_column(42, sample_create_generated_column_job_spec())
.await
.expect_err("fixed-version submit must fail preflight");
assert!(
matches!(err, Error::NotSupported { .. }),
"expected NotSupported preflight, got {err:?}"
);
assert_eq!(
submit_hits.load(Ordering::SeqCst),
0,
"preflight must happen before any submit request"
);
}
/// Retry reuses request ID and byte-equivalent JSON body.
#[tokio::test]
async fn test_submit_create_generated_column_retry_preserves_request_id_and_body() {
use std::sync::OnceLock;
let spec = sample_create_generated_column_job_spec();
let seen_request_id = Arc::new(OnceLock::new());
let seen_request_id_ref = seen_request_id.clone();
let attempts = Arc::new(AtomicUsize::new(0));
let attempts_ref = attempts.clone();
let first_body = Arc::new(OnceLock::new());
let first_body_ref = first_body.clone();
let table = Table::new_with_handler_and_config(
"my_table",
move |request| {
assert_eq!(
request.url().path(),
"/v1/table/my_table/generated_columns/create/"
);
let body = request
.body()
.and_then(|b| b.as_bytes())
.expect("submit must carry JSON body")
.to_vec();
let actual: serde_json::Value =
serde_json::from_slice(&body).expect("body must be JSON");
assert_create_gen_col_envelope(
&actual,
11,
&sample_create_generated_column_job_spec(),
None,
);
let seen_body = first_body_ref.get_or_init(|| body.clone());
assert_eq!(
&body, seen_body,
"retry must reuse byte-equivalent JSON body"
);
let request_id = request.headers()["x-request-id"]
.to_str()
.unwrap()
.to_string();
assert!(!request_id.is_empty());
let seen = seen_request_id_ref.get_or_init(|| request_id.clone());
assert_eq!(
&request_id, seen,
"request id must be identical across retries"
);
let n = attempts_ref.fetch_add(1, Ordering::SeqCst);
if n == 0 {
http::Response::builder()
.status(500)
.body("transient create gen column failure")
.unwrap()
} else {
http::Response::builder()
.status(200)
.body(r#"{"job_id":"job-create-gen-col-retry"}"#)
.unwrap()
}
},
ClientConfig {
retry_config: RetryConfig {
retries: Some(2),
backoff_factor: Some(0.0),
backoff_jitter: Some(0.0),
..Default::default()
},
..Default::default()
},
);
let job = table
.submit_create_generated_column(11, spec)
.await
.expect("submit must succeed after one retry");
assert_eq!(job.id(), Some("job-create-gen-col-retry"));
assert_eq!(attempts.load(Ordering::SeqCst), 2);
assert!(seen_request_id.get().is_some());
}
/// Malformed/missing/empty/non-string job_id and 4xx/5xx markers stay payload-free.
#[tokio::test]
async fn test_submit_create_generated_column_errors_omit_request_and_response_markers() {
// Every 200 shape that fails closed includes the unique response marker
// wherever JSON permits, so absence from the error chain is meaningful.
let invalid_job_id_cases: Vec<(&str, String)> = vec![
(
"missing",
format!(r#"{{"server_extra":"{CREATE_GEN_COL_RESPONSE_MARKER}"}}"#),
),
(
"null",
format!(r#"{{"job_id":null,"server_extra":"{CREATE_GEN_COL_RESPONSE_MARKER}"}}"#),
),
(
"empty",
format!(r#"{{"job_id":"","server_extra":"{CREATE_GEN_COL_RESPONSE_MARKER}"}}"#),
),
(
"wrong_type",
format!(r#"{{"job_id":123,"server_extra":"{CREATE_GEN_COL_RESPONSE_MARKER}"}}"#),
),
(
"malformed",
format!("not-json {CREATE_GEN_COL_RESPONSE_MARKER}"),
),
];
let mut unexpected = Vec::new();
for (label, response_body) in invalid_job_id_cases {
let body_for_handler = response_body.clone();
let expected_request_id = Arc::new(std::sync::Mutex::new(None::<String>));
let expected_request_id_c = expected_request_id.clone();
let table = Table::new_with_handler("my_table", move |request| {
assert_eq!(
request.url().path(),
"/v1/table/my_table/generated_columns/create/"
);
let request_id = request
.headers()
.get("x-request-id")
.expect("client must set x-request-id")
.to_str()
.unwrap()
.to_string();
*expected_request_id_c.lock().unwrap() = Some(request_id);
http::Response::builder()
.status(200)
.body(body_for_handler.clone())
.unwrap()
});
match table
.submit_create_generated_column(1, sample_create_generated_column_job_spec())
.await
{
Err(err) => match &err {
Error::Http {
request_id,
status_code,
..
} => {
let expected = expected_request_id
.lock()
.unwrap()
.clone()
.expect("handler must observe request id");
assert_eq!(
request_id, &expected,
"{label}: Error::Http.request_id must equal outgoing x-request-id"
);
assert!(
!request_id.is_empty(),
"{label}: Error::Http.request_id must be nonempty"
);
assert!(
status_code.is_none(),
"{label}: local protocol validation must not invent a status code"
);
assert_create_gen_col_markers_absent(&err);
let text = create_gen_col_error_chain_text(&err);
assert!(
!text.contains(&response_body),
"{label}: raw response body must be absent from error/debug/source chain: {text}"
);
}
other => unexpected.push(format!("{label}: {other:?}")),
},
Ok(ok) => unexpected.push(format!("{label}: Ok({ok:?})")),
}
}
assert!(
unexpected.is_empty(),
"invalid job_id shapes must fail closed as Error::Http: {unexpected:?}"
);
let echoed = format!(
"create failed with {CREATE_GEN_COL_LITERAL_MARKER} and {CREATE_GEN_COL_RESPONSE_MARKER}"
);
{
let body = echoed.clone();
let table = Table::new_with_handler("my_table", move |_| {
http::Response::builder()
.status(400)
.body(body.clone())
.unwrap()
});
let err = table
.submit_create_generated_column(1, sample_create_generated_column_job_spec())
.await
.expect_err("4xx submit must fail");
assert!(matches!(err, Error::Http { .. }), "got {err:?}");
assert_create_gen_col_markers_absent(&err);
}
{
let body = echoed.clone();
let attempts = Arc::new(AtomicUsize::new(0));
let attempts_ref = attempts.clone();
let table = Table::new_with_handler_and_config(
"my_table",
move |_| {
attempts_ref.fetch_add(1, Ordering::SeqCst);
http::Response::builder()
.status(500)
.body(body.clone())
.unwrap()
},
ClientConfig {
retry_config: RetryConfig {
retries: Some(2),
backoff_factor: Some(0.0),
backoff_jitter: Some(0.0),
..Default::default()
},
..Default::default()
},
);
let err = table
.submit_create_generated_column(1, sample_create_generated_column_job_spec())
.await
.expect_err("exhausted 5xx submit must fail");
assert!(matches!(err, Error::Retry { .. }), "got {err:?}");
assert_eq!(attempts.load(Ordering::SeqCst), 2);
assert_create_gen_col_markers_absent(&err);
}
}
/// Successful submit alone does not invalidate schema cache or advance freshness.
///
/// Write watermark (9), read watermark (7), and submit `source_table_version`
/// (42) are three distinct values so an erroneous
/// `track_write_version(source_table_version)` /
/// `track_read_version(source_table_version)` cannot hide behind a no-op
/// max(same, same).
#[tokio::test]
async fn test_submit_create_generated_column_preserves_schema_cache_and_freshness() {
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
let seeded = describe_response(&schema);
let seeded_for_handler = seeded.clone();
let describe_calls = Arc::new(AtomicUsize::new(0));
let describe_calls_ref = describe_calls.clone();
let count_calls = Arc::new(AtomicUsize::new(0));
let count_calls_ref = count_calls.clone();
let post_submit_headers = Arc::new(std::sync::Mutex::new(None::<http::HeaderMap>));
let post_submit_headers_ref = post_submit_headers.clone();
let remote = RemoteTable::new_mock(
"my_table".into(),
move |request| match request.url().path() {
"/v1/table/my_table/update/" => http::Response::builder()
.status(200)
.body(r#"{"rows_updated":1,"version":9}"#.to_string())
.unwrap(),
"/v1/table/my_table/generated_columns/create/" => {
let body = request_body_json(&request);
assert_eq!(
body["source_table_version"], 42,
"submit must carry the distinct higher source_table_version"
);
http::Response::builder()
.status(200)
.body(r#"{"job_id":"job-create-gen-col-cache"}"#.to_string())
.unwrap()
}
"/v1/table/my_table/describe/" => {
describe_calls_ref.fetch_add(1, Ordering::SeqCst);
http::Response::builder()
.status(200)
.body(seeded_for_handler.clone())
.unwrap()
}
"/v1/table/my_table/count_rows/" => {
let n = count_calls_ref.fetch_add(1, Ordering::SeqCst);
if n == 0 {
// Real read response establishes read watermark 7.
http::Response::builder()
.status(200)
.header("x-lancedb-version", "7")
.body("1".to_string())
.unwrap()
} else {
*post_submit_headers_ref.lock().unwrap() = Some(request.headers().clone());
http::Response::builder()
.status(200)
.header("x-lancedb-version", "7")
.body("1".to_string())
.unwrap()
}
}
path => panic!("unexpected path: {path}"),
},
None,
);
remote.seed_schema(&seeded);
let table = Table::from(Arc::new(remote) as Arc<dyn BaseTable>);
// Distinct prior write watermark (9) via a real write response.
table.update().column("a", "a + 1").execute().await.unwrap();
// Distinct prior read watermark (7) via a real read response.
table.count_rows(None).await.unwrap();
assert_eq!(
count_calls.load(Ordering::SeqCst),
1,
"pre-submit read must establish the read watermark"
);
let schema_before = table.schema().await.unwrap();
assert_eq!(
describe_calls.load(Ordering::SeqCst),
0,
"seeded schema must satisfy schema() before submit"
);
let job = table
.submit_create_generated_column(42, sample_create_generated_column_job_spec())
.await
.expect("submit must succeed");
assert_eq!(job.id(), Some("job-create-gen-col-cache"));
let schema_after = table.schema().await.unwrap();
assert_eq!(
Arc::as_ptr(&schema_before),
Arc::as_ptr(&schema_after),
"submit must not invalidate the schema cache"
);
assert_eq!(
describe_calls.load(Ordering::SeqCst),
0,
"submit must not force a describe/schema refetch"
);
table.count_rows(None).await.unwrap();
assert_eq!(
count_calls.load(Ordering::SeqCst),
2,
"post-submit read must observe freshness headers"
);
let headers = post_submit_headers
.lock()
.unwrap()
.clone()
.expect("post-submit count_rows headers");
let write_wm = headers
.get("x-lancedb-min-version")
.expect("write watermark must remain")
.to_str()
.unwrap();
let read_wm = headers
.get("x-lancedb-min-read-version")
.expect("read watermark must remain")
.to_str()
.unwrap();
assert_eq!(
write_wm, "9",
"submit must not clear or advance write freshness beyond prior writes"
);
assert_eq!(
read_wm, "7",
"submit must not clear or advance read freshness beyond prior reads"
);
assert_ne!(
write_wm, "42",
"write watermark must not become the submit source_table_version"
);
assert_ne!(
read_wm, "42",
"read watermark must not become the submit source_table_version"
);
}
/// Test that schema cache expires after 30 seconds TTL
#[tokio::test]
async fn test_schema_cache_invalidation_after_ttl() {
rust/lancedb/src/table.rs
+35 -1
View File
@@ -54,7 +54,7 @@ use crate::database::listing::LANCE_FILE_EXTENSION;
use crate::database::read_freshness::TableFreshness;
use crate::embeddings::{EmbeddingDefinition, EmbeddingRegistry, MemoryRegistry};
use crate::error::{Error, Result};
use crate::function::GeneratedColumnBindingSnapshot;
use crate::function::{CreateGeneratedColumnJobSpec, GeneratedColumnBindingSnapshot};
use crate::index::IndexStatistics;
use crate::index::{Index, IndexBuilder};
use crate::index::{IndexConfig, IndexStatisticsImpl, IndexType};
@@ -616,6 +616,23 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
message: "generated_column_binding_snapshot is not supported on this table type".into(),
})
}
/// Submit a create-generated-column Job from an already-bound snapshot pair.
///
/// Consumes `(source_table_version, CreateGeneratedColumnJobSpec)` produced
/// by one binding snapshot. Default returns [`Error::NotSupported`] so
/// existing third-party [`BaseTable`] implementations keep compiling.
/// Implementations must not re-read schema/version, rebind the call,
/// catalog-lookup, execute, stage, or mutate locally.
#[doc(hidden)]
async fn submit_create_generated_column(
&self,
_source_table_version: u64,
_spec: CreateGeneratedColumnJobSpec,
) -> Result<Job> {
Err(Error::NotSupported {
message: "submit_create_generated_column is not supported on this table type".into(),
})
}
/// Count the number of rows in this table.
async fn count_rows(&self, filter: Option<Filter>) -> Result<usize>;
/// Create a physical plan for the query.
@@ -1139,6 +1156,23 @@ impl Table {
self.inner.generated_column_binding_snapshot().await
}
/// Submit a create-generated-column Job from an already-bound snapshot pair.
///
/// Hidden table submit seam for enterprise generated-column create. Input
/// must already be the `(source_table_version, CreateGeneratedColumnJobSpec)`
/// pairing from one binding snapshot. Returns the unified [`Job`]; does not
/// expose a public request/envelope type.
#[doc(hidden)]
pub async fn submit_create_generated_column(
&self,
source_table_version: u64,
spec: CreateGeneratedColumnJobSpec,
) -> Result<Job> {
self.inner
.submit_create_generated_column(source_table_version, spec)
.await
}
/// Count the number of rows in this dataset.
///
/// # Arguments
rust/lancedb/tests/first_class_generated_column_create_submit_contract.rs
+122
View File
@@ -0,0 +1,122 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! Contract tests for the hidden create-generated-column table submit seam
//! (FF-031).
//!
//! Scope is the Native/default `BaseTable`/`Table` submit path only: already-
//! bound `(source_table_version, CreateGeneratedColumnJobSpec)` must return
//! [`lancedb::Error::NotSupported`] without mutating dataset version or
//! schema. Remote transport and Job result decoding live in crate unit tests.
use std::sync::Arc;
use arrow_array::{Int32Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use lancedb::Error;
use lancedb::connect;
use lancedb::function::{
CreateGeneratedColumnJobSpec, Function, FunctionArgument, FunctionCall, FunctionId,
FunctionOutput, FunctionParameter, FunctionSignature,
};
fn sample_function() -> Function {
let id = FunctionId::try_new("fn.exact.create-submit.native").expect("valid FunctionId");
let signature = FunctionSignature::try_new(
vec![
FunctionParameter::new("x", DataType::Int32),
FunctionParameter::new("label", DataType::Utf8),
],
FunctionOutput::new(DataType::Int32, true),
)
.expect("valid FunctionSignature");
Function::new(id, signature)
}
fn sample_spec() -> CreateGeneratedColumnJobSpec {
let function = sample_function();
let call = FunctionCall::try_new(
&function,
vec![
(
"x".to_string(),
FunctionArgument::try_field(1, DataType::Int32).expect("field arg"),
),
(
"label".to_string(),
FunctionArgument::try_literal(Arc::new(StringArray::from(vec![Some(
"SENSITIVE_CREATE_GEN_COL_LITERAL_MARKER",
)])) as _)
.expect("literal arg"),
),
],
)
.expect("valid FunctionCall");
CreateGeneratedColumnJobSpec::try_new("normalized", &function, call).expect("valid spec")
}
/// Native tables reject create-generated-column submit without dataset change.
#[tokio::test]
async fn native_submit_create_generated_column_returns_not_supported_without_mutation() {
let dir = tempfile::tempdir().expect("tempdir");
let conn = connect(dir.path().to_str().unwrap())
.execute()
.await
.expect("local connect");
let schema = Arc::new(Schema::new(vec![
Field::new("x", DataType::Int32, false),
Field::new("label", DataType::Utf8, true),
]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int32Array::from(vec![1])),
Arc::new(StringArray::from(vec![Some("a")])),
],
)
.expect("batch");
let table = conn
.create_table("create_submit_native", batch)
.execute()
.await
.expect("create table");
let version_before = table.version().await.expect("version before");
let schema_before = table.schema().await.expect("schema before");
let field_names_before: Vec<_> = schema_before
.fields()
.iter()
.map(|f| f.name().clone())
.collect();
let err = table
.submit_create_generated_column(version_before, sample_spec())
.await
.expect_err("native submit must be unsupported");
assert!(
matches!(err, Error::NotSupported { .. }),
"expected NotSupported, got {err:?}"
);
let version_after = table.version().await.expect("version after");
let schema_after = table.schema().await.expect("schema after");
let field_names_after: Vec<_> = schema_after
.fields()
.iter()
.map(|f| f.name().clone())
.collect();
assert_eq!(
version_before, version_after,
"unsupported submit must not advance dataset version"
);
assert_eq!(
field_names_before, field_names_after,
"unsupported submit must not change schema field names"
);
assert_eq!(
schema_before.fields().len(),
schema_after.fields().len(),
"unsupported submit must not add schema fields"
);
}