feat(nodejs): materialized view bindings

Exposes materialized views to TypeScript: createMaterializedView,
openMaterializedView and listMaterializedViews on Connection, and a
MaterializedView handle carrying the parsed definition and
refresh({full, sourceVersion}), which returns the typed refresh result.
select accepts column names, [alias, expression] pairs, or a record of the
same; the definition reads back off the stored schema, so a reopened handle
needs no side channel. Remote connections surface the core's not-supported
error up front.

The napi crate needed the same recursion-limit raise as the core crate: the
refresh future's type graph overflows the default trait-recursion depth.

nodejs/__test__/embedding.test.ts

@@ -487,4 +487,52 @@ describe("embedding functions", () => {
       expect(stringSchema3).toEqual(stringExpectedSchema);
     },
   );
+  test("parses one function writing several vector columns", async () => {
+    class MockEmbeddingFunction extends EmbeddingFunction<string> {
+      ndims() {
+        return 3;
+      }
+      embeddingDataType(): Float {
+        return new Float32();
+      }
+      async computeQueryEmbeddings(_data: string) {
+        return [1, 2, 3];
+      }
+      async computeSourceEmbeddings(data: string[]) {
+        return Array.from({ length: data.length }).fill([
+          1, 2, 3,
+        ]) as number[][];
+      }
+    }
+    const registry = getRegistry();
+    registry.register("multi_output_mock")(MockEmbeddingFunction);
+
+    // A materialized view can project one source vector column under two
+    // names, so a table's configuration names the same function twice.
+    const parsed = await registry.parseFunctions(
+      new Map([
+        [
+          "embedding_functions",
+          JSON.stringify([
+            {
+              name: "multi_output_mock",
+              sourceColumn: "text",
+              vectorColumn: "vector_a",
+              model: {},
+            },
+            {
+              name: "multi_output_mock",
+              sourceColumn: "text",
+              vectorColumn: "vector_b",
+              model: {},
+            },
+          ]),
+        ],
+      ]),
+    );
+
+    expect(
+      [...parsed.values()].map(({ vectorColumn }) => vectorColumn).sort(),
+    ).toEqual(["vector_a", "vector_b"]);
+  });
 });

nodejs/__test__/materialized_view.test.ts

@@ -0,0 +1,123 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import * as tmp from "tmp";
+
+import { Connection, connect } from "../lancedb";
+
+describe("materialized views", () => {
+  let tmpDir: tmp.DirResult;
+  let db: Connection;
+
+  beforeEach(async () => {
+    tmpDir = tmp.dirSync({ unsafeCleanup: true });
+    db = await connect(tmpDir.name);
+    await db.createTable(
+      "people",
+      [
+        { name: "ada", age: 36 },
+        { name: "kid", age: 7 },
+        { name: "grace", age: 85 },
+      ],
+      { storageOptions: { newTableEnableStableRowIds: "true" } },
+    );
+  });
+  afterEach(() => tmpDir.removeCallback());
+
+  it("creates, refreshes and queries a view", async () => {
+    const view = await db.createMaterializedView("adults", "people", {
+      select: ["name", ["shout", "upper(name)"]],
+      where: "age >= 18",
+    });
+    expect(view.name).toBe("adults");
+    expect(await view.table().countRows()).toBe(0);
+
+    const result = await view.refresh();
+    expect(result.mode).toBe("rebuild");
+    expect(Number(result.rowsWritten)).toBe(2);
+
+    const rows = await view.table().query().toArray();
+    expect(rows.map((r) => r.shout).sort()).toEqual(["ADA", "GRACE"]);
+  });
+
+  it("round-trips the definition", async () => {
+    await db.createMaterializedView("adults", "people", {
+      where: "age >= 18",
+    });
+    const view = await db.openMaterializedView("adults");
+    const definition = await view.definition();
+    expect(definition.sourceTable).toBe("people");
+    expect(definition.filter).toBe("age >= 18");
+    expect(definition.projections).toEqual([
+      ["name", "`name`"],
+      ["age", "`age`"],
+    ]);
+    expect(definition.inputs).toEqual(["age", "name"]);
+  });
+
+  it("refreshes incrementally after an append", async () => {
+    const view = await db.createMaterializedView("copy", "people");
+    await view.refresh();
+
+    const people = await db.openTable("people");
+    await people.add([{ name: "alan", age: 41 }]);
+    const result = await view.refresh();
+    expect(result.mode).toBe("incremental");
+    expect(Number(result.rowsWritten)).toBe(1);
+    expect(await view.table().countRows()).toBe(4);
+
+    expect((await view.refresh()).mode).toBe("no_op");
+  });
+
+  it("lists views and rejects non-views", async () => {
+    await db.createMaterializedView("adults", "people", {
+      where: "age >= 18",
+    });
+    expect(await db.listMaterializedViews()).toEqual(["adults"]);
+    await expect(db.openMaterializedView("people")).rejects.toThrow(
+      "not a materialized view",
+    );
+  });
+
+  it("rejects an invalid expression at create time", async () => {
+    await expect(
+      db.createMaterializedView("bad", "people", {
+        select: [["x", "missing + 1"]],
+      }),
+    ).rejects.toThrow("missing");
+  });
+
+  it("rejects invalid numeric options before creating anything", async () => {
+    for (const limit of [-5, 1.5, Infinity, NaN]) {
+      await expect(
+        db.createMaterializedView("bad", "people", { limit }),
+      ).rejects.toThrow("non-negative integer");
+    }
+    expect(await db.listMaterializedViews()).toEqual([]);
+
+    const view = await db.createMaterializedView("copy", "people");
+    for (const sourceVersion of [-1, 1.5, Infinity, NaN]) {
+      await expect(view.refresh({ sourceVersion })).rejects.toThrow(
+        "non-negative integer",
+      );
+    }
+  });
+
+  it("quotes bare select names", async () => {
+    await db.createTable("odd_names", [{ "order item": "widget" }], {
+      storageOptions: { newTableEnableStableRowIds: "true" },
+    });
+    const view = await db.createMaterializedView("quoted", "odd_names", {
+      select: ["order item"],
+    });
+    const result = await view.refresh();
+    expect(Number(result.rowsWritten)).toBe(1);
+  });
+
+  it("requires stable row ids on the source", async () => {
+    await db.createTable("plain", [{ x: 1 }]);
+    await expect(db.createMaterializedView("v", "plain")).rejects.toThrow(
+      "stable row ids",
+    );
+  });
+});

