diff --git a/docs/src/js/classes/MaterializedView.md b/docs/src/js/classes/MaterializedView.md index bb02f5dc2..3ba4d56b3 100644 --- a/docs/src/js/classes/MaterializedView.md +++ b/docs/src/js/classes/MaterializedView.md @@ -66,14 +66,18 @@ refresh(options?): Promise Recompute the view from its source. Local refresh requires a source build with `experimental-functions`. -Every refresh rebuilds the complete view. For local views, -`sourceVersion` refreshes to that source version instead of the latest; -remote SQL refreshes do not support source-version pinning. +`full: true` forces a remote rebuild; otherwise the server chooses the +refresh mode. Local views always rebuild, and `sourceVersion` selects a +source version instead of the latest. Remote SQL refreshes do not support +source-version pinning. Their results leave `mode`, `rowsWritten`, and +`sourceVersion` absent because the SQL job adapter does not expose these metrics. #### Parameters * **options?** +* **options.full?**: `boolean` + * **options.sourceVersion?**: `number` #### Returns diff --git a/docs/src/js/interfaces/RefreshMaterializedViewResult.md b/docs/src/js/interfaces/RefreshMaterializedViewResult.md index 1c8b8bc21..05321683e 100644 --- a/docs/src/js/interfaces/RefreshMaterializedViewResult.md +++ b/docs/src/js/interfaces/RefreshMaterializedViewResult.md @@ -8,30 +8,34 @@ ## Properties -### mode +### mode? ```ts -mode: string; +optional mode: string; ``` -How the view was brought up to date: "rebuild". +How the view was brought up to date: "rebuild", or absent for remote SQL refreshes. *** -### rowsWritten +### rowsWritten? ```ts -rowsWritten: number; +optional rowsWritten: number; ``` +Rows written by this refresh, or absent for remote SQL refreshes. + *** -### sourceVersion +### sourceVersion? ```ts -sourceVersion: number; +optional sourceVersion: number; ``` +Source version refreshed to, or absent for remote SQL refreshes. + *** ### version diff --git a/nodejs/__test__/materialized_view.test.ts b/nodejs/__test__/materialized_view.test.ts index 2324e1fcf..0840395fe 100644 --- a/nodejs/__test__/materialized_view.test.ts +++ b/nodejs/__test__/materialized_view.test.ts @@ -6,6 +6,7 @@ import * as tmp from "tmp"; import { Connection, connect } from "../lancedb"; import { DEFINITION_META_KEY, + MaterializedView, definitionFromMetadata, } from "../lancedb/materialized_view"; import { experimentalFunctionsEnabled } from "../lancedb/native"; @@ -94,20 +95,33 @@ describe("materialized views", () => { ); }); - experimentalTest("fully refreshes after an append", async () => { - const view = await db.createMaterializedView("copy", "people", { - withNoData: true, - }); - await view.refresh(); + experimentalTest.each([false, true])( + "fully refreshes after an append (full=%s)", + async (full) => { + const view = await db.createMaterializedView("copy", "people", { + withNoData: true, + }); + 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("rebuild"); - expect(Number(result.rowsWritten)).toBe(4); - expect(await view.table().countRows()).toBe(4); + const people = await db.openTable("people"); + await people.add([{ name: "alan", age: 41 }]); + const result = await view.refresh({ full }); + expect(result.mode).toBe("rebuild"); + expect(Number(result.rowsWritten)).toBe(4); + expect(await view.table().countRows()).toBe(4); - expect((await view.refresh()).mode).toBe("rebuild"); + expect((await view.refresh()).mode).toBe("rebuild"); + }, + ); + + it("forwards a forced refresh and preserves absent remote metrics", async () => { + const view = new MaterializedView(await db.openTable("people")); + const remoteResult = { version: 42 }; + const refresh = jest + .spyOn(view.table(), "refreshMaterializedView") + .mockResolvedValue(remoteResult); + expect(await view.refresh({ full: true })).toEqual(remoteResult); + expect(refresh).toHaveBeenCalledWith(undefined, true); }); experimentalTest("lists views and rejects non-views", async () => { diff --git a/nodejs/lancedb/materialized_view.ts b/nodejs/lancedb/materialized_view.ts index cc209b14d..afb61b9ab 100644 --- a/nodejs/lancedb/materialized_view.ts +++ b/nodejs/lancedb/materialized_view.ts @@ -188,14 +188,20 @@ export class MaterializedView { * Recompute the view from its source. * Local refresh requires a source build with `experimental-functions`. * - * Every refresh rebuilds the complete view. For local views, - * `sourceVersion` refreshes to that source version instead of the latest; - * remote SQL refreshes do not support source-version pinning. + * `full: true` forces a remote rebuild; otherwise the server chooses the + * refresh mode. Local views always rebuild, and `sourceVersion` selects a + * source version instead of the latest. Remote SQL refreshes do not support + * source-version pinning. Their results leave `mode`, `rowsWritten`, and + * `sourceVersion` absent because the SQL job adapter does not expose these metrics. */ async refresh(options?: { + full?: boolean; sourceVersion?: number; }): Promise { validateNonNegativeInteger(options?.sourceVersion, "sourceVersion"); - return await this.inner.refreshMaterializedView(options?.sourceVersion); + return await this.inner.refreshMaterializedView( + options?.sourceVersion, + options?.full, + ); } } diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 4738b6140..b88603fb7 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -699,6 +699,7 @@ export abstract class Table { */ abstract refreshMaterializedView( sourceVersion?: number, + full?: boolean, ): Promise; /** @ignore */ @@ -1442,8 +1443,9 @@ export class LocalTable extends Table { async refreshMaterializedView( sourceVersion?: number, + full?: boolean, ): Promise { - return await this.inner.refreshMaterializedView(sourceVersion); + return await this.inner.refreshMaterializedView(sourceVersion, full); } async materializedViewDefinition(): Promise { diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index 0a40e0a13..a1410ad85 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -470,11 +470,12 @@ impl Table { pub async fn refresh_materialized_view( &self, source_version: Option, + full: Option, ) -> napi::Result { let view = lancedb::MaterializedView::from_table(self.inner_ref()?.clone()) .await .default_error()?; - let mut builder = view.refresh(); + 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") @@ -1615,22 +1616,24 @@ impl From for FunctionErrors { #[napi(object)] pub struct RefreshMaterializedViewResult { - /// How the view was brought up to date: "rebuild". - pub mode: String, - pub rows_written: i64, - pub source_version: i64, + /// How the view was brought up to date: "rebuild", or absent for remote SQL refreshes. + pub mode: Option, + /// Rows written by this refresh, or absent for remote SQL refreshes. + pub rows_written: Option, + /// Source version refreshed to, or absent for remote SQL refreshes. + pub source_version: Option, pub version: i64, } impl From for RefreshMaterializedViewResult { fn from(value: lancedb::RefreshMaterializedViewResult) -> Self { - let mode = match value.mode { + let mode = value.mode.map(|mode| match mode { lancedb::RefreshMode::Rebuild => "rebuild", - }; + }); Self { - mode: mode.to_string(), - rows_written: value.rows_written as i64, - source_version: value.source_version as i64, + mode: mode.map(str::to_string), + rows_written: value.rows_written.map(|rows| rows as i64), + source_version: value.source_version.map(|version| version as i64), version: value.version as i64, } } diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 333fca92e..7e1307a4a 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -528,10 +528,10 @@ class Table: limit: Optional[int] = None, ) -> FunctionErrors: ... async def refresh_materialized_view( - self, source_version: Optional[int] = None + self, source_version: Optional[int] = None, *, full: bool = False ) -> RefreshMaterializedViewResult: ... async def refresh_materialized_view_async( - self, source_version: Optional[int] = None + self, source_version: Optional[int] = None, *, full: bool = False ) -> Job: ... async def materialized_view_definition(self) -> str: ... async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ... @@ -921,9 +921,9 @@ class FunctionErrors: class RefreshMaterializedViewResult: @staticmethod def from_json(value: str) -> RefreshMaterializedViewResult: ... - mode: str - rows_written: int - source_version: int + mode: Optional[str] + rows_written: Optional[int] + source_version: Optional[int] version: int class AlterColumnsResult: diff --git a/python/python/lancedb/materialized_view.py b/python/python/lancedb/materialized_view.py index 8efe31f05..7fc110dcc 100644 --- a/python/python/lancedb/materialized_view.py +++ b/python/python/lancedb/materialized_view.py @@ -184,31 +184,36 @@ class AsyncMaterializedView: return _definition_from_json(raw) async def refresh( - self, *, source_version: Optional[int] = None + self, *, full: bool = False, source_version: Optional[int] = None ) -> "RefreshMaterializedViewResult": """Recompute the view from its source. - Every refresh rebuilds the complete view. For local views, - ``source_version`` refreshes to that source version instead of the - latest; remote SQL refreshes do not support source-version pinning. + ``full=True`` forces a remote rebuild; otherwise the server chooses + the refresh mode. Local views always rebuild, and ``source_version`` + selects a source version instead of the latest. Remote SQL refreshes + do not support source-version pinning. + + Remote results leave ``mode``, ``rows_written``, and ``source_version`` + as ``None`` because the SQL job adapter does not expose these metrics. """ return await self._table._inner.refresh_materialized_view( - source_version=source_version + source_version=source_version, full=full ) async def refresh_async( - self, *, source_version: Optional[int] = None + self, *, full: bool = False, source_version: Optional[int] = None ) -> "AsyncJob[RefreshMaterializedViewResult]": """Submit a refresh and return its job without waiting. The job may already be complete for a local view. Remote refreshes are - submitted through the SQL service. + submitted through the SQL service. Options and result metrics follow + [refresh][lancedb.materialized_view.AsyncMaterializedView.refresh]. """ from ._lancedb import RefreshMaterializedViewResult return _typed_job( await self._table._inner.refresh_materialized_view_async( - source_version=source_version + source_version=source_version, full=full ), RefreshMaterializedViewResult.from_json, ) @@ -240,18 +245,22 @@ class MaterializedView: return LOOP.run(self._async.definition()) def refresh( - self, *, source_version: Optional[int] = None + self, *, full: bool = False, source_version: Optional[int] = None ) -> "RefreshMaterializedViewResult": """Recompute the view from its source. See [AsyncMaterializedView.refresh][lancedb.materialized_view.AsyncMaterializedView.refresh].""" - return LOOP.run(self._async.refresh(source_version=source_version)) + return LOOP.run(self._async.refresh(full=full, source_version=source_version)) def refresh_async( - self, *, source_version: Optional[int] = None + self, *, full: bool = False, source_version: Optional[int] = None ) -> "Job[RefreshMaterializedViewResult]": """Submit a refresh and return its job without waiting. See [AsyncMaterializedView.refresh_async][lancedb.materialized_view.AsyncMaterializedView.refresh_async]. """ - return Job(LOOP.run(self._async.refresh_async(source_version=source_version))) + return Job( + LOOP.run( + self._async.refresh_async(full=full, source_version=source_version) + ) + ) diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py index 337fcad78..04dbe2fec 100644 --- a/python/python/tests/test_materialized_views.py +++ b/python/python/tests/test_materialized_views.py @@ -8,6 +8,7 @@ import threading import lancedb import pytest +import pyarrow as pa from lancedb.db import AsyncConnection from lancedb.materialized_view import MaterializedViewDefinition from lancedb.remote.db import RemoteDBConnection @@ -112,6 +113,98 @@ def test_remote_drop_async_returns_server_job(): assert requests == [("/v1/materialized_view/adults/drop", {})] +@contextlib.contextmanager +def mock_remote_materialized_view_refresh(): + requests = [] + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + requests.append(self.path) + if self.path == "/v1/materialized_view/view/describe": + response = {"query": "SELECT * FROM source"} + else: + assert self.path == "/v1/table/view/describe/" + response = {"version": 42, "schema": {"fields": []}} + encoded = json.dumps(response).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + with http.server.HTTPServer(("localhost", 0), Handler) as server: + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + yield f"http://localhost:{server.server_address[1]}", requests + finally: + server.shutdown() + thread.join() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [False, True]) +@pytest.mark.parametrize("background", [False, True]) +@pytest.mark.parametrize("full", [False, True]) +async def test_remote_refresh_reports_unknown_metrics(sync, background, full): + flight = pytest.importorskip("pyarrow.flight") + + class RefreshServer(flight.FlightServerBase): + def __init__(self): + super().__init__("grpc://127.0.0.1:0") + self.commands = [] + + def do_get(self, context, ticket): + self.commands.append(ticket.ticket) + # SQL completion carries no refresh statistics. + return flight.RecordBatchStream(pa.table({"status": ["ok"]})) + + with ( + mock_remote_materialized_view_refresh() as (host, requests), + RefreshServer() as server, + ): + options = {"full": True} if full else {} + connect = lancedb.connect if sync else lancedb.connect_async + db = connect( + "db://dev", + api_key="fake", + host_override=host, + sql_host_override=f"grpc://127.0.0.1:{server.port}", + client_config={"retry_config": {"retries": 0}}, + ) + if sync: + view = db.open_materialized_view("view") + result = ( + view.refresh_async(**options).wait() + if background + else view.refresh(**options) + ) + else: + db = await db + view = await db.open_materialized_view("view") + if background: + job = await view.refresh_async(**options) + result = await job.wait() + else: + result = await view.refresh(**options) + + assert result.mode is None + assert result.rows_written is None + assert result.source_version is None + assert result.version == 42 + assert repr(result) == ( + "RefreshMaterializedViewResult(mode=None, rows_written=None, " + "source_version=None, version=42)" + ) + (command,) = server.commands + assert b'REFRESH MATERIALIZED VIEW "view"' in command + assert (b"WITH (full = true)" in command) is full + assert not any("count_rows" in path for path in requests) + + def test_sync_remote_create_uses_public_async_connection(): calls = [] @@ -226,20 +319,37 @@ def test_definition_round_trips(tmp_path): @pytest.mark.experimental_functions -def test_full_refresh_after_append(tmp_path): +@pytest.mark.parametrize("full", [False, True]) +def test_full_refresh_after_append(tmp_path, full): db = make_db(tmp_path) view = db.create_materialized_view("copy", "people", with_no_data=True) view.refresh() db.open_table("people").add([{"name": "alan", "age": 41}]) - result = view.refresh() + result = view.refresh(full=full) assert result.mode == "rebuild" assert result.rows_written == 4 + assert result.source_version == db.open_table("people").version assert view.table.count_rows() == 4 assert view.refresh().mode == "rebuild" +@pytest.mark.experimental_functions +def test_full_refresh_can_pin_local_source_version(tmp_path): + db = make_db(tmp_path) + source = db.open_table("people") + version = source.version + view = db.create_materialized_view("copy", "people", with_no_data=True) + source.add([{"name": "alan", "age": 41}]) + + result = view.refresh(full=True, source_version=version) + assert result.mode == "rebuild" + assert result.rows_written == 3 + assert result.source_version == version + assert view.table.count_rows() == 3 + + @pytest.mark.experimental_functions def test_full_refresh_after_update(tmp_path): db = make_db(tmp_path) diff --git a/python/src/table.rs b/python/src/table.rs index 62fc9e611..d77fd14b8 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -694,9 +694,9 @@ impl From for FunctionErrors { #[pyclass(get_all, from_py_object)] #[derive(Clone, Debug)] pub struct RefreshMaterializedViewResult { - pub mode: String, - pub rows_written: u64, - pub source_version: u64, + pub mode: Option, + pub rows_written: Option, + pub source_version: Option, pub version: u64, } @@ -716,18 +716,23 @@ impl RefreshMaterializedViewResult { pub fn __repr__(&self) -> String { format!( "RefreshMaterializedViewResult(mode={}, rows_written={}, source_version={}, version={})", - self.mode, self.rows_written, self.source_version, self.version + self.mode.as_deref().unwrap_or("None"), + self.rows_written + .map_or_else(|| "None".into(), |value| value.to_string()), + self.source_version + .map_or_else(|| "None".into(), |value| value.to_string()), + self.version ) } } impl From for RefreshMaterializedViewResult { fn from(result: lancedb::RefreshMaterializedViewResult) -> Self { - let mode = match result.mode { + let mode = result.mode.map(|mode| match mode { lancedb::RefreshMode::Rebuild => "rebuild", - }; + }); Self { - mode: mode.to_string(), + mode: mode.map(str::to_string), rows_written: result.rows_written, source_version: result.source_version, version: result.version, @@ -2054,17 +2059,18 @@ impl Table { }) } - #[pyo3(signature = (source_version=None))] + #[pyo3(signature = (source_version=None, *, full=false))] pub fn refresh_materialized_view( self_: PyRef<'_, Self>, source_version: Option, + full: bool, ) -> PyResult> { let inner = self_.inner_ref()?.clone(); future_into_py(self_.py(), async move { let view = lancedb::MaterializedView::from_table(inner) .await .infer_error()?; - let mut builder = view.refresh(); + let mut builder = view.refresh().full(full); if let Some(version) = source_version { builder = builder.source_version(version); } @@ -2073,17 +2079,18 @@ impl Table { }) } - #[pyo3(signature = (source_version=None))] + #[pyo3(signature = (source_version=None, *, full=false))] pub fn refresh_materialized_view_async( self_: PyRef<'_, Self>, source_version: Option, + full: bool, ) -> PyResult> { let inner = self_.inner_ref()?.clone(); future_into_py(self_.py(), async move { let view = lancedb::MaterializedView::from_table(inner) .await .infer_error()?; - let mut builder = view.refresh(); + let mut builder = view.refresh().full(full); if let Some(version) = source_version { builder = builder.source_version(version); } diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index b7124d930..5a7169b7c 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -1844,14 +1844,17 @@ impl MaterializedView { /// Recompute the view from its source. /// - /// Every refresh rebuilds the view from the selected source version. + /// Local refreshes rebuild the view from the selected source version. + /// Remote refreshes let the server select a mode unless + /// [`RefreshMaterializedViewBuilder::full`] requests a rebuild. Their + /// result leaves unavailable refresh metrics as `None`. /// /// ```no_run /// # #![recursion_limit = "256"] /// # use lancedb::materialized_view::MaterializedView; /// # async fn refresh(view: &MaterializedView) -> Result<(), Box> { /// let result = view.refresh().execute().await?; - /// println!("{:?}: {} rows", result.mode, result.rows_written); + /// println!("{:?}: {:?} rows", result.mode, result.rows_written); /// # Ok(()) /// # } /// ``` @@ -1859,6 +1862,7 @@ impl MaterializedView { RefreshMaterializedViewBuilder { view: self.clone(), source_version: None, + full: false, } } } @@ -1867,9 +1871,26 @@ impl MaterializedView { pub struct RefreshMaterializedViewBuilder { view: MaterializedView, source_version: Option, + full: bool, } impl RefreshMaterializedViewBuilder { + /// Force a complete rebuild instead of letting the remote server choose + /// the refresh mode. Defaults to `false`. Local views always rebuild. + /// + /// ```no_run + /// # #![recursion_limit = "256"] + /// # use lancedb::materialized_view::MaterializedView; + /// # async fn refresh(view: &MaterializedView) -> lancedb::Result<()> { + /// view.refresh().full(true).execute().await?; + /// # Ok(()) + /// # } + /// ``` + pub fn full(mut self, full: bool) -> Self { + self.full = full; + self + } + /// Refresh a local view to this source table version instead of the latest. /// Remote SQL refreshes do not support source-version pinning. pub fn source_version(mut self, version: u64) -> Self { @@ -1884,7 +1905,7 @@ impl RefreshMaterializedViewBuilder { .view .table .base_table() - .refresh_materialized_view_async(self.source_version) + .refresh_materialized_view_async(self.source_version, self.full) .await; } crate::require_experimental_functions()?; @@ -2102,19 +2123,23 @@ mod tests { .execute() .await .unwrap(); - let refreshed = view.refresh().execute().await.unwrap(); - assert_eq!(refreshed.mode, RefreshMode::Rebuild); - assert_eq!(refreshed.rows_written, 3); + let refreshed = view.refresh().full(true).execute().await.unwrap(); + assert_eq!(refreshed.mode, Some(RefreshMode::Rebuild)); + assert_eq!(refreshed.rows_written, Some(3)); + assert_eq!( + refreshed.source_version, + Some(source.version().await.unwrap()) + ); assert_eq!(view.table().count_rows(None).await.unwrap(), 3); let unchanged = view.refresh().execute().await.unwrap(); - assert_eq!(unchanged.mode, RefreshMode::Rebuild); - assert_eq!(unchanged.rows_written, 3); + assert_eq!(unchanged.mode, Some(RefreshMode::Rebuild)); + assert_eq!(unchanged.rows_written, Some(3)); source.delete("true").await.unwrap(); let emptied = view.refresh().execute().await.unwrap(); - assert_eq!(emptied.mode, RefreshMode::Rebuild); - assert_eq!(emptied.rows_written, 0); + assert_eq!(emptied.mode, Some(RefreshMode::Rebuild)); + assert_eq!(emptied.rows_written, Some(0)); assert_eq!(view.table().count_rows(None).await.unwrap(), 0); } @@ -2274,10 +2299,13 @@ mod tests { // Refresh, chained, is the writer. source.add(row()).execute().await.unwrap(); - assert_eq!(view.refresh().execute().await.unwrap().rows_written, 3); + assert_eq!( + view.refresh().execute().await.unwrap().rows_written, + Some(3) + ); assert_eq!( downstream.refresh().execute().await.unwrap().rows_written, - 3 + Some(3) ); assert_eq!( downstream @@ -2312,7 +2340,10 @@ mod tests { .await .unwrap(); source.drop_columns(&["y"]).await.unwrap(); - assert_eq!(view.refresh().execute().await.unwrap().rows_written, 3); + assert_eq!( + view.refresh().execute().await.unwrap().rows_written, + Some(3) + ); } #[tokio::test] @@ -2355,7 +2386,10 @@ mod tests { .execute() .await .unwrap(); - assert_eq!(view.refresh().execute().await.unwrap().rows_written, 4); + assert_eq!( + view.refresh().execute().await.unwrap().rows_written, + Some(4) + ); view.table().optimize(OptimizeAction::All).await.unwrap(); assert_eq!( view.table().count_rows(Some("x > 2".into())).await.unwrap(), diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index 1186ac576..70a795eb3 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/rust/lancedb/src/materialized_view/refresh.rs @@ -45,13 +45,16 @@ pub enum RefreshMode { /// The result of refreshing a materialized view. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct RefreshMaterializedViewResult { - /// How the view was brought up to date. - pub mode: RefreshMode, - /// Rows written to the replacement view. - pub rows_written: u64, - /// The source table version the view reflects. - pub source_version: u64, - /// The view table version after the refresh. + /// How the view was brought up to date, or `None` when unavailable. + /// Remote SQL refreshes currently do not expose this metric. + pub mode: Option, + /// Rows written by this refresh, or `None` when unavailable. + /// Remote SQL refreshes currently do not expose this metric. + pub rows_written: Option, + /// The source table version the view reflects, or `None` when unavailable. + /// Remote SQL refreshes currently do not expose this metric. + pub source_version: Option, + /// The view table version observed after the refresh. pub version: u64, } @@ -182,9 +185,9 @@ pub(crate) async fn execute_refresh( let version = committed.version().version; native.dataset.update(committed); Ok(RefreshMaterializedViewResult { - mode: RefreshMode::Rebuild, - rows_written: rows_written.load(Ordering::Relaxed), - source_version, + mode: Some(RefreshMode::Rebuild), + rows_written: Some(rows_written.load(Ordering::Relaxed)), + source_version: Some(source_version), version, }) } diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs index 1368dd753..260471612 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/rust/lancedb/src/remote/sql_test.rs @@ -1021,7 +1021,7 @@ async fn rejected_job_cancellation_does_not_abort_sql_completion() { } #[tokio::test] -async fn remote_refresh_twice_keeps_latest_selector() { +async fn remote_refresh_omits_unknown_metrics_and_forwards_full() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let incoming = futures::stream::try_unfold(listener, |listener| async { @@ -1029,6 +1029,8 @@ async fn remote_refresh_twice_keeps_latest_selector() { Ok::<_, std::io::Error>(Some((socket, listener))) }); let service = TestSqlService::default(); + let statements = service.statements.clone(); + let headers = service.headers.clone(); let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); let server = tokio::spawn( tonic::transport::Server::builder() @@ -1052,11 +1054,11 @@ async fn remote_refresh_twice_keeps_latest_selector() { .body() .and_then(reqwest::Body::as_bytes) .and_then(|bytes| serde_json::from_slice::(bytes).ok()) - .unwrap(); + .unwrap_or_default(); captured.lock().unwrap().push(body["version"].clone()); let text = match request.url().path() { + "/v1/materialized_view/ns$view/describe" => r#"{"query": "SELECT * FROM source"}"#, "/v1/table/ns$view/describe/" => r#"{"version": 42, "schema": {"fields": []}}"#, - "/v1/table/ns$view/count_rows/" => "2", other => panic!("unexpected REST request: {other}"), }; http::Response::builder() @@ -1072,24 +1074,54 @@ async fn remote_refresh_twice_keeps_latest_selector() { Default::default(), Some(sql_client), ); + let database = RemoteDatabase::new_mock(|request| -> http::Response { + panic!("unexpected database request: {}", request.url()); + }); + let view = + crate::MaterializedView::from_table(crate::Table::new(Arc::new(table), Arc::new(database))) + .await + .unwrap(); - let first = table - .refresh_materialized_view_async(None) - .await - .unwrap() - .wait() - .await - .unwrap(); - let second = table - .refresh_materialized_view_async(None) + let first = view.refresh().execute().await.unwrap(); + let second = view + .refresh() + .full(true) + .execute_async() .await .unwrap() .wait() .await .unwrap(); - assert_eq!(first.version, 42); - assert_eq!(second.version, 42); + for result in [first, second] { + assert_eq!(result.mode, None); + assert_eq!(result.rows_written, None); + assert_eq!(result.source_version, None); + assert_eq!(result.version, 42); + // Query-backed jobs serialize their typed result before returning it. + let json = serde_json::to_value(result).unwrap(); + assert!(json["mode"].is_null()); + assert!(json["rows_written"].is_null()); + assert!(json["source_version"].is_null()); + } + assert!(matches!( + view.refresh().source_version(7).execute_async().await, + Err(Error::NotSupported { .. }) + )); + assert_eq!( + *statements.lock().unwrap(), + [ + "REFRESH MATERIALIZED VIEW \"view\"", + "REFRESH MATERIALIZED VIEW \"view\" WITH (full = true)", + ] + ); + assert!( + headers + .lock() + .unwrap() + .iter() + .all(|h| h.namespace_path == "ns") + ); assert!( selectors .lock() @@ -1147,7 +1179,10 @@ async fn cancelling_remote_refresh_cancels_sql_query() { Default::default(), Some(sql_client), ); - let job = table.refresh_materialized_view_async(None).await.unwrap(); + let job = table + .refresh_materialized_view_async(None, false) + .await + .unwrap(); tokio::time::timeout(Duration::from_secs(10), async { while do_get_count.load(Ordering::SeqCst) == 0 { tokio::task::yield_now().await; diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 29cf774fc..061ae8bf2 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -2224,6 +2224,7 @@ impl BaseTable for RemoteTable { async fn refresh_materialized_view_async( &self, source_version: Option, + full: bool, ) -> Result> { self.check_mutable().await?; if source_version.is_some() { @@ -2236,8 +2237,9 @@ impl BaseTable for RemoteTable { message: "SQL is unavailable for this remote table client".to_string(), })?; let statement = format!( - "REFRESH MATERIALIZED VIEW {}", - quote_sql_identifier(&self.name) + "REFRESH MATERIALIZED VIEW {}{}", + quote_sql_identifier(&self.name), + if full { " WITH (full = true)" } else { "" }, ); let namespace = self.namespace.clone(); let client = self.client.clone(); @@ -2259,15 +2261,13 @@ impl BaseTable for RemoteTable { ); table.checkout_latest().await?; let version = table.version().await?; - let rows_written = - u64::try_from(table.count_rows(None).await?).map_err(|_| Error::Runtime { - message: "materialized-view row count exceeds u64".to_string(), - })?; freshness_request.observe_version(&freshness, version); Ok(RefreshMaterializedViewResult { - mode: crate::materialized_view::RefreshMode::Rebuild, - rows_written, - source_version: 0, + // The SQL job adapter does not expose refresh metrics. Table + // metadata cannot tell us what this refresh actually did. + mode: None, + rows_written: None, + source_version: None, version, }) }) diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 7097644e4..a7db1ae5e 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -585,6 +585,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { async fn refresh_materialized_view_async( &self, _source_version: Option, + _full: bool, ) -> Result> { Err(Error::NotSupported { message: "remote materialized-view refresh is not supported on this table type".into(),