Merge remote-tracking branch 'origin/main' into feature/wal-support-all-sdk

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.38.0-beta.0"
+version = "0.38.0-beta.1"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"

python/python/lancedb/query.py

@@ -2235,6 +2235,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
             reranker=self._reranker,
             limit=self._limit,
             with_row_ids=True,
+            offset=self._offset,
         )
         return self._finish_hybrid_results(results)
 
@@ -2256,6 +2257,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
         reranker,
         limit: int,
         with_row_ids: bool,
+        offset: Optional[int] = None,
     ) -> pa.Table:
         if norm == "rank":
             vector_results = LanceHybridQueryBuilder._rank(vector_results, "_distance")
@@ -2332,7 +2334,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
             score_i = results.column_names.index("_score")
             results = results.set_column(score_i, "_score", original_scores)
 
-        results = results.slice(length=limit)
+        results = results.slice(offset=offset or 0, length=limit)
 
         if not with_row_ids:
             results = results.drop(["_rowid"])
@@ -2679,8 +2681,12 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
 
         # Apply common configurations
         if self._limit:
-            self._vector_query.limit(self._limit)
-            self._fts_query.limit(self._limit)
+            # The final offset/limit window is sliced out of the combined,
+            # reranked results, so each sub-query must fetch enough rows to
+            # cover the skipped prefix as well as the window itself.
+            sub_query_limit = self._limit + (self._offset or 0)
+            self._vector_query.limit(sub_query_limit)
+            self._fts_query.limit(sub_query_limit)
         if self._columns:
             self._vector_query.select(self._columns)
             self._fts_query.select(self._columns)

python/python/lancedb/remote/table.py

@@ -990,17 +990,39 @@ class RemoteTable(Table):
         return LOOP.run(self._table.set_unenforced_primary_key(columns))
 
     def set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:
-        """Not supported on LanceDB Cloud."""
+        """Install an LsmWriteSpec."""
         return LOOP.run(self._table.set_lsm_write_spec(spec))
 
     def unset_lsm_write_spec(self) -> None:
-        """Not supported on LanceDB Cloud."""
+        """Remove the LsmWriteSpec."""
         return LOOP.run(self._table.unset_lsm_write_spec())
 
     def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
         """Read the installed LsmWriteSpec, or ``None``."""
         return LOOP.run(self._table.get_lsm_write_spec())
 
+    def checkpoint_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
+        return LOOP.run(self._table.checkpoint_lsm())
+
+    def flush_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
+        return LOOP.run(self._table.flush_lsm())
+
+    def compact_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
+        return LOOP.run(self._table.compact_lsm())
+
+    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
+        """Synchronous version of
+        [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
+        return LOOP.run(
+            self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
+        )
+
     def close_lsm_writers(self) -> None:
         """No-op on LanceDB Cloud (no local shard writers)."""
         return LOOP.run(self._table.close_lsm_writers())

python/python/lancedb/table.py

@@ -4846,7 +4846,7 @@ class AsyncTable:
         ``asyncio.wait_for`` for a wall-clock bound; abandoning it partway
         costs nothing.
         """
-        return await self._inner.checkpoint_lsm()
+        await self._inner.checkpoint_lsm()
 
     async def flush_lsm(self) -> None:
         """Seal every bucket's active memtable into L0.
@@ -4855,7 +4855,7 @@ class AsyncTable:
         `compact_lsm`. On a node that has not claimed this table, this claims
         it and replays its WAL log first.
         """
-        return await self._inner.flush_lsm()
+        await self._inner.flush_lsm()
 
     async def compact_lsm(self) -> None:
         """Trigger a background L0 to base compaction pass per bucket.
@@ -4864,7 +4864,7 @@ class AsyncTable:
         ``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop
         until the current L0 has reached base.
         """
-        return await self._inner.compact_lsm()
+        await self._inner.compact_lsm()
 
     async def get_lsm_stats(
         self, *, include_generation_rows: bool = False

python/python/tests/test_expr.py

@@ -632,3 +632,101 @@ class TestExprBytesIntegration:
             .to_arrow()
         )
         assert result.num_rows == 2