nodejs/__test__/remote.test.ts

@@ -75,6 +75,25 @@ async function withMockDatabase(
 }
 
 describe("remote connection", () => {
+  it("refuses materialized views before issuing any request", async () => {
+    const paths: string[] = [];
+    await withMockDatabase(
+      (req, res) => {
+        paths.push(req.url ?? "");
+        res.writeHead(404).end();
+      },
+      async (db) => {
+        await expect(db.openMaterializedView("secret_table")).rejects.toThrow(
+          /only on local databases/,
+        );
+        await expect(db.listMaterializedViews()).rejects.toThrow(
+          /only on local databases/,
+        );
+        expect(paths).toEqual([]);
+      },
+    );
+  });
+
   it("should accept partial connection options", async () => {
     await connect("db://test", {
       apiKey: "fake",

nodejs/lancedb/connection.ts

@@ -16,6 +16,12 @@ import {
   makeEmptyTable,
 } from "./arrow";
 import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
+import {
+  MaterializedView,
+  MaterializedViewSelect,
+  normalizeSelect,
+  validateNonNegativeInteger,
+} from "./materialized_view";
 import { Connection as LanceDbConnection } from "./native";
 import type {
   CreateNamespaceResponse,
@@ -247,6 +253,41 @@ export abstract class Connection {
    * @param {string[]} namespacePath - The namespace path of the table (defaults to root namespace)
    * @param {Partial<OpenTableOptions>} options - Additional options
    */
+  /**
+   * Define a materialized view named `name` over the table `source`.
+   *
+   * The view is created empty, with the query recorded in its schema
+   * metadata; `view.refresh()` computes the rows. The view is a normal
+   * table: it can be queried, indexed and searched, and it appears in
+   * `tableNames`. The source table must have stable row ids (create it with
+   * the `newTableEnableStableRowIds` storage option); they keep the view's
+   * provenance valid across source compactions and cannot be enabled after
+   * a table exists. Local databases only.
+   */
+  abstract createMaterializedView(
+    name: string,
+    source: string,
+    options?: {
+      select?: MaterializedViewSelect;
+      where?: string;
+      limit?: number;
+    },
+  ): Promise<MaterializedView>;
+
+  /**
+   * Open the materialized view named `name`.
+   *
+   * Rejects a table that exists but is not a materialized view.
+   */
+  abstract openMaterializedView(name: string): Promise<MaterializedView>;
+
+  /**
+   * The names of the materialized views in this database.
+   *
+   * Found by reading every table's schema, so this costs an open per table.
+   */
+  abstract listMaterializedViews(): Promise<string[]>;
+
   abstract openTable(
     name: string,
     namespacePath?: string[],
@@ -531,6 +572,35 @@ export class LocalConnection extends Connection {
     );
   }
 
+  async createMaterializedView(
+    name: string,
+    source: string,
+    options?: {
+      select?: MaterializedViewSelect;
+      where?: string;
+      limit?: number;
+    },
+  ): Promise<MaterializedView> {
+    validateNonNegativeInteger(options?.limit, "limit");
+    const innerTable = await this.inner.createMaterializedView(
+      name,
+      source,
+      normalizeSelect(options?.select),
+      options?.where,
+      options?.limit,
+    );
+    return new MaterializedView(new LocalTable(innerTable));
+  }
+
+  async openMaterializedView(name: string): Promise<MaterializedView> {
+    const innerTable = await this.inner.openMaterializedView(name);
+    return new MaterializedView(new LocalTable(innerTable));
+  }
+
+  async listMaterializedViews(): Promise<string[]> {
+    return await this.inner.listMaterializedViews();
+  }
+
   async openTable(
     name: string,
     namespacePath?: string[],

nodejs/lancedb/embedding/registry.ts

@@ -126,8 +126,10 @@ export class EmbeddingFunctionRegistry {
             throw new Error(`Function "${f.name}" not found in registry`);
           }
           const func = await this.get(f.name)!.create(f.model);
+          // Keyed by the column written, not the function: one function can
+          // write several, and keying by name would drop all but the last.
           return [
-            f.name,
+            f.vectorColumn ?? f.name,
             {
               sourceColumn: f.sourceColumn,
               vectorColumn: f.vectorColumn,

nodejs/lancedb/index.ts

@@ -21,6 +21,11 @@ import type { BaseTokenizer } from "./indices";
 import type { FtsToken } from "./table";
 
 // Re-export native header provider for use with connectWithHeaderProvider
+export {
+  MaterializedView,
+  MaterializedViewDefinition,
+  MaterializedViewSelect,
+} from "./materialized_view";
 export { JsHeaderProvider as NativeJsHeaderProvider } from "./native.js";
 
 // OpenTelemetry metrics bridge. Only the high-level entry point is public; the
@@ -51,6 +56,7 @@ export {
   AddResult,
   AddColumnsResult,
   RefreshColumnResult,
+  RefreshMaterializedViewResult,
   AlterColumnsResult,
   UpdateFieldMetadataResult,
   DeleteResult,

nodejs/lancedb/materialized_view.ts

@@ -0,0 +1,152 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import { RefreshMaterializedViewResult } from "./native";
+import { Table } from "./table";
+
+/** Schema metadata key holding a materialized view's definition. */
+export const DEFINITION_META_KEY = "mv.definition";
+
+/** The query that defines a materialized view. */
+export interface MaterializedViewDefinition {
+  /** Name of the source table, in the same database as the view. */
+  sourceTable: string;
+  /** `[output column, SQL expression]` pairs, in view schema order. */
+  projections: [string, string][];
+  /** SQL predicate selecting the source rows the view holds. */
+  filter?: string;
+  /** Cap on the number of rows the view holds. */
+  limit?: number;
+  /** Source columns the projections and filter read. */
+  inputs: string[];
+}
+
+/**
+ * The view's columns: column names, `[alias, SQL expression]` pairs, or a
+ * record of the same. A bare name projects itself.
+ */
+export type MaterializedViewSelect =
+  | (string | [string, string])[]
+  | Record<string, string>;
+
+/**
+ * @internal Reject a numeric option N-API would otherwise silently coerce:
+ * `Infinity` reaches Rust as 0, `1.5` as 1.
+ */
+export function validateNonNegativeInteger(
+  value: number | undefined,
+  name: string,
+): void {
+  if (value !== undefined && !(Number.isSafeInteger(value) && value >= 0)) {
+    throw new Error(`${name} must be a non-negative integer`);
+  }
+}
+
+/** @internal Quote a column name as a Lance SQL identifier (backticks). */
+function quoteIdentifier(name: string): string {
+  return "`" + name.replace(/`/g, "``") + "`";
+}
+
+/**
+ * @internal Normalize a select argument into `[alias, expression]` pairs.
+ * A bare name projects itself and is quoted, so any valid column name works;
+ * pair and record entries are kept verbatim because their right side is an
+ * expression.
+ */
+export function normalizeSelect(
+  select?: MaterializedViewSelect,
+): [string, string][] | undefined {
+  if (select === undefined) {
+    return undefined;
+  }
+  if (Array.isArray(select)) {
+    return select.map((item) =>
+      typeof item === "string" ? [item, quoteIdentifier(item)] : item,
+    );
+  }
+  return Object.entries(select);
+}
+
+/** @internal Parse a definition off a table's stored schema metadata. */
+export function definitionFromMetadata(
+  metadata: Map<string, string>,
+  name: string,
+): MaterializedViewDefinition {
+  const raw = metadata.get(DEFINITION_META_KEY);
+  if (raw === undefined) {
+    throw new Error(`Table '${name}' is not a materialized view`);
+  }
+  // biome-ignore lint/suspicious/noExplicitAny: raw JSON
+  const value: any = JSON.parse(raw);
+  if (value.kind !== "select") {
+    throw new Error(
+      `materialized view '${name}' is defined by '${value.kind}', which this ` +
+        "version of lancedb cannot refresh",
+    );
+  }
+  return {
+    sourceTable: value.source_table,
+    // biome-ignore lint/suspicious/noExplicitAny: raw JSON
+    projections: (value.projections ?? []).map((p: any) => [
+      p.output,
+      p.expression,
+    ]),
+    filter: value.filter ?? undefined,
+    limit: value.limit ?? undefined,
+    inputs: value.inputs ?? [],
+  };
+}
+
+/**
+ * A handle on a materialized view: its table plus its definition.
+ *
+ * Obtained from {@link Connection#createMaterializedView} or
+ * {@link Connection#openMaterializedView}. The view is a normal table --
+ * queries, indexes and search all apply through {@link MaterializedView#table}
+ * -- whose contents are maintained by {@link MaterializedView#refresh}.
+ */
+export class MaterializedView {
+  private readonly inner: Table;
+
+  constructor(table: Table) {
+    this.inner = table;
+  }
+
+  get name(): string {
+    return this.inner.name;
+  }
+
+  /** The view, as the table it is. */
+  table(): Table {
+    return this.inner;
+  }
+
+  /** The query that defines the view, read from its stored schema. */
+  async definition(): Promise<MaterializedViewDefinition> {
+    const schema = await this.inner.schema();
+    return definitionFromMetadata(schema.metadata, this.name);
+  }
+
+  /**
+   * Recompute the view from its source.
+   *
+   * The refresh is incremental when the source's changes can be reconciled
+   * into the view -- rows added, changed or removed since the last one --
+   * and otherwise rebuilds. `full` forces a rebuild; `sourceVersion`
+   * refreshes to that source version instead of the latest.
+   *
+   * Concurrent refreshes of one view do not duplicate its rows. Two that
+   * plan the same source rows conflict on commit, and the loser throws
+   * rather than writing them a second time.
+   */
+  async refresh(options?: {
+    full?: boolean;
+    sourceVersion?: number;
+  }): Promise<RefreshMaterializedViewResult> {
+    validateNonNegativeInteger(options?.sourceVersion, "sourceVersion");
+    return await this.inner.refreshMaterializedView(
+      options?.full,
+      options?.sourceVersion,
+    );
+  }
+}

nodejs/lancedb/table.ts

@@ -34,6 +34,7 @@ import {
   Branches as NativeBranches,
   OptimizeStats,
   RefreshColumnResult,
+  RefreshMaterializedViewResult,
   TableStatistics,
   Tags,
   UpdateFieldMetadataResult,
@@ -595,6 +596,18 @@ export abstract class Table {
    */
   abstract refreshColumnAsync(column: string): Promise<Job>;
 
+  /**
+   * Recompute this table's contents from its materialized-view definition.
+   *
+   * Plumbing for {@link MaterializedView.refresh}, which is the way to call
+   * it: rejects tables that carry no view definition. Local tables only.
+   * @ignore
+   */
+  abstract refreshMaterializedView(
+    full?: boolean,
+    sourceVersion?: number,
+  ): Promise<RefreshMaterializedViewResult>;
+
   /**
    * Alter the name or nullability of columns.
    * @param {ColumnAlteration[]} columnAlterations One or more alterations to
@@ -1204,6 +1217,13 @@ export class LocalTable extends Table {
     return await this.inner.refreshColumnAsync(column);
   }
 
+  async refreshMaterializedView(
+    full?: boolean,
+    sourceVersion?: number,
+  ): Promise<RefreshMaterializedViewResult> {
+    return await this.inner.refreshMaterializedView(full, sourceVersion);
+  }
+
   async alterColumns(
     columnAlterations: ColumnAlteration[],
   ): Promise<AlterColumnsResult> {

nodejs/src/connection.rs

@@ -266,6 +266,58 @@ impl Connection {
         Ok(Table::new(tbl))
     }
 
+    #[napi(catch_unwind)]
+    pub async fn create_materialized_view(
+        &self,
+        name: String,
+        source: String,
+        projections: Option<Vec<Vec<String>>>,
+        filter: Option<String>,
+        limit: Option<i64>,
+    ) -> napi::Result<Table> {
+        let mut builder = self.get_inner()?.create_materialized_view(name, source);
+        if let Some(projections) = projections {
+            let mut pairs = Vec::with_capacity(projections.len());
+            for pair in projections {
+                let [output, expression]: [String; 2] = pair.try_into().map_err(|_| {
+                    napi::Error::from_reason("each projection must be an [output, expression] pair")
+                })?;
+                pairs.push((output, expression));
+            }
+            builder = builder.select(pairs);
+        }
+        if let Some(filter) = filter {
+            builder = builder.only_if(filter);
+        }
+        if let Some(limit) = limit {
+            let limit = u64::try_from(limit)
+                .map_err(|_| napi::Error::from_reason("limit must be a non-negative integer"))?;
+            builder = builder.limit(limit);
+        }
+        let view = builder.execute().await.default_error()?;
+        Ok(Table::new(view.table().clone()))
+    }
+
+    #[napi(catch_unwind)]
+    pub async fn open_materialized_view(&self, name: String) -> napi::Result<Table> {
+        let view = self
+            .get_inner()?
+            .open_materialized_view(&name)
+            .await
+            .default_error()?;
+        Ok(Table::new(view.table().clone()))
+    }
+
+    #[napi(catch_unwind)]
+    pub async fn list_materialized_views(&self) -> napi::Result<Vec<String>> {
+        let views = self
+            .get_inner()?
+            .list_materialized_views()
+            .await
+            .default_error()?;
+        Ok(views.into_iter().map(|v| v.name).collect())
+    }
+
     #[napi(catch_unwind)]
     pub async fn open_table(
         &self,

nodejs/src/lib.rs

@@ -1,6 +1,10 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
+// The materialized-view refresh future deepens the type graph past the
+// default trait-recursion depth; same raise as the core crate applies.
+#![recursion_limit = "256"]
+
 use std::collections::HashMap;
 
 use env_logger::Env;

nodejs/src/table.rs

@@ -381,6 +381,26 @@ impl Table {
         Ok(crate::job::Job::new(job))
     }
 
+    #[napi(catch_unwind)]
+    pub async fn refresh_materialized_view(
+        &self,
+        full: Option<bool>,
+        source_version: Option<i64>,
+    ) -> napi::Result<RefreshMaterializedViewResult> {
+        let view = lancedb::MaterializedView::from_table(self.inner_ref()?.clone())
+            .await
+            .default_error()?;
+        let mut builder = view.refresh().full(full.unwrap_or(false));
+        if let Some(version) = source_version {
+            let version = u64::try_from(version).map_err(|_| {
+                napi::Error::from_reason("sourceVersion must be a non-negative integer")
+            })?;
+            builder = builder.source_version(version);
+        }
+        let result = builder.execute().await.default_error()?;
+        Ok(result.into())
+    }
+
     #[napi(catch_unwind)]
     pub async fn add_columns_with_schema(
         &self,
@@ -1236,6 +1256,31 @@ pub struct RefreshColumnResult {
     pub version: i64,
 }
 
+#[napi(object)]
+pub struct RefreshMaterializedViewResult {
+    /// How the view was brought up to date: "rebuild", "incremental" or "no_op".
+    pub mode: String,
+    pub rows_written: i64,
+    pub source_version: i64,
+    pub version: i64,
+}
+
+impl From<lancedb::RefreshMaterializedViewResult> for RefreshMaterializedViewResult {
+    fn from(value: lancedb::RefreshMaterializedViewResult) -> Self {
+        let mode = match value.mode {
+            lancedb::RefreshMode::Rebuild => "rebuild",
+            lancedb::RefreshMode::Incremental => "incremental",
+            lancedb::RefreshMode::NoOp => "no_op",
+        };
+        Self {
+            mode: mode.to_string(),
+            rows_written: value.rows_written as i64,
+            source_version: value.source_version as i64,
+            version: value.version as i64,
+        }
+    }
+}
+
 impl From<lancedb::table::RefreshColumnResult> for RefreshColumnResult {
     fn from(value: lancedb::table::RefreshColumnResult) -> Self {
         Self {