diff --git a/python/python/tests/test_read_consistency_interval.py b/python/python/tests/test_read_consistency_interval.py new file mode 100644 index 000000000..bc69775ea --- /dev/null +++ b/python/python/tests/test_read_consistency_interval.py @@ -0,0 +1,83 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +from datetime import timedelta + +import lancedb +import pytest +from lance_namespace import connect as connect_namespace_client +from lancedb import _lancedb +from lancedb.namespace import LanceNamespaceDBConnection + + +@pytest.mark.parametrize("seconds", [-1.0, -0.000001]) +@pytest.mark.parametrize("kind", ["sync", "namespace", "namespace_async", "client"]) +def test_negative_read_consistency_interval(tmp_path, seconds, kind): + interval = timedelta(seconds=seconds) + with pytest.raises(ValueError, match="read_consistency_interval"): + if kind == "sync": + lancedb.connect(tmp_path, read_consistency_interval=interval) + elif kind == "client": + client = connect_namespace_client("dir", {"root": str(tmp_path)}) + LanceNamespaceDBConnection(client, read_consistency_interval=interval) + else: + connect = ( + lancedb.connect_namespace + if kind == "namespace" + else lancedb.connect_namespace_async + ) + connect("dir", {"root": str(tmp_path)}, read_consistency_interval=interval) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("seconds", [-1.0, -0.000001]) +async def test_negative_async_read_consistency_interval(tmp_path, seconds): + with pytest.raises(ValueError, match="read_consistency_interval"): + await lancedb.connect_async( + tmp_path, read_consistency_interval=timedelta(seconds=seconds) + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("seconds", [float("nan"), float("inf"), -float("inf"), 1e30]) +@pytest.mark.parametrize("kind", ["connect", "namespace", "client"]) +async def test_binding_rejects_invalid_read_consistency_interval( + tmp_path, seconds, kind +): + with pytest.raises(ValueError, match="read_consistency_interval"): + if kind == "connect": + await _lancedb.connect(str(tmp_path), read_consistency_interval=seconds) + elif kind == "namespace": + _lancedb.connect_namespace( + "dir", {"root": str(tmp_path)}, read_consistency_interval=seconds + ) + else: + client = connect_namespace_client("dir", {"root": str(tmp_path)}) + _lancedb.connect_namespace_client(client, read_consistency_interval=seconds) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "interval", [None, timedelta(0), timedelta(microseconds=1), timedelta(seconds=5)] +) +async def test_valid_read_consistency_intervals(tmp_path, interval): + db = lancedb.connect(tmp_path / "sync", read_consistency_interval=interval) + assert db.read_consistency_interval == interval + db.close() + + db = await lancedb.connect_async( + tmp_path / "async", read_consistency_interval=interval + ) + assert await db.get_read_consistency_interval() == interval + db.close() + + db = lancedb.connect_namespace( + "dir", {"root": str(tmp_path / "namespace")}, read_consistency_interval=interval + ) + assert db.read_consistency_interval == interval + db.close() + + client = connect_namespace_client("dir", {"root": str(tmp_path / "client")}) + db = LanceNamespaceDBConnection(client, read_consistency_interval=interval) + assert db.read_consistency_interval == interval + db.close() diff --git a/python/src/connection.rs b/python/src/connection.rs index 7e20f5f99..baeb29780 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -49,6 +49,18 @@ impl Connection { } } +fn parse_read_consistency_interval(interval: Option) -> PyResult> { + interval + .map(|seconds| { + Duration::try_from_secs_f64(seconds).map_err(|_| { + PyValueError::new_err( + "read_consistency_interval must be finite, non-negative, and within the supported range", + ) + }) + }) + .transpose() +} + fn parse_namespace_client_pushdown_operations( operations: Option>, ) -> PyResult> { @@ -1077,6 +1089,7 @@ pub fn connect( namespace_client_properties: Option>, oauth_config: Option, ) -> PyResult> { + let read_consistency_interval = parse_read_consistency_interval(read_consistency_interval)?; future_into_py(py, async move { let mut builder = lancedb::connect(&uri); if let Some(api_key) = api_key { @@ -1095,7 +1108,6 @@ pub fn connect( #[cfg(not(feature = "remote"))] let _ = sql_host_override; if let Some(read_consistency_interval) = read_consistency_interval { - let read_consistency_interval = Duration::from_secs_f64(read_consistency_interval); builder = builder.read_consistency_interval(read_consistency_interval); } if let Some(storage_options) = storage_options { @@ -1144,7 +1156,7 @@ pub fn connect_namespace_client( namespace_client_impl: Option, namespace_client_properties: Option>, ) -> PyResult { - let read_consistency_interval = read_consistency_interval.map(Duration::from_secs_f64); + let read_consistency_interval = parse_read_consistency_interval(read_consistency_interval)?; let namespace_client_pushdown_operations = parse_namespace_client_pushdown_operations(namespace_client_pushdown_operations)?; let ns_properties = namespace_client_properties.unwrap_or_default(); @@ -1201,7 +1213,7 @@ pub fn connect_namespace( session: Option, namespace_client_pushdown_operations: Option>, ) -> PyResult { - let read_consistency_interval = read_consistency_interval.map(Duration::from_secs_f64); + let read_consistency_interval = parse_read_consistency_interval(read_consistency_interval)?; let namespace_client_pushdown_operations = parse_namespace_client_pushdown_operations(namespace_client_pushdown_operations)?;