mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-01 02:58:36 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5cbd979455 | |||
| e773d1e093 |
Generated
+26
-24
@@ -535,9 +535,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "async-trait"
|
||||
version = "0.1.91"
|
||||
version = "0.1.92"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec"
|
||||
checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -1443,9 +1443,9 @@ checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e"
|
||||
|
||||
[[package]]
|
||||
name = "bytemuck"
|
||||
version = "1.25.0"
|
||||
version = "1.25.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec"
|
||||
checksum = "95832e849adfb21180ccb6826a99da14e5d266ae5c2e668e1602cf234f153797"
|
||||
dependencies = [
|
||||
"bytemuck_derive",
|
||||
]
|
||||
@@ -5748,9 +5748,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "log"
|
||||
version = "0.4.33"
|
||||
version = "0.4.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
|
||||
checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
|
||||
|
||||
[[package]]
|
||||
name = "loom"
|
||||
@@ -6001,9 +6001,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "moka"
|
||||
version = "0.12.15"
|
||||
version = "0.12.16"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "957228ad12042ee839f93c8f257b62b4c0ab5eaae1d4fa60de53b27c9d7c5046"
|
||||
checksum = "4293f18e7567a1caf3c584855554377025c65e0aa445344d04171f5ad63d19b9"
|
||||
dependencies = [
|
||||
"async-lock",
|
||||
"crossbeam-channel",
|
||||
@@ -6097,14 +6097,15 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "napi"
|
||||
version = "3.11.0"
|
||||
version = "3.12.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "de33522036981030a75c231829566bc63414e08101a6f5ff4ac6cef19c8e0941"
|
||||
checksum = "58c5f4d5375213fdb7be2655e152386e82f026f9a5ba36a75556e11359aafe09"
|
||||
dependencies = [
|
||||
"bitflags 2.11.1",
|
||||
"chrono",
|
||||
"ctor 1.0.12",
|
||||
"futures",
|
||||
"libc",
|
||||
"napi-build",
|
||||
"napi-sys",
|
||||
"nohash-hasher",
|
||||
@@ -6116,15 +6117,15 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "napi-build"
|
||||
version = "2.4.0"
|
||||
version = "2.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5282704fbe8d49b0cf8b08e3f33233416a528658f205c7e5ace63b582de0b11c"
|
||||
checksum = "60fdf9b392c50e7c4170fa633bd909490ed7835cea4c046776d1a4dd8d2ae0ab"
|
||||
|
||||
[[package]]
|
||||
name = "napi-derive"
|
||||
version = "3.6.1"
|
||||
version = "3.6.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4d5c9c02556ea6dc99dffd36c1ce60141411657438501a125b675776d011ce92"
|
||||
checksum = "0fa55ea69990c90b888e9e77044410e304ce7f35de599dc6d0b5c1923d2e59af"
|
||||
dependencies = [
|
||||
"convert_case",
|
||||
"ctor 1.0.12",
|
||||
@@ -6136,9 +6137,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "napi-derive-backend"
|
||||
version = "6.1.1"
|
||||
version = "6.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d60b5d773ad46c698c8cc2cd9fde0b283d39cbb7f71c04bee633c7bdba4423bd"
|
||||
checksum = "df4056ac7c18e4438ccf0edaed4340ca0d269278c8ec19284f7b23cb039fd0ae"
|
||||
dependencies = [
|
||||
"convert_case",
|
||||
"proc-macro2",
|
||||
@@ -8601,9 +8602,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "roaring"
|
||||
version = "0.11.4"
|
||||
version = "0.11.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1dedc5658c6ecb3bdb5ef5f3295bb9253f42dcf3fd1402c03f6b1f7659c3c4a9"
|
||||
checksum = "18bd8a37d17a58532776dcdf6041ce64929adca78e8489d5cacbafe99229d3e1"
|
||||
dependencies = [
|
||||
"bytemuck",
|
||||
"byteorder",
|
||||
@@ -9063,9 +9064,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "serde_with"
|
||||
version = "3.21.0"
|
||||
version = "3.22.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c"
|
||||
checksum = "ee78f1fbe43ac4a0e47aadb3dbd357b69eb0d3793e948624cd03dd2750ab1c0a"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"bs58",
|
||||
@@ -9073,6 +9074,7 @@ dependencies = [
|
||||
"hex",
|
||||
"indexmap 1.9.3",
|
||||
"indexmap 2.14.0",
|
||||
"jiff",
|
||||
"schemars 0.9.0",
|
||||
"schemars 1.2.1",
|
||||
"serde_core",
|
||||
@@ -9083,9 +9085,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "serde_with_macros"
|
||||
version = "3.21.0"
|
||||
version = "3.22.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "84d57bc0c8b9a17920c178daa6bb924850d54a9c97ab45194bb8c17ad66bb660"
|
||||
checksum = "8705578779c2b6bd90d84d66eb2e206b708b1a4d7b9f17641b293545bf1c7e46"
|
||||
dependencies = [
|
||||
"darling 0.23.0",
|
||||
"proc-macro2",
|
||||
@@ -10452,9 +10454,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
|
||||
|
||||
[[package]]
|
||||
name = "uuid"
|
||||
version = "1.24.0"
|
||||
version = "1.26.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
|
||||
checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
|
||||
dependencies = [
|
||||
"getrandom 0.4.2",
|
||||
"js-sys",
|
||||
|
||||
@@ -50,16 +50,6 @@ projections: [string, string][];
|
||||
|
||||
***
|
||||
|
||||
### sourceNamespace
|
||||
|
||||
```ts
|
||||
sourceNamespace: string[];
|
||||
```
|
||||
|
||||
Namespace holding the source table; empty is the root namespace.
|
||||
|
||||
***
|
||||
|
||||
### sourceTable
|
||||
|
||||
```ts
|
||||
|
||||
@@ -48,28 +48,6 @@ describe("materialized views", () => {
|
||||
expect(definitionFromMetadata(safe, "v").limit).toBe(42);
|
||||
});
|
||||
|
||||
it("reads the namespaced select kind and refuses unknown kinds", () => {
|
||||
// "namespaced_select" is the namespaced form of "select": same shape, a
|
||||
// separate kind so readers that predate it refuse instead of resolving
|
||||
// the source at the root.
|
||||
const namespaced = new Map([
|
||||
[
|
||||
DEFINITION_META_KEY,
|
||||
'{"kind":"namespaced_select","source_table":"people","source_namespace":["ns"]}',
|
||||
],
|
||||
]);
|
||||
const definition = definitionFromMetadata(namespaced, "v");
|
||||
expect(definition.sourceTable).toBe("people");
|
||||
expect(definition.sourceNamespace).toEqual(["ns"]);
|
||||
|
||||
const unknown = new Map([
|
||||
[DEFINITION_META_KEY, '{"kind":"select_v3","source_table":"people"}'],
|
||||
]);
|
||||
expect(() => definitionFromMetadata(unknown, "v")).toThrow(
|
||||
/cannot refresh/,
|
||||
);
|
||||
});
|
||||
|
||||
it("creates, refreshes and queries a view", async () => {
|
||||
const view = await db.createMaterializedView("adults", "people", {
|
||||
select: ["name", ["shout", "upper(name)"]],
|
||||
|
||||
@@ -19,8 +19,6 @@ export interface MaterializedViewDefinition {
|
||||
limit?: number;
|
||||
/** Source columns the projections and filter read. */
|
||||
inputs: string[];
|
||||
/** Namespace holding the source table; empty is the root namespace. */
|
||||
sourceNamespace: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -80,8 +78,7 @@ export function definitionFromMetadata(
|
||||
}
|
||||
// biome-ignore lint/suspicious/noExplicitAny: raw JSON
|
||||
const value: any = JSON.parse(raw);
|
||||
// "namespaced_select" keeps older readers from resolving the source at root.
|
||||
if (value.kind !== "select" && value.kind !== "namespaced_select") {
|
||||
if (value.kind !== "select") {
|
||||
throw new Error(
|
||||
`materialized view '${name}' is defined by '${value.kind}', which this ` +
|
||||
"version of lancedb cannot refresh",
|
||||
@@ -106,7 +103,6 @@ export function definitionFromMetadata(
|
||||
filter: value.filter ?? undefined,
|
||||
limit,
|
||||
inputs: value.inputs ?? [],
|
||||
sourceNamespace: value.source_namespace ?? [],
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
+5
-1
@@ -664,7 +664,11 @@ impl JsFullTextQuery {
|
||||
}
|
||||
|
||||
fn parse_fts_query(query: Object) -> napi::Result<FullTextSearchQuery> {
|
||||
if let Ok(Some(query)) = query.get::<&JsFullTextQuery>("query") {
|
||||
// `&JsFullTextQuery` recovers a native class reference through napi's borrow-tracked
|
||||
// path, which is only usable from generated `#[napi]` argument conversion. This is a
|
||||
// manual lookup on a nested `Object` property instead, so use `ClassInstance`, which
|
||||
// unwraps the class without requiring a borrow scope.
|
||||
if let Ok(Some(query)) = query.get::<ClassInstance<JsFullTextQuery>>("query") {
|
||||
Ok(FullTextSearchQuery::new_query(query.inner.clone()))
|
||||
} else if let Ok(Some(query_text)) = query.get::<String>("query") {
|
||||
let mut query_text = query_text;
|
||||
|
||||
@@ -42,8 +42,6 @@ class MaterializedViewDefinition:
|
||||
"""Cap on the number of rows the view holds."""
|
||||
inputs: List[str] = field(default_factory=list)
|
||||
"""Source columns the projections and filter read."""
|
||||
source_namespace: List[str] = field(default_factory=list)
|
||||
"""Namespace holding the source table; empty is the root namespace."""
|
||||
|
||||
|
||||
def _definition_from_schema(
|
||||
@@ -55,8 +53,7 @@ def _definition_from_schema(
|
||||
raise ValueError(f"Table '{name}' is not a materialized view")
|
||||
value = json.loads(raw)
|
||||
kind = value.get("kind")
|
||||
# "namespaced_select" keeps older readers from resolving the source at root.
|
||||
if kind not in ("select", "namespaced_select"):
|
||||
if kind != "select":
|
||||
raise NotImplementedError(
|
||||
f"materialized view '{name}' is defined by '{kind}', which this "
|
||||
"version of lancedb cannot refresh"
|
||||
@@ -69,7 +66,6 @@ def _definition_from_schema(
|
||||
filter=value.get("filter"),
|
||||
limit=value.get("limit"),
|
||||
inputs=value.get("inputs", []),
|
||||
source_namespace=value.get("source_namespace", []),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -266,38 +266,3 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
|
||||
handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust
|
||||
)
|
||||
assert handle._namespace_path == through_namespace._namespace_path
|
||||
|
||||
|
||||
def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
|
||||
import json
|
||||
|
||||
import pyarrow as pa
|
||||
|
||||
from lancedb.materialized_view import _definition_from_schema
|
||||
|
||||
def schema_with(definition: dict) -> pa.Schema:
|
||||
return pa.schema([pa.field("id", pa.int32())]).with_metadata(
|
||||
{b"mv.definition": json.dumps(definition).encode()}
|
||||
)
|
||||
|
||||
# "namespaced_select" is the namespaced form of "select": same shape,
|
||||
# a separate kind so readers that predate it refuse instead of
|
||||
# resolving the source at the root.
|
||||
definition = _definition_from_schema(
|
||||
schema_with(
|
||||
{
|
||||
"kind": "namespaced_select",
|
||||
"source_table": "people",
|
||||
"source_namespace": ["ns"],
|
||||
"projections": [{"output": "name", "expression": "name"}],
|
||||
}
|
||||
),
|
||||
"v",
|
||||
)
|
||||
assert definition.source_table == "people"
|
||||
assert definition.source_namespace == ["ns"]
|
||||
|
||||
with pytest.raises(NotImplementedError, match="cannot refresh"):
|
||||
_definition_from_schema(
|
||||
schema_with({"kind": "select_v3", "source_table": "people"}), "v"
|
||||
)
|
||||
|
||||
@@ -539,9 +539,7 @@ impl Database for LanceNamespaceDatabase {
|
||||
self.namespace
|
||||
.drop_table(drop_request)
|
||||
.await
|
||||
.map_err(|e| Error::Runtime {
|
||||
message: format!("Failed to drop table: {}", e),
|
||||
})?;
|
||||
.map_err(|e| map_namespace_lance_error(e, name))?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -1495,6 +1493,15 @@ mod tests {
|
||||
.expect("Failed to list tables");
|
||||
assert!(!table_names_after.contains(&"drop_test".to_string()));
|
||||
|
||||
let error = conn
|
||||
.drop_table("drop_test", &["test_ns".into()])
|
||||
.await
|
||||
.expect_err("dropping a missing table should fail");
|
||||
assert!(
|
||||
matches!(error, Error::TableNotFound { ref name, .. } if name == "drop_test"),
|
||||
"expected TableNotFound, got: {error:?}"
|
||||
);
|
||||
|
||||
// Verify: Cannot open dropped table
|
||||
let open_result = conn.open_table("drop_test").execute().await;
|
||||
assert!(open_result.is_err());
|
||||
|
||||
@@ -74,15 +74,8 @@ const EMBEDDING_FUNCTIONS_META_KEY: &str = "embedding_functions";
|
||||
const COLUMN_DEFINITIONS_META_KEY: &str = "lancedb::column_definitions";
|
||||
|
||||
/// Value of the definition's `kind` tag for the projected `select` form.
|
||||
/// Reserved for root-namespace sources; see [`NAMESPACED_SELECT_KIND`].
|
||||
pub const SELECT_KIND: &str = "select";
|
||||
|
||||
/// The `select` form over a namespaced source: its own kind, because released
|
||||
/// readers drop unknown fields and resolve a `select` source at the root, so
|
||||
/// this routes them to the [`MaterializedViewKind::Unrecognized`] refusal
|
||||
/// instead of a wrong-table refresh.
|
||||
pub const NAMESPACED_SELECT_KIND: &str = "namespaced_select";
|
||||
|
||||
/// Which view outputs each source column is projected to directly. A column
|
||||
/// may be projected more than once, so each carries every name the view gives
|
||||
/// it, in projection order.
|
||||
@@ -102,10 +95,6 @@ pub struct ViewProjection {
|
||||
pub struct MaterializedViewDefinition {
|
||||
/// Name of the source table, in the same database as the view.
|
||||
pub source_table: String,
|
||||
/// Namespace path holding the source table; empty is the root namespace.
|
||||
/// A definition written before namespaced sources reads as root.
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub source_namespace: Vec<String>,
|
||||
/// The projected output columns, in view schema order.
|
||||
pub projections: Vec<ViewProjection>,
|
||||
/// SQL predicate selecting the source rows the view holds.
|
||||
@@ -140,12 +129,7 @@ pub(crate) fn definition_to_metadata(definition: &MaterializedViewDefinition) ->
|
||||
let mut value = serde_json::to_value(definition).map_err(|e| Error::Runtime {
|
||||
message: format!("failed to serialize view definition: {e}"),
|
||||
})?;
|
||||
let kind = if definition.source_namespace.is_empty() {
|
||||
SELECT_KIND
|
||||
} else {
|
||||
NAMESPACED_SELECT_KIND
|
||||
};
|
||||
value["kind"] = serde_json::Value::String(kind.to_string());
|
||||
value["kind"] = serde_json::Value::String(SELECT_KIND.to_string());
|
||||
Ok(value.to_string())
|
||||
}
|
||||
|
||||
@@ -166,21 +150,12 @@ pub fn materialized_view_kind(
|
||||
.get("kind")
|
||||
.and_then(|k| k.as_str())
|
||||
.ok_or_else(|| unreadable(&"missing kind tag"))?;
|
||||
if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND {
|
||||
if kind != SELECT_KIND {
|
||||
return Ok(Some(MaterializedViewKind::Unrecognized {
|
||||
kind: kind.to_string(),
|
||||
}));
|
||||
}
|
||||
let kind = kind.to_string();
|
||||
let definition: MaterializedViewDefinition =
|
||||
serde_json::from_value(value).map_err(|e| unreadable(&e))?;
|
||||
// No correct writer produces a kind that disagrees with its namespace.
|
||||
if (kind == SELECT_KIND) != definition.source_namespace.is_empty() {
|
||||
return Err(unreadable(&format!(
|
||||
"kind '{kind}' does not match its source namespace {:?}",
|
||||
definition.source_namespace
|
||||
)));
|
||||
}
|
||||
let definition = serde_json::from_value(value).map_err(|e| unreadable(&e))?;
|
||||
Ok(Some(MaterializedViewKind::Select(definition)))
|
||||
}
|
||||
|
||||
@@ -191,7 +166,6 @@ pub fn materialized_view_kind(
|
||||
pub(crate) fn plan(
|
||||
source_schema: SchemaRef,
|
||||
source_table: &str,
|
||||
source_namespace: &[String],
|
||||
projections: &[(String, String)],
|
||||
filter: Option<&str>,
|
||||
limit: Option<u64>,
|
||||
@@ -345,7 +319,6 @@ pub(crate) fn plan(
|
||||
|
||||
let definition = MaterializedViewDefinition {
|
||||
source_table: source_table.to_string(),
|
||||
source_namespace: source_namespace.to_vec(),
|
||||
projections: projections
|
||||
.into_iter()
|
||||
.map(|(output, expression)| ViewProjection { output, expression })
|
||||
@@ -629,7 +602,7 @@ pub struct PreparedDeclaration {
|
||||
definition: MaterializedViewDefinition,
|
||||
/// The source's own database: the only place
|
||||
/// [`PreparedDeclaration::create`] will put the view, because refresh
|
||||
/// resolves the recorded source coordinate through the view's database.
|
||||
/// resolves the recorded source name through the view's database.
|
||||
database: Arc<dyn Database>,
|
||||
}
|
||||
|
||||
@@ -649,21 +622,10 @@ impl PreparedDeclaration {
|
||||
|
||||
/// Create the view table and verify it, consuming the declaration.
|
||||
///
|
||||
/// The view goes at the root of the source's own database, where refresh
|
||||
/// resolves the recorded source coordinate. Stable row ids are requested
|
||||
/// at both levels and verified rather than trusted; nothing is rolled
|
||||
/// back on failure.
|
||||
/// The view goes in the source's own database, where refresh resolves the
|
||||
/// recorded source name. Stable row ids are requested at both levels and
|
||||
/// verified rather than trusted; nothing is rolled back on failure.
|
||||
pub async fn create(self, name: &str) -> Result<MaterializedView> {
|
||||
self.create_in(&[], name).await
|
||||
}
|
||||
|
||||
/// Create the view in `namespace_path`, empty for the root namespace.
|
||||
/// Otherwise [`PreparedDeclaration::create`].
|
||||
pub async fn create_in(
|
||||
self,
|
||||
namespace_path: &[String],
|
||||
name: &str,
|
||||
) -> Result<MaterializedView> {
|
||||
let empty: Vec<std::result::Result<arrow_array::RecordBatch, arrow_schema::ArrowError>> =
|
||||
vec![];
|
||||
// Minted here, not at preparation: a declaration can be cloned and
|
||||
@@ -678,7 +640,6 @@ impl PreparedDeclaration {
|
||||
let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
|
||||
Box::new(arrow_array::RecordBatchIterator::new(empty, schema));
|
||||
let mut request = CreateTableRequest::new(name.to_string(), Box::new(reader));
|
||||
request.namespace_path = namespace_path.to_vec();
|
||||
let write_params = request
|
||||
.write_options
|
||||
.lance_write_params
|
||||
@@ -719,8 +680,8 @@ impl PreparedDeclaration {
|
||||
|
||||
/// Validate a view declaration against its live source and hold what its
|
||||
/// creation needs. The declaration is canonicalized through the coordinate a
|
||||
/// refresh will resolve -- name and namespace both -- so a handle that does
|
||||
/// not resolve back to itself is rejected. Same creation-time checks as
|
||||
/// refresh will resolve, so a handle that does not resolve back to itself is
|
||||
/// rejected, as is a namespaced source. Same creation-time checks as
|
||||
/// [`Connection::create_materialized_view`].
|
||||
///
|
||||
/// ```no_run
|
||||
@@ -749,9 +710,17 @@ pub async fn prepare_declaration(
|
||||
message: "materialized views are supported only on local databases".into(),
|
||||
});
|
||||
};
|
||||
// Refresh resolves the source at exactly this coordinate, so the
|
||||
// definition records the namespace alongside the name.
|
||||
let source_namespace = source.namespace().to_vec();
|
||||
// The definition records the source by bare name; any other source
|
||||
// form would be recorded as a name its refresh cannot resolve.
|
||||
if !source.namespace().is_empty() {
|
||||
return Err(Error::NotSupported {
|
||||
message: format!(
|
||||
"a namespaced source cannot be recorded in a view definition; \
|
||||
'{}' must be a root-namespace table",
|
||||
source.name()
|
||||
),
|
||||
});
|
||||
}
|
||||
let database = source
|
||||
.database_opt()
|
||||
.ok_or_else(|| Error::InvalidInput {
|
||||
@@ -765,7 +734,7 @@ pub async fn prepare_declaration(
|
||||
let resolved = database
|
||||
.open_table(OpenTableRequest {
|
||||
name: source.name().to_string(),
|
||||
namespace_path: source_namespace.clone(),
|
||||
namespace_path: vec![],
|
||||
index_cache_size: None,
|
||||
lance_read_params: None,
|
||||
location: None,
|
||||
@@ -811,7 +780,6 @@ pub async fn prepare_declaration(
|
||||
let (definition, mut fields, lineage) = plan(
|
||||
source_schema.clone(),
|
||||
resolved.name(),
|
||||
&source_namespace,
|
||||
projections,
|
||||
filter,
|
||||
limit,
|
||||
@@ -871,9 +839,7 @@ fn ensure_local(connection: &Connection) -> Result<()> {
|
||||
pub struct CreateMaterializedViewBuilder {
|
||||
connection: Connection,
|
||||
name: String,
|
||||
namespace: Vec<String>,
|
||||
source: String,
|
||||
source_namespace: Vec<String>,
|
||||
projections: Vec<(String, String)>,
|
||||
filter: Option<String>,
|
||||
limit: Option<u64>,
|
||||
@@ -884,28 +850,13 @@ impl CreateMaterializedViewBuilder {
|
||||
Self {
|
||||
connection,
|
||||
name,
|
||||
namespace: Vec::new(),
|
||||
source,
|
||||
source_namespace: Vec::new(),
|
||||
projections: Vec::new(),
|
||||
filter: None,
|
||||
limit: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// The namespace to create the view in. Defaults to the root namespace.
|
||||
pub fn namespace(mut self, namespace_path: Vec<String>) -> Self {
|
||||
self.namespace = namespace_path;
|
||||
self
|
||||
}
|
||||
|
||||
/// The namespace holding the source table; recorded in the definition
|
||||
/// for refresh to resolve. Defaults to the root namespace.
|
||||
pub fn source_namespace(mut self, namespace_path: Vec<String>) -> Self {
|
||||
self.source_namespace = namespace_path;
|
||||
self
|
||||
}
|
||||
|
||||
/// The view's columns, as `(name, SQL expression)` pairs. Not calling
|
||||
/// this selects every source column, expanded at creation time.
|
||||
pub fn select(
|
||||
@@ -936,12 +887,7 @@ impl CreateMaterializedViewBuilder {
|
||||
/// provenance across compaction, and cannot be enabled later.
|
||||
pub async fn execute(self) -> Result<MaterializedView> {
|
||||
ensure_local(&self.connection)?;
|
||||
let source = self
|
||||
.connection
|
||||
.open_table(&self.source)
|
||||
.namespace(self.source_namespace.clone())
|
||||
.execute()
|
||||
.await?;
|
||||
let source = self.connection.open_table(&self.source).execute().await?;
|
||||
let prepared = prepare_declaration(
|
||||
&source,
|
||||
&self.projections,
|
||||
@@ -949,7 +895,7 @@ impl CreateMaterializedViewBuilder {
|
||||
self.limit,
|
||||
)
|
||||
.await?;
|
||||
prepared.create_in(&self.namespace, &self.name).await
|
||||
prepared.create(&self.name).await
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1206,7 +1152,6 @@ mod tests {
|
||||
view.definition(),
|
||||
&MaterializedViewDefinition {
|
||||
source_table: "people".into(),
|
||||
source_namespace: Vec::new(),
|
||||
projections: vec![
|
||||
ViewProjection {
|
||||
output: "name".into(),
|
||||
@@ -2138,138 +2083,33 @@ mod tests {
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(err.to_string().contains("custom_loc"), "{err}");
|
||||
}
|
||||
|
||||
/// A view declared over a namespaced source records that namespace, and
|
||||
/// refresh resolves the source through it -- the coordinate round-trips.
|
||||
#[tokio::test]
|
||||
async fn a_namespaced_source_round_trips_through_refresh() {
|
||||
use lance_namespace::models::CreateNamespaceRequest;
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let mut properties = std::collections::HashMap::new();
|
||||
properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
|
||||
let conn = crate::connect_namespace("dir", properties)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
conn.create_namespace(CreateNamespaceRequest {
|
||||
id: Some(vec!["ns".into()]),
|
||||
..Default::default()
|
||||
})
|
||||
// A namespaced source cannot be recorded in the definition: the
|
||||
// bare name refresh resolves would reach a different table or none.
|
||||
let namespaced = crate::table::NativeTable::create(
|
||||
"memory://ns_src",
|
||||
"ns_src",
|
||||
vec!["ns".to_string()],
|
||||
Box::new(arrow_array::RecordBatchIterator::new(
|
||||
vec![],
|
||||
std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
|
||||
"id",
|
||||
arrow_schema::DataType::Int32,
|
||||
true,
|
||||
)])),
|
||||
)) as Box<dyn arrow_array::RecordBatchReader + Send>,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
std::collections::HashSet::new(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let batch = record_batch!(
|
||||
("name", Utf8, ["ada", "grace", "alan"]),
|
||||
("age", Int32, [36, 85, 41])
|
||||
)
|
||||
.unwrap();
|
||||
conn.create_table("people", batch)
|
||||
.namespace(vec!["ns".to_string()])
|
||||
.write_options(stable_row_ids())
|
||||
.execute()
|
||||
let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone());
|
||||
let err = prepare_declaration(&namespaced, &[], None, None)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// A decoy of the same name at the root: resolving the source at the
|
||||
// wrong namespace materializes one row here instead of three.
|
||||
let decoy = record_batch!(("name", Utf8, ["mallory"]), ("age", Int32, [42])).unwrap();
|
||||
conn.create_table("people", decoy)
|
||||
.write_options(stable_row_ids())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let view = conn
|
||||
.create_materialized_view("adults", "people")
|
||||
.namespace(vec!["ns".to_string()])
|
||||
.source_namespace(vec!["ns".to_string()])
|
||||
.select([("name", "name")])
|
||||
.only_if("age >= 18")
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(view.definition().source_table, "people");
|
||||
assert_eq!(view.definition().source_namespace, vec!["ns".to_string()]);
|
||||
assert_eq!(view.table().namespace(), &["ns"]);
|
||||
|
||||
// Refresh resolves the source at the recorded namespace, not at root.
|
||||
let result = view.refresh().execute().await.unwrap();
|
||||
assert_eq!(result.rows_written, 3);
|
||||
}
|
||||
|
||||
/// A definition stored before namespaced sources existed carries no
|
||||
/// namespace key and must read as the root namespace.
|
||||
#[test]
|
||||
fn a_definition_without_a_namespace_reads_as_root() {
|
||||
let stored =
|
||||
r#"{"source_table":"people","projections":[{"output":"name","expression":"name"}]}"#;
|
||||
let definition: MaterializedViewDefinition = serde_json::from_str(stored).unwrap();
|
||||
assert!(definition.source_namespace.is_empty());
|
||||
}
|
||||
|
||||
fn definition(source_namespace: Vec<String>) -> MaterializedViewDefinition {
|
||||
MaterializedViewDefinition {
|
||||
source_table: "people".to_string(),
|
||||
source_namespace,
|
||||
projections: vec![ViewProjection {
|
||||
output: "name".to_string(),
|
||||
expression: "name".to_string(),
|
||||
}],
|
||||
filter: None,
|
||||
limit: None,
|
||||
inputs: vec!["name".to_string()],
|
||||
}
|
||||
}
|
||||
|
||||
/// A root definition keeps the pre-namespace `select` form byte-stably;
|
||||
/// a namespaced one moves off `select`, which sends pre-namespace readers
|
||||
/// to the `Unrecognized` refusal instead of a root resolve.
|
||||
#[test]
|
||||
fn a_namespaced_definition_is_refused_by_the_pre_namespace_reader() {
|
||||
let root = definition_to_metadata(&definition(Vec::new())).unwrap();
|
||||
let root: serde_json::Value = serde_json::from_str(&root).unwrap();
|
||||
assert_eq!(root["kind"], "select");
|
||||
assert!(
|
||||
root.get("source_namespace").is_none(),
|
||||
"a root definition must not grow new keys: {root}"
|
||||
);
|
||||
|
||||
let stored = definition_to_metadata(&definition(vec!["ns".to_string()])).unwrap();
|
||||
let value: serde_json::Value = serde_json::from_str(&stored).unwrap();
|
||||
// The pre-namespace discriminator is `kind == "select"`; anything
|
||||
// else lands in its Unrecognized refusal rather than in a root open.
|
||||
assert_eq!(value["kind"], "namespaced_select");
|
||||
|
||||
// The current reader round-trips the coordinate.
|
||||
let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), stored)]);
|
||||
match materialized_view_kind(&metadata).unwrap() {
|
||||
Some(MaterializedViewKind::Select(read)) => {
|
||||
assert_eq!(read.source_namespace, vec!["ns".to_string()])
|
||||
}
|
||||
other => panic!("expected the namespaced select form, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// A kind that disagrees with its namespace is an error, not a view:
|
||||
/// under `select` it is the shape old readers would resolve at the root.
|
||||
#[test]
|
||||
fn a_kind_namespace_mismatch_is_refused() {
|
||||
for (kind, namespace) in [
|
||||
(SELECT_KIND, vec!["ns".to_string()]),
|
||||
(NAMESPACED_SELECT_KIND, Vec::new()),
|
||||
] {
|
||||
let mut value = serde_json::to_value(definition(namespace)).unwrap();
|
||||
value["kind"] = serde_json::Value::String(kind.to_string());
|
||||
let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), value.to_string())]);
|
||||
let err = materialized_view_kind(&metadata).unwrap_err();
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("does not match its source namespace"),
|
||||
"kind '{kind}': {err}"
|
||||
);
|
||||
}
|
||||
.unwrap_err();
|
||||
assert!(err.to_string().contains("namespaced source"), "{err}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -170,7 +170,6 @@ pub(crate) async fn execute_refresh(
|
||||
let (replanned, mut planned_fields, _renames) = super::plan(
|
||||
source_schema,
|
||||
&definition.source_table,
|
||||
&definition.source_namespace,
|
||||
&projections,
|
||||
definition.filter.as_deref(),
|
||||
definition.limit,
|
||||
@@ -591,7 +590,7 @@ async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> R
|
||||
let source = database
|
||||
.open_table(OpenTableRequest {
|
||||
name: definition.source_table.clone(),
|
||||
namespace_path: definition.source_namespace.clone(),
|
||||
namespace_path: Vec::new(),
|
||||
index_cache_size: None,
|
||||
lance_read_params: None,
|
||||
location: None,
|
||||
@@ -2920,7 +2919,6 @@ mod tests {
|
||||
|
||||
let replacement = crate::materialized_view::MaterializedViewDefinition {
|
||||
source_table: "src".into(),
|
||||
source_namespace: Vec::new(),
|
||||
projections: vec![
|
||||
crate::materialized_view::ViewProjection {
|
||||
output: "x".into(),
|
||||
@@ -2960,7 +2958,6 @@ mod tests {
|
||||
|
||||
let narrower = crate::materialized_view::MaterializedViewDefinition {
|
||||
source_table: "src".into(),
|
||||
source_namespace: Vec::new(),
|
||||
projections: vec![crate::materialized_view::ViewProjection {
|
||||
output: "x".into(),
|
||||
expression: "x".into(),
|
||||
|
||||
Reference in New Issue
Block a user