fix: correct remote materialized view refresh results and options (#4672)

Remote materialized-view refreshes drained the SQL result and then
fabricated `mode=rebuild`, `rows_written=<total view rows>`, and
`source_version=0`. This made incremental refreshes report incorrect
statistics.

Make these three result fields optional across Rust, Python, and
TypeScript, and leave them unset for remote SQL refreshes. Remove the
post-refresh row count request. Local refreshes continue to populate all
metrics, and `version` remains the view version observed after
completion. Remote per-operation statistics remain unavailable until the
SQL job adapter can retrieve them.

Expose the forced-rebuild option through the Rust builder,
synchronous/asynchronous Python refresh and job methods, and TypeScript
refresh. Remote `full=true` sends `WITH (full = true)`; the default lets
the server choose its refresh mode. Local source-version pinning and
remote job cancellation are preserved.

Local refresh regression tests respect the `experimental-functions`
build gate, while remote refresh coverage also runs in default builds.

Validation:

- Rust: 20 materialized-view tests and 21 remote SQL tests passed with
`remote,experimental-functions`; `cargo fmt --all` passed.
- Python: rebuilt the extension in the uv-managed environment. The
default build passed 23 tests with 27 feature-dependent skips. The
opt-in build passed 61 tests with 3 expected default-build-only skips,
covering the materialized-view suite and 14 async query-serialization
cases from current main. Python Ruff lint and formatting checks passed.
- TypeScript: default and opt-in native builds and TypeScript
compilation passed. The default build passed 5 tests with 10 expected
feature-dependent skips; the opt-in build passed all 14
materialized-view tests with the default-build-only test skipped.
Lint/format checks and documentation generation passed.
- Repository-wide `ruff check .` reports 48 pre-existing violations in
unchanged `docs/web-tests` files; `ruff check python` passes.

Fixes #4670

<!-- lance-gatekeeper-fix:v1 agent=ece43ec4b88a48d1789f60005fa3ecfc
generation=1 -->

Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
Co-authored-by: Xuanwo <github@xuanwo.io>
This commit is contained in:
lancedb-gatefixer[bot]
2026-10-27 21:34:01 +08:00
committed by GitHub
co-authored by Xuanwo
parent c167b99fa0
commit 2cd635d41c
15 changed files with 347 additions and 115 deletions
+7 -3
View File
@@ -66,14 +66,18 @@ refresh(options?): Promise<RefreshMaterializedViewResult>
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
@@ -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
+26 -12
View File
@@ -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 () => {
+10 -4
View File
@@ -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<RefreshMaterializedViewResult> {
validateNonNegativeInteger(options?.sourceVersion, "sourceVersion");
return await this.inner.refreshMaterializedView(options?.sourceVersion);
return await this.inner.refreshMaterializedView(
options?.sourceVersion,
options?.full,
);
}
}
+3 -1
View File
@@ -699,6 +699,7 @@ export abstract class Table {
*/
abstract refreshMaterializedView(
sourceVersion?: number,
full?: boolean,
): Promise<RefreshMaterializedViewResult>;
/** @ignore */
@@ -1442,8 +1443,9 @@ export class LocalTable extends Table {
async refreshMaterializedView(
sourceVersion?: number,
full?: boolean,
): Promise<RefreshMaterializedViewResult> {
return await this.inner.refreshMaterializedView(sourceVersion);
return await this.inner.refreshMaterializedView(sourceVersion, full);
}
async materializedViewDefinition(): Promise<string> {
+13 -10
View File
@@ -470,11 +470,12 @@ impl Table {
pub async fn refresh_materialized_view(
&self,
source_version: Option<i64>,
full: Option<bool>,
) -> napi::Result<RefreshMaterializedViewResult> {
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<lancedb::function::FunctionErrors> 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<String>,
/// Rows written by this refresh, or absent for remote SQL refreshes.
pub rows_written: Option<i64>,
/// Source version refreshed to, or absent for remote SQL refreshes.
pub source_version: Option<i64>,
pub version: i64,
}
impl From<lancedb::RefreshMaterializedViewResult> 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,
}
}
+5 -5
View File
@@ -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:
+21 -12
View File
@@ -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)
)
)
+112 -2
View File
@@ -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)
+18 -11
View File
@@ -694,9 +694,9 @@ impl From<lancedb::function::FunctionErrors> 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<String>,
pub rows_written: Option<u64>,
pub source_version: Option<u64>,
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<lancedb::RefreshMaterializedViewResult> 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<u64>,
full: bool,
) -> PyResult<Bound<'_, PyAny>> {
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<u64>,
full: bool,
) -> PyResult<Bound<'_, PyAny>> {
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);
}
+48 -14
View File
@@ -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<dyn std::error::Error>> {
/// 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<u64>,
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(),
+13 -10
View File
@@ -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<RefreshMode>,
/// Rows written by this refresh, or `None` when unavailable.
/// Remote SQL refreshes currently do not expose this metric.
pub rows_written: Option<u64>,
/// The source table version the view reflects, or `None` when unavailable.
/// Remote SQL refreshes currently do not expose this metric.
pub source_version: Option<u64>,
/// 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,
})
}
+50 -15
View File
@@ -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::<serde_json::Value>(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<String> {
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;
+9 -9
View File
@@ -2224,6 +2224,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
async fn refresh_materialized_view_async(
&self,
source_version: Option<u64>,
full: bool,
) -> Result<Job<RefreshMaterializedViewResult>> {
self.check_mutable().await?;
if source_version.is_some() {
@@ -2236,8 +2237,9 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
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<S: HttpSend> BaseTable for RemoteTable<S> {
);
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,
})
})
+1
View File
@@ -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<u64>,
_full: bool,
) -> Result<Job<crate::materialized_view::RefreshMaterializedViewResult>> {
Err(Error::NotSupported {
message: "remote materialized-view refresh is not supported on this table type".into(),