+
+
+# ── datetime / timezone integration for lit() (issue #3262) ──────────────────
+
+
+class TestExprDatetimeTimezoneIntegration:
+    """Integration coverage for lit(datetime) against table timestamp columns.
+
+    PyArrow stores naive timestamps as UTC wall-clock microseconds. Python's
+    datetime.timestamp() treats naive values as *local* time, which used to
+    shift lit(naive) by the host UTC offset and break equality filters on
+    non-UTC machines. These cases lock the expected semantics.
+    """
+
+    def test_both_naive_match(self, tmp_path):
+        """Table naive + lit naive with the same wall clock must match."""
+        db = lancedb.connect(str(tmp_path / "naive"))
+        ts = datetime(2024, 7, 1, 10, 0, 0)
+        table = db.create_table(
+            "t", [{"id": 1, "ts": ts}, {"id": 2, "ts": datetime(2024, 7, 2, 10, 0, 0)}]
+        )
+        result = table.search().where(col("ts") == lit(ts)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_both_same_timezone_match(self, tmp_path):
+        """Table UTC + lit UTC for the same instant must match."""
+        db = lancedb.connect(str(tmp_path / "utc"))
+        ts = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        table = db.create_table(
+            "t",
+            pa.table(
+                {
+                    "id": [1, 2],
+                    "ts": pa.array(
+                        [ts, datetime(2024, 7, 2, 10, 0, 0, tzinfo=timezone.utc)],
+                        type=pa.timestamp("us", tz="UTC"),
+                    ),
+                }
+            ),
+        )
+        result = table.search().where(col("ts") == lit(ts)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_different_timezones_same_instant(self, tmp_path):
+        """UTC table row equals lit of the same instant in a different zone."""
+        db = lancedb.connect(str(tmp_path / "diff_tz"))
+        ts_utc = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        # Same instant as 06:00 in UTC-4
+        ts_est = datetime(2024, 7, 1, 6, 0, 0, tzinfo=timezone(timedelta(hours=-4)))
+        table = db.create_table(
+            "t",
+            pa.table(
+                {
+                    "id": [1],
+                    "ts": pa.array([ts_utc], type=pa.timestamp("us", tz="UTC")),
+                }
+            ),
+        )
+        result = table.search().where(col("ts") == lit(ts_est)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_table_tz_literal_naive(self, tmp_path):
+        """UTC table + naive lit uses wall-clock equality (10:00 == 10:00 UTC)."""
+        db = lancedb.connect(str(tmp_path / "tz_naive"))
+        ts_utc = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        ts_naive = datetime(2024, 7, 1, 10, 0, 0)
+        table = db.create_table(
+            "t",
+            pa.table(
+                {
+                    "id": [1],
+                    "ts": pa.array([ts_utc], type=pa.timestamp("us", tz="UTC")),
+                }
+            ),
+        )
+        result = table.search().where(col("ts") == lit(ts_naive)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_table_naive_literal_aware(self, tmp_path):
+        """Naive table + UTC lit with the same wall clock must match."""
+        db = lancedb.connect(str(tmp_path / "naive_aware"))
+        ts_naive = datetime(2024, 7, 1, 10, 0, 0)
+        ts_utc = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        table = db.create_table("t", [{"id": 1, "ts": ts_naive}])
+        result = table.search().where(col("ts") == lit(ts_utc)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_naive_lit_sql_is_wall_clock_not_local_shifted(self):
+        """Regression: naive lit must not apply the host local UTC offset."""
+        ts = datetime(2024, 7, 1, 10, 0, 0)
+        sql = lit(ts).to_sql()
+        # Must encode 10:00 wall clock, not 10:00+local_offset.
+        assert "2024-07-01 10:00:00" in sql

python/python/tests/test_hybrid_query.py

@@ -203,6 +203,31 @@ async def test_async_hybrid_query_default_limit(table: AsyncTable):
     assert texts.count("a") == 1
 
 
+def test_hybrid_query_offset(sync_table: Table):
+    # The offset window of a hybrid query must be a suffix of the same query
+    # run without an offset -- it must not be silently ignored.
+    full = (
+        sync_table.search(query_type="hybrid")
+        .vector([0.0, 0.4])
+        .text("dog")
+        .limit(4)
+        .with_row_id(True)
+        .to_arrow()
+    )
+    assert len(full) == 4
+
+    offset_result = (
+        sync_table.search(query_type="hybrid")
+        .vector([0.0, 0.4])
+        .text("dog")
+        .offset(2)
+        .limit(2)
+        .with_row_id(True)
+        .to_arrow()
+    )
+    assert offset_result["_rowid"].to_pylist() == full["_rowid"].to_pylist()[2:]
+
+
 def test_hybrid_query_minimum_nprobes_zero_raises(sync_table: Table):
     # minimum_nprobes(0) must raise the same validation error a plain vector
     # query raises, not silently no-op because 0 is falsy.

python/python/tests/test_remote_db.py

@@ -1133,6 +1133,131 @@ def test_stats():
         assert res == stats
 
 
+@contextlib.contextmanager
+def lsm_test_table(lsm_handler):
+    """A remote table whose LSM routes are served by ``lsm_handler``.
+
+    ``lsm_handler(request, route)`` is called for ``/v1/table/test/<route>/``
+    where route is one of flush_lsm, compact_lsm, get_lsm_stats, and is
+    responsible for writing the response.
+    """
+    routes = ("flush_lsm", "compact_lsm", "get_lsm_stats")
+
+    def handler(request):
+        match = re.fullmatch(r"/v1/table/test/(\w+)/", request.path)
+        route = match.group(1) if match else None
+        if route in routes:
+            lsm_handler(request, route)
+        elif route == "describe":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        yield db.open_table("test")
+
+
+def read_json_body(request):
+    content_len = int(request.headers.get("Content-Length"))
+    return json.loads(request.rfile.read(content_len))
+
+
+def send_json(request, payload, status=200):
+    request.send_response(status)
+    request.send_header("Content-Type", "application/json")
+    request.end_headers()
+    request.wfile.write(json.dumps(payload).encode())
+
+
+def test_get_lsm_stats_sync():
+    """The sync wrapper round-trips the server payload into a dict."""
+    bucket = {
+        "shard_id": "b0",
+        "status": "Active",
+        "writer_epoch": 3,
+        "manifest_version": 12,
+        "current_generation": 6,
+        "replay_after_wal_entry_position": 40,
+        "wal_entry_position_last_seen": 42,
+        "generations": [{"generation": 5, "bytes": 1024, "rows": 7}],
+        "compacting": False,
+        "memtables": [
+            {
+                "generation": 6,
+                "rows": 2,
+                "bytes": 64,
+                "batches": 1,
+                "indexes": ["vec_idx"],
+            }
+        ],
+    }
+    seen_bodies = []
+
+    def lsm_handler(request, route):
+        assert route == "get_lsm_stats"
+        seen_bodies.append(read_json_body(request))
+        send_json(request, {"lsm_stats": {"buckets": [bucket]}})
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.get_lsm_stats() == {"buckets": [bucket]}
+        # Off by default, and forwarded when asked for.
+        assert seen_bodies == [{"include_generation_rows": False}]
+        table.get_lsm_stats(include_generation_rows=True)
+        assert seen_bodies[-1] == {"include_generation_rows": True}
+
+
+def test_get_lsm_stats_sync_returns_none_when_lsm_disabled():
+    """A null envelope means the LSM write path is not enabled, not an error."""
+
+    def lsm_handler(request, route):
+        send_json(request, {"lsm_stats": None})
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.get_lsm_stats() is None
+
+
+def test_flush_and_compact_lsm_sync():
+    """Both are one-shot POSTs answered 202 with no body."""
+    called = []
+
+    def lsm_handler(request, route):
+        called.append(route)
+        request.send_response(202)
+        request.end_headers()
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.flush_lsm() is None
+        assert table.compact_lsm() is None
+        assert called == ["flush_lsm", "compact_lsm"]
+
+
+def test_checkpoint_lsm_sync():
+    """Seal, read the watermark, and return once L0 holds nothing.
+
+    The convergence loop itself is covered in Rust; this pins the sync
+    binding to the endpoints it drives.
+    """
+    called = []
+
+    def lsm_handler(request, route):
+        called.append(route)
+        if route == "get_lsm_stats":
+            # An empty L0 yields no target watermark, so the loop is done
+            # after the seal without ever polling compaction.
+            send_json(request, {"lsm_stats": {"buckets": []}})
+        else:
+            request.send_response(202)
+            request.end_headers()
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.checkpoint_lsm() is None
+        assert called == ["flush_lsm", "get_lsm_stats"]
+
+
 @contextlib.contextmanager
 def query_test_table(query_handler, *, server_version=Version("0.1.0")):
     def handler(request):

python/src/expr.rs

@@ -191,8 +191,27 @@ pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
     }
 
     // datetime.datetime is a subclass of datetime.date, so it must be checked first.
+    //
+    // Python's datetime.timestamp() treats *naive* datetimes as local wall time.
+    // PyArrow (and therefore Lance table storage) encodes naive timestamps as
+    // UTC wall-clock microseconds. Using .timestamp() for naive values therefore
+    // shifts the literal by the local UTC offset on non-UTC machines, so
+    // `col("ts") == lit(naive_dt)` fails against a table that holds the same
+    // naive value. Fix: treat naive datetimes as UTC wall clock (match Arrow);
+    // keep aware datetimes on the real .timestamp() path (correct epoch).
     if let Ok(dt) = value.cast::<PyDateTime>() {
-        let ts: f64 = dt.call_method0("timestamp")?.extract()?;
+        let ts: f64 = if dt.getattr("tzinfo")?.is_none() {
+            // Force UTC interpretation of the naive wall clock.
+            let utc = pyo3::types::PyModule::import(value.py(), "datetime")?
+                .getattr("timezone")?
+                .getattr("utc")?;
+            let kwargs = pyo3::types::PyDict::new(value.py());
+            kwargs.set_item("tzinfo", utc)?;
+            let aware = dt.call_method("replace", (), Some(&kwargs))?;
+            aware.call_method0("timestamp")?.extract()?
+        } else {
+            dt.call_method0("timestamp")?.extract()?
+        };
         let micros = (ts * 1_000_000.0).round() as i64;
         return Ok(PyExpr(df_lit(ScalarValue::TimestampMicrosecond(
             Some(micros),