Merge branch 'merge-insert-source-stats' into merge-insert-source-stats-bindings

python/src/table.rs

@@ -38,16 +38,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)?;
@@ -63,15 +63,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(
@@ -93,9 +93,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())
 }
 
@@ -269,17 +269,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),
         }
     }
 
@@ -293,7 +293,7 @@ impl LsmWriteSpec {
     #[staticmethod]
     pub fn identity(column: String) -> Self {
         Self {
-            inner: lancedb::table::LsmWriteSpec::identity(column),
+            inner: lancedb::table::MemWalWriteSpec::identity(column),
         }
     }
 
@@ -302,7 +302,7 @@ impl LsmWriteSpec {
     #[staticmethod]
     pub fn unsharded() -> Self {
         Self {
-            inner: lancedb::table::LsmWriteSpec::unsharded(),
+            inner: lancedb::table::MemWalWriteSpec::unsharded(),
         }
     }
 
@@ -326,33 +326,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,
             ),
@@ -364,9 +364,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",
         }
     }
 
@@ -374,9 +374,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,
         }
     }
 
@@ -384,7 +384,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,
         }
     }
@@ -402,14 +402,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 }
     }
 }
@@ -696,6 +696,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>> {
@@ -704,7 +712,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())
     }
 
@@ -712,10 +720,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()))
         })
     }
@@ -738,7 +743,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 {
@@ -751,7 +756,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> {
@@ -760,7 +765,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 {
@@ -777,17 +782,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())
     }
 }
@@ -1461,14 +1488,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
@@ -1591,8 +1634,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);
@@ -1620,29 +1663,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))
         })
     }
 
@@ -1661,55 +1704,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()
         })
     }
 
@@ -2041,7 +2083,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>,
     source_collect_threshold_bytes: Option<usize>,
 }