rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-18 17:52:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/lancedb/lancedb into merge-insert-source-stats

Browse Source 
# Conflicts:
#	rust/lancedb/src/table/computed_columns.rs
#	rust/lancedb/src/table/refresh.rs
This commit is contained in:
Will Jones
2026-10-16 21:18:02 +00:00
parent 300d494d6e 5f746d1e01
commit e7d060dca8
312 changed files with 21498 additions and 7261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/src/table.rs
+128 -86
View File
@@ -39,16 +39,16 @@ use pyo3::{
mod scannable;
/// Convert `LsmStats` to a Python dict, preserving the per-bucket list.
/// Convert `MemWalStats` to a Python dict, preserving the per-table-shard list.
///
/// Deliberately not flattened to a table-level summary: a table is N
/// buckets on one node, and the per-bucket detail is the reason the
/// endpoint exists — flattening hides the single hot bucket someone opened
/// table shards on one node, and the per-shard detail is the reason the
/// endpoint exists — flattening hides the single hot table shard someone opened
/// it to find.
fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult<Py<PyDict>> {
fn memwal_stats_to_py(py: Python<'_>, stats: &lancedb::table::MemWalStats) -> PyResult<Py<PyDict>> {
let out = PyDict::new(py);
let buckets = PyList::empty(py);
for b in &stats.buckets {
let table_shards = PyList::empty(py);
for b in &stats.table_shards {
let e = PyDict::new(py);
e.set_item("shard_id", &b.shard_id)?;
e.set_item("status", &b.status)?;
@@ -64,15 +64,15 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
b.wal_entry_position_last_seen,
)?;
let generations = PyList::empty(py);
for g in &b.generations {
let sstables = PyList::empty(py);
for g in &b.sstables {
let ge = PyDict::new(py);
ge.set_item("generation", g.generation)?;
ge.set_item("bytes", g.bytes)?;
ge.set_item("rows", g.rows)?;
generations.append(ge)?;
sstables.append(ge)?;
}
e.set_item("generations", generations)?;
e.set_item("sstables", sstables)?;
e.set_item("compacting", b.compacting)?;
e.set_item(
@@ -94,9 +94,9 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
})
.transpose()?,
)?;
buckets.append(e)?;
table_shards.append(e)?;
}
out.set_item("buckets", buckets)?;
out.set_item("table_shards", table_shards)?;
Ok(out.unbind())
}
@@ -270,17 +270,17 @@ fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
/// table has, including ones created later.
#[pyclass(module = "lancedb._lancedb", from_py_object)]
#[derive(Clone, Debug)]
pub struct LsmWriteSpec {
inner: lancedb::table::LsmWriteSpec,
pub struct MemWalWriteSpec {
inner: lancedb::table::MemWalWriteSpec,
}
#[pymethods]
impl LsmWriteSpec {
impl MemWalWriteSpec {
/// Hash-bucket sharding by the unenforced primary key column.
#[staticmethod]
pub fn bucket(column: String, num_buckets: u32) -> Self {
Self {
inner: lancedb::table::LsmWriteSpec::bucket(column, num_buckets),
inner: lancedb::table::MemWalWriteSpec::bucket(column, num_buckets),
}
}
@@ -294,7 +294,7 @@ impl LsmWriteSpec {
#[staticmethod]
pub fn identity(column: String) -> Self {
Self {
inner: lancedb::table::LsmWriteSpec::identity(column),
inner: lancedb::table::MemWalWriteSpec::identity(column),
}
}
@@ -303,7 +303,7 @@ impl LsmWriteSpec {
#[staticmethod]
pub fn unsharded() -> Self {
Self {
inner: lancedb::table::LsmWriteSpec::unsharded(),
inner: lancedb::table::MemWalWriteSpec::unsharded(),
}
}
@@ -327,33 +327,33 @@ impl LsmWriteSpec {
pub fn __repr__(&self) -> String {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket {
lancedb::table::MemWalWriteSpec::Bucket {
column,
num_buckets,
maintained_indexes,
writer_config_defaults,
} => format!(
"LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
"MemWalWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
column,
num_buckets,
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
lancedb::table::LsmWriteSpec::Identity {
lancedb::table::MemWalWriteSpec::Identity {
column,
maintained_indexes,
writer_config_defaults,
} => format!(
"LsmWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
"MemWalWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
column,
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
lancedb::table::LsmWriteSpec::Unsharded {
lancedb::table::MemWalWriteSpec::Unsharded {
maintained_indexes,
writer_config_defaults,
} => format!(
"LsmWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
"MemWalWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
@@ -365,9 +365,9 @@ impl LsmWriteSpec {
#[getter]
pub fn spec_type(&self) -> &'static str {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket { .. } => "bucket",
lancedb::table::LsmWriteSpec::Identity { .. } => "identity",
lancedb::table::LsmWriteSpec::Unsharded { .. } => "unsharded",
lancedb::table::MemWalWriteSpec::Bucket { .. } => "bucket",
lancedb::table::MemWalWriteSpec::Identity { .. } => "identity",
lancedb::table::MemWalWriteSpec::Unsharded { .. } => "unsharded",
}
}
@@ -375,9 +375,9 @@ impl LsmWriteSpec {
#[getter]
pub fn column(&self) -> Option<String> {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket { column, .. }
| lancedb::table::LsmWriteSpec::Identity { column, .. } => Some(column.clone()),
lancedb::table::LsmWriteSpec::Unsharded { .. } => None,
lancedb::table::MemWalWriteSpec::Bucket { column, .. }
| lancedb::table::MemWalWriteSpec::Identity { column, .. } => Some(column.clone()),
lancedb::table::MemWalWriteSpec::Unsharded { .. } => None,
}
}
@@ -385,7 +385,7 @@ impl LsmWriteSpec {
#[getter]
pub fn num_buckets(&self) -> Option<u32> {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket { num_buckets, .. } => Some(*num_buckets),
lancedb::table::MemWalWriteSpec::Bucket { num_buckets, .. } => Some(*num_buckets),
_ => None,
}
}
@@ -403,14 +403,14 @@ impl LsmWriteSpec {
}
}
impl From<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
fn from(spec: LsmWriteSpec) -> Self {
impl From<MemWalWriteSpec> for lancedb::table::MemWalWriteSpec {
fn from(spec: MemWalWriteSpec) -> Self {
spec.inner
}
}
impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
fn from(inner: lancedb::table::LsmWriteSpec) -> Self {
impl From<lancedb::table::MemWalWriteSpec> for MemWalWriteSpec {
fn from(inner: lancedb::table::MemWalWriteSpec) -> Self {
Self { inner }
}
}
@@ -697,6 +697,14 @@ impl PyBlobFile {
}
}
fn blob_file_error(operation: &str, error: lance_core::Error) -> pyo3::PyErr {
let message = format!("blob {operation} failed: {error}");
match error {
lance_core::Error::InvalidInput { .. } => PyValueError::new_err(message),
_ => PyRuntimeError::new_err(message),
}
}
#[pymethods]
impl PyBlobFile {
fn read_bytes(self_: PyRef<'_, Self>) -> PyResult<Py<PyBytes>> {
@@ -705,7 +713,7 @@ impl PyBlobFile {
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
.map_err(|e| blob_file_error("read", e))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
@@ -713,10 +721,7 @@ impl PyBlobFile {
self_.ensure_open()?;
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
let bytes = inner
.read()
.await
.map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
let bytes = inner.read().await.map_err(|e| blob_file_error("read", e))?;
Python::attach(|py| Ok(PyBytes::new(py, bytes.as_ref()).unbind()))
})
}
@@ -739,7 +744,7 @@ impl PyBlobFile {
self_
.py()
.detach(move || block_on(async move { inner.close().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob close failed: {e}")))
.map_err(|e| blob_file_error("close", e))
}
fn is_closed(self_: PyRef<'_, Self>) -> bool {
@@ -752,7 +757,7 @@ impl PyBlobFile {
self_
.py()
.detach(move || block_on(async move { inner.seek(position).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob seek failed: {e}")))
.map_err(|e| blob_file_error("seek", e))
}
fn tell(self_: PyRef<'_, Self>) -> PyResult<u64> {
@@ -761,7 +766,7 @@ impl PyBlobFile {
self_
.py()
.detach(move || block_on(async move { inner.tell().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob tell failed: {e}")))
.map_err(|e| blob_file_error("tell", e))
}
fn size(self_: PyRef<'_, Self>) -> u64 {
@@ -778,17 +783,39 @@ impl PyBlobFile {
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_range(offset..end).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob read_range failed: {e}")))?;
.map_err(|e| blob_file_error("read_range", e))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
/// Read blob-local byte ranges without moving the cursor.
fn read_ranges(self_: PyRef<'_, Self>, ranges: Vec<(u64, u64)>) -> PyResult<Vec<Py<PyBytes>>> {
let ranges = ranges
.into_iter()
.map(|(offset, length)| {
offset
.checked_add(length)
.map(|end| offset..end)
.ok_or_else(|| PyValueError::new_err("offset + length overflowed"))
})
.collect::<PyResult<Vec<_>>>()?;
let inner = self_.inner.clone();
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_ranges(&ranges).await }))
.map_err(|e| blob_file_error("read_ranges", e))?;
Ok(bytes
.iter()
.map(|bytes| PyBytes::new(py, bytes.as_ref()).unbind())
.collect())
}
fn read_up_to(self_: PyRef<'_, Self>, length: usize) -> PyResult<Py<PyBytes>> {
self_.ensure_open()?;
let inner = self_.inner.clone();
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_up_to(length).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob read_up_to failed: {e}")))?;
.map_err(|e| blob_file_error("read_up_to", e))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
}
@@ -1462,14 +1489,30 @@ impl Table {
pub fn fetch_blob_ranges(
self_: PyRef<'_, Self>,
column: String,
requests: Vec<(u64, u64, u64)>,
requests: Vec<Vec<i128>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
let requests = requests
.into_iter()
.map(|values| {
let [row_id, offset, length] = <[i128; 3]>::try_from(values).map_err(|_| {
PyValueError::new_err("each blob range request must have three values")
})?;
let to_u64 = |value, name| {
u64::try_from(value).map_err(|_| {
PyValueError::new_err(format!(
"blob range {name} must be a non-negative 64-bit integer"
))
})
};
Ok(BlobRangeRequest::new(
to_u64(row_id, "row id")?,
to_u64(offset, "offset")?,
to_u64(length, "length")?,
))
})
.collect::<PyResult<Vec<_>>>()?;
future_into_py(self_.py(), async move {
let requests = requests
.into_iter()
.map(|(row_id, offset, length)| BlobRangeRequest::new(row_id, offset, length))
.collect::<Vec<_>>();
let blobs: LargeBinaryArray = inner
.fetch_blob_ranges(column, requests)
.await
@@ -1593,8 +1636,8 @@ impl Table {
if let Some(use_index) = parameters.use_index {
builder.use_index(use_index);
}
if let Some(use_lsm) = parameters.use_lsm {
builder.use_lsm(use_lsm);
if let Some(use_memwal) = parameters.use_memwal {
builder.use_memwal(use_memwal);
}
if let Some(validate_single_shard) = parameters.validate_single_shard {
builder.validate_single_shard(validate_single_shard);
@@ -1622,29 +1665,29 @@ impl Table {
})
}
pub fn set_lsm_write_spec<'a>(
pub fn enable_memwal<'a>(
self_: PyRef<'a, Self>,
spec: LsmWriteSpec,
spec: MemWalWriteSpec,
) -> PyResult<Bound<'a, PyAny>> {
let inner = self_.inner_ref()?.clone();
let native_spec = lancedb::table::LsmWriteSpec::from(spec);
let native_spec = lancedb::table::MemWalWriteSpec::from(spec);
future_into_py(self_.py(), async move {
inner.set_lsm_write_spec(native_spec).await.infer_error()
inner.enable_memwal(native_spec).await.infer_error()
})
}
pub fn unset_lsm_write_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn disable_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.unset_lsm_write_spec().await.infer_error()
inner.disable_memwal().await.infer_error()
})
}
pub fn get_lsm_write_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn get_memwal_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let spec = inner.get_lsm_write_spec().await.infer_error()?;
Ok(spec.map(LsmWriteSpec::from))
let spec = inner.get_memwal_spec().await.infer_error()?;
Ok(spec.map(MemWalWriteSpec::from))
})
}
@@ -1663,55 +1706,54 @@ impl Table {
Ok(())
}
/// Converge the table's LSM write path into its base table.
/// Converge the table's MemWAL write path into its base table.
///
/// Best-effort: with writes flowing, new rows may land after the last
/// pass. Errors if the table stops making progress.
pub fn checkpoint_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn checkpoint_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.checkpoint_lsm().await.infer_error()
inner.checkpoint_memwal().await.infer_error()
})
}
/// Seal every bucket's active memtable into L0.
pub fn flush_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(
self_.py(),
async move { inner.flush_lsm().await.infer_error() },
)
}
/// Trigger a background L0 → base pass per bucket. Returns once the
/// passes are dispatched, not once they finish — watch `get_lsm_stats`.
pub fn compact_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
/// Freeze every table shard's active memtable into an SSTable.
pub fn flush_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.compact_lsm().await.infer_error()
inner.flush_memwal().await.infer_error()
})
}
/// Live LSM state, or `None` when the LSM write path is not enabled.
#[pyo3(signature = (include_generation_rows=false))]
pub fn get_lsm_stats(
/// Trigger a background SSTable compaction pass per table shard. Returns once the
/// passes are dispatched, not once they finish — watch `get_memwal_stats`.
pub fn compact_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.compact_memwal().await.infer_error()
})
}
/// Live MemWAL state, or `None` when the MemWAL write path is not enabled.
#[pyo3(signature = (include_sstable_rows=false))]
pub fn get_memwal_stats(
self_: PyRef<'_, Self>,
include_generation_rows: bool,
include_sstable_rows: bool,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let stats = inner
.get_lsm_stats(include_generation_rows)
.get_memwal_stats(include_sstable_rows)
.await
.infer_error()?;
Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
Python::attach(|py| stats.map(|s| memwal_stats_to_py(py, &s)).transpose())
})
}
pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn close_memwal_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.close_lsm_writers().await.infer_error()
inner.close_memwal_writers().await.infer_error()
})
}
@@ -2043,7 +2085,7 @@ pub struct MergeInsertParams {
when_not_matched_by_source_condition_expr: Option<PyExpr>,
timeout: Option<std::time::Duration>,
use_index: Option<bool>,
use_lsm: Option<bool>,
use_memwal: Option<bool>,
validate_single_shard: Option<bool>,
}