Compare commits

..

1 Commits

Author SHA1 Message Date
Lance Release eef4f5c6ac Bump version: 0.32.0-beta.0 → 0.32.0-beta.1 2026-07-10 16:13:17 +00:00
143 changed files with 1265 additions and 10215 deletions
-20
View File
@@ -1,20 +0,0 @@
{
"name": "lancedb",
"interface": {
"displayName": "LanceDB"
},
"plugins": [
{
"name": "lancedb",
"source": {
"source": "local",
"path": "./plugins/lancedb"
},
"policy": {
"installation": "AVAILABLE",
"authentication": "ON_INSTALL"
},
"category": "Developer Tools"
}
]
}
-4
View File
@@ -5,7 +5,3 @@ This directory contains repo-scoped code agent skills for the LanceDB project.
Each skill is a folder that contains a required `SKILL.md` and optional bundled resources.
Codex discovers skills from `.agents/skills` in the current working directory and parent directories.
The `lancedb` skill lives in the `plugins/lancedb` plugin (see `plugins/lancedb/skills/lancedb`)
so it can be installed via the plugin marketplaces (`.claude-plugin/marketplace.json` and
`.agents/plugins/marketplace.json`); the `lancedb` entry here is a symlink into that plugin.
-1
View File
@@ -1 +0,0 @@
../../plugins/lancedb/skills/lancedb
+137
View File
@@ -0,0 +1,137 @@
---
name: lancedb-branch-ops
description: Branch management for LanceDB tables via the REST API. Use this skill whenever someone wants to create, delete, list, or switch branches on a LanceDB table — or needs to make sure a write (metadata update, index build, etc.) lands on a specific branch instead of main. Invoke it even without the word "branch" if context makes clear they want an experimental copy of a table, want to isolate changes, or want to confirm a mutation didn't touch main. Covers: branches/list, branches/create, branches/delete, and passing "branch" in describe/update_field_metadata/create_index to target a non-main version.
---
## Goal
Manage branches on a LanceDB table: list what exists, create new ones, delete stale ones, and direct read/write operations at a specific branch without touching main.
## Step 0: Establish the connection
Use the `lancedb-connect` skill to resolve the base URL and auth headers (`x-api-key`, `x-lancedb-database`). Skip this only if the connection is already known from the current conversation.
All examples below use `{base_url}` — substitute the resolved endpoint and include the auth headers on every request.
## The branch model (important)
LanceDB branches are named snapshots that diverge from the table's current state at creation time. There is **no checkout command** — you never switch the whole table to a branch. Instead, you **pass `"branch": "<name>"` in the request body** of any operation to target that branch. Omitting the key (or sending an empty body) always targets main.
`branches/list` returns only non-main branches. Main always exists and is not listed.
## List branches
```http
POST {base_url}/v1/table/{table_id}/branches/list
Content-Type: application/json
{}
```
Response:
```json
{
"branches": {
"experiment-reindex": {"parentVersion": 1, "createAt": 1782506085, "manifestSize": 1029}
}
}
```
If `branches` is `{}`, the table has no branches besides main.
## Create a branch
```http
POST {base_url}/v1/table/{table_id}/branches/create
Content-Type: application/json
{"name": "experiment-reindex"}
```
HTTP 200 with `{}` body = success. The branch is created off the table's current state on main.
Verify by calling `branches/list` and confirming the new name appears.
## Delete a branch
```http
POST {base_url}/v1/table/{table_id}/branches/delete
Content-Type: application/json
{"name": "stale-2024"}
```
HTTP 200 with `{}` body = success. Only the branch pointer is removed — main and all row data remain intact.
Verify by calling `branches/list` (name gone) and `describe` with no branch param (main still responds).
## Operate on a specific branch
Pass `"branch": "<name>"` in the body of any operation to scope it to that branch:
**Read schema on a branch:**
```http
POST {base_url}/v1/table/{table_id}/describe
Content-Type: application/json
{"branch": "wip-branch"}
```
**Write metadata to a branch (not main):**
```http
POST {base_url}/v1/table/{table_id}/update_field_metadata
Content-Type: application/json
{
"branch": "wip-branch",
"updates": [
{
"path": "category",
"metadata": {"lancedb:description": "Product category label."},
"replace": false
}
]
}
```
**Build an index on a branch:**
```http
POST {base_url}/v1/table/{table_id}/create_index
Content-Type: application/json
{
"branch": "wip-branch",
"column": "category",
"index_type": "BTREE"
}
```
## Verifying isolation
After writing to a branch, always confirm the change did NOT land on main:
```bash
# Should show the new metadata
curl -s -X POST {base_url}/v1/table/{table_id}/describe \
-H "x-api-key: <key>" -H "x-lancedb-database: <db>" \
-H "content-type: application/json" \
-d '{"branch": "wip-branch"}'
# Should NOT show the new metadata
curl -s -X POST {base_url}/v1/table/{table_id}/describe \
-H "x-api-key: <key>" -H "x-lancedb-database: <db>" \
-H "content-type: application/json" \
-d '{}'
```
## Quick reference
| Goal | Endpoint | Body |
|------|----------|------|
| List all branches | `branches/list` | `{}` |
| Create a branch | `branches/create` | `{"name": "..."}` |
| Delete a branch | `branches/delete` | `{"name": "..."}` |
| Read schema on branch | `describe` | `{"branch": "..."}` |
| Write metadata on branch | `update_field_metadata` | `{"branch": "...", "updates": [...]}` |
| Build index on branch | `create_index` | `{"branch": "...", "column": ..., "index_type": ...}` |
| Target main (default) | any endpoint | omit `"branch"` key |
@@ -0,0 +1,178 @@
---
name: lancedb-column-metadata
description: Column metadata authoring for LanceDB tables via the REST API. This skill is required for tasks like writing field descriptions, setting tags on columns (field_type, model, project_id, version), classifying columns as embeddings vs labels vs eval metrics, or grouping versioned columns into logical families — because it has the API integration needed to read the schema and persist metadata back. Invoke whenever someone wants to document, annotate, tag, or classify what their table columns ARE. Trigger even without an explicit "LanceDB" mention, as long as the context is column-level documentation or tagging for an ML or vector database table.
metadata:
short-description: Write column descriptions, tags, and logical groupings to a LanceDB table
---
## Overview
This skill authors column-level metadata for a LanceDB table. It connects to a LanceDB deployment over its REST API, inspects the table schema, generates appropriate metadata, and writes it back.
## Step 0: Establish the connection
Use the `lancedb-connect` skill (invoke it via the Skill tool) to resolve the base URL and auth headers (`x-api-key`, `x-lancedb-database`) for whichever deployment the user is working against — enterprise/self-hosted or a local dev server. Skip it only if the connection details are already established in the conversation.
All examples below use `{base_url}` — substitute the resolved endpoint and include the resolved headers on every request.
## Metadata keys
All metadata uses namespaced keys:
| Key | Purpose | Example value |
|-----|---------|---------------|
| `lancedb:description` | Human-readable explanation of what the column contains | `"CLIP ViT-L/14 image embedding, L2-normalized (768-dim)"` |
| `lancedb:tag:<name>` | Flexible key-value tag; the suffix names the tag category | `lancedb:tag:field_type: "embedding"`, `lancedb:tag:model: "clip"`, `lancedb:tag:project_id: "foo"` |
| `lancedb:logical-column` | Logical group/family this column belongs to | `"clip_features"` |
Tags are open-ended — use whatever key suffix and value make sense given the user's intent. The tag suffix should describe *what is being classified* (e.g., `field_type`, `model`, `project_id`) and the value describes *how*.
## Step 1: Resolve the table identifier
You need:
- **Table name** (required) — e.g., `my_table` or `my_namespace.my_table`
- **Database name** — ask if not provided and not inferable from context; it goes in the `x-lancedb-database` header, never in the URL path
The table identifier in the URL path is typically `table_name` for a top-level table, or `namespace$table_name` if the table lives in a namespace. The API accepts a `delimiter` query parameter to parse compound identifiers (default `$`).
## Step 2: Describe the table
```http
POST {base_url}/v1/table/{table_id}/describe
Content-Type: application/json
{}
```
The response contains `schema.fields` — an array of field objects:
```json
{
"schema": {
"fields": [
{
"name": "clip_embedding_v3",
"type": { "type": "FixedSizeList", "fields": [...], "listSize": 768 },
"nullable": true,
"metadata": { "lancedb:description": "..." }
}
]
}
}
```
Each field has:
- `name` — field name
- `type` — Arrow data type (check `type.type` for the type string)
- `nullable` — boolean
- `metadata` — existing key-value metadata (read this before writing to avoid redundant updates)
For struct/nested fields, recurse into `type.fields` and represent them as dot-notation paths (e.g., `parent.child`).
If the user hasn't specified which columns to update, work with all columns.
## Step 3: Generate metadata
Decide what to generate based on the user's request.
### Writing descriptions (`lancedb:description`)
Base descriptions on:
- The column name and Arrow type (e.g., `FixedSizeList` of floats → likely an embedding)
- User-supplied context (upstream pipeline, sample values, domain knowledge)
- Name patterns: `_embedding`/`_vec`/`_embed` → vector; `_label`/`_class` → label; `_score`/`_eval`/`_metric` → evaluation metric
Be specific and concise. Good: `"Sentence-BERT embedding of the query text (768-dim)."` Not: `"An embedding column."`
### Tagging columns (`lancedb:tag:<name>`)
Choose tag key names that match what the user asked to annotate. Common patterns:
- Semantic field type → `lancedb:tag:field_type: "embedding"` / `"text"` / `"image"` / `"label"` / `"eval"` / `"id"` / `"metadata"`
- Model or source → `lancedb:tag:model: "clip"` / `"bert"` / `"vit"`
- Project affiliation → `lancedb:tag:project_id: "<name>"`
- Version → `lancedb:tag:version: "v3"` (and `lancedb:tag:latest: "true"` for the newest)
Use Arrow type as a hint: `FixedSizeList` + float → embedding; `Utf8`/`LargeUtf8` → text; `Binary` → image or blob.
Multiple tags on the same column are fine — each is a separate key.
### Grouping into logical columns (`lancedb:logical-column`)
Look for naming patterns across columns:
- `clip_v1`, `clip_v2`, `clip_v3` → logical column `"clip"`, latest is `v3`
- `text_embed_20240101`, `text_embed_20240601` → logical column `"text_embed"`, latest is the most recent date suffix
Write `lancedb:logical-column` on all members of a group. Mark the newest with `lancedb:tag:latest: "true"` (in addition to its version tag).
## Step 4: Write the metadata
```http
POST {base_url}/v1/table/{table_id}/update_field_metadata
Content-Type: application/json
{
"updates": [
{
"path": "clip_v3",
"metadata": {
"lancedb:description": "CLIP ViT-L/14 image embedding, L2-normalized (1024-dim).",
"lancedb:tag:field_type": "embedding",
"lancedb:tag:model": "clip",
"lancedb:tag:version": "v3",
"lancedb:tag:latest": "true",
"lancedb:logical-column": "clip"
},
"replace": false
},
{
"path": "clip_v2",
"metadata": {
"lancedb:description": "CLIP ViT-B/32 image embedding (768-dim), superseded by v3.",
"lancedb:tag:field_type": "embedding",
"lancedb:tag:model": "clip",
"lancedb:tag:version": "v2",
"lancedb:logical-column": "clip"
},
"replace": false
}
]
}
```
Rules:
- **Use `"replace": false`** (merge) by default — this preserves existing metadata the user didn't ask to change
- Use `"replace": true` only if the user explicitly asks to overwrite all existing metadata on a column
- Set a value to `null` to delete a specific key
- Batch all updates in a single request when possible
The response includes `version` (new table version) and `fields` (the updated metadata per field).
## Step 5: Confirm
Report back:
- Which columns were updated and what was written
- The new table version number
- Any columns skipped (e.g., already had up-to-date metadata)
---
## Quick examples
**"Write descriptions for all columns in the `product_embeddings` table"**
1. POST `/v1/table/product_embeddings/describe` → get all fields
2. Generate a `lancedb:description` for each column based on name + type
3. POST `update_field_metadata` with descriptions
4. Report
**"Tag the columns in `model_outputs` with their field type and model"**
1. Describe `model_outputs`
2. For each field, classify by name + Arrow type → set `lancedb:tag:field_type` and `lancedb:tag:model` where applicable
3. POST `update_field_metadata`
4. Report
**"Group the feature columns in `training_features` into logical families and mark the latest version"**
1. Describe the table
2. Find version patterns → assign `lancedb:logical-column` and `lancedb:tag:version`; mark newest with `lancedb:tag:latest: "true"`
3. POST `update_field_metadata`
4. Show the grouping
+42
View File
@@ -0,0 +1,42 @@
---
name: lancedb-connect
description: Resolve how to connect to a LanceDB deployment over the REST API — figure out the base URL, API key, and database header. Use this before making any REST requests to a LanceDB table, whenever the endpoint or auth setup is not already known. Also useful on its own when someone asks how to connect, authenticate, or curl their LanceDB instance.
metadata:
short-description: Resolve the base URL and auth headers for a LanceDB deployment
---
## Goal
Produce two things every REST request needs:
1. **Base URL** — the endpoint
2. **Headers**`x-api-key`, and usually `x-lancedb-database`
## Resolution steps
1. If the user already gave a URL and API key (or said which environment they're working against), use that.
2. Otherwise, look for credentials already available in the environment:
- Env vars like `LANCEDB_URI` / `LANCEDB_HOST` / `LANCEDB_API_KEY`
- A LanceDB endpoint already running or port-forwarded locally (the REST default port is 2333, i.e. `http://localhost:2333`)
3. If you didn't find both pieces, ask the user directly: **"What's your LanceDB endpoint's URL, and what's your API key?"** Also ask which database to use if it isn't obvious. Don't guess or probe further — the user knows their deployment.
## Validating the connection
Make a cheap authenticated request and check the status:
```bash
curl -s -w "\n%{http_code}" "{base_url}/v1/table/?limit=1" \
-H "x-api-key: <key>" \
-H "x-lancedb-database: <database>"
```
- `200` — connection, key, and database header all good
- `401` — API key missing or wrong
- `400` mentioning a database header — this deployment expects `x-lancedb-database`
## Non-REST equivalents
If the caller would rather use the SDK or CLI than raw REST, the same credentials work:
- Python SDK: `lancedb.connect("db://<database>", api_key="<key>", host_override="<base_url>")`
- `lancedb` CLI: a `[profiles.<name>]` entry in `~/.lancedb/config.toml` with `http_server_url`, `api_key`, `database`
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion]
current_version = "0.32.0-beta.2"
current_version = "0.32.0-beta.1"
parse = """(?x)
(?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\.
-19
View File
@@ -1,19 +0,0 @@
{
"name": "lancedb",
"owner": {
"name": "LanceDB"
},
"description": "LanceDB plugins for Claude Code.",
"plugins": [
{
"name": "lancedb",
"source": "./plugins/lancedb",
"description": "Write, review, debug, and document LanceDB pipelines in Python and TypeScript that work across local LanceDB OSS tables and remote LanceDB Enterprise/Cloud tables.",
"version": "0.1.0",
"author": {
"name": "LanceDB"
},
"category": "development"
}
]
}
+21
View File
@@ -0,0 +1,21 @@
# CODEOWNERS
#
# These owners will be the default owners for everything in the repo.
# They will be requested for review when someone opens a pull request.
#
# See https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners
# Default owners for everything
* @jackye1995 @wjones127
# Release and publish workflows — changes here can affect supply chain security
/.github/workflows/ @jackye1995 @wjones127 @Xuanwo
# Remote client and auth — sensitive networking and auth code
/rust/lancedb/src/remote/ @jackye1995 @wjones127
# Python FFI boundary
/python/src/ @jackye1995 @wjones127 @AyushExel
# NodeJS FFI boundary
/nodejs/src/ @jackye1995 @wjones127
+2 -22
View File
@@ -18,14 +18,6 @@ inputs:
description: "The manylinux version to build for"
required: false
default: "2_17"
package-name:
description: "Override [project] name in python/pyproject.toml (e.g. 'lancedb-compat'). Default keeps 'lancedb'."
required: false
default: "lancedb"
rustflags:
description: "RUSTFLAGS for the build container, as a single whitespace-free token (e.g. '-Ctarget-cpu=x86-64-v2'). Empty leaves RUSTFLAGS unset, keeping the defaults from .cargo/config.toml."
required: false
default: ""
runs:
using: "composite"
steps:
@@ -35,18 +27,6 @@ runs:
ARM_BUILD: ${{ inputs.arm-build }}
run: |
echo "ARM BUILD: $ARM_BUILD"
- name: Patch package name for variant build
if: ${{ inputs.package-name != 'lancedb' }}
shell: bash
env:
PACKAGE_NAME: ${{ inputs.package-name }}
run: |
# Swap the [project] name so this build produces e.g. lancedb-compat
# wheels. The package still installs files under the lancedb/
# namespace -- import lancedb still works after pip install.
sed -i.bak 's/^name = "lancedb"$/name = "'"$PACKAGE_NAME"'"/' python/pyproject.toml
rm -f python/pyproject.toml.bak
grep '^name = ' python/pyproject.toml
- name: Build x86_64 Manylinux wheel
if: ${{ inputs.arm-build == 'false' }}
uses: PyO3/maturin-action@v1
@@ -54,7 +34,7 @@ runs:
maturin-version: "1.12.4"
command: build
working-directory: python
docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc ${{ inputs.rustflags != '' && format('-e RUSTFLAGS={0}', inputs.rustflags) || '' }}"
docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc"
target: x86_64-unknown-linux-gnu
manylinux: ${{ inputs.manylinux }}
args: ${{ inputs.args }}
@@ -71,7 +51,7 @@ runs:
maturin-version: "1.12.4"
command: build
working-directory: python
docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc ${{ inputs.rustflags != '' && format('-e RUSTFLAGS={0}', inputs.rustflags) || '' }}"
docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc"
target: aarch64-unknown-linux-gnu
manylinux: ${{ inputs.manylinux }}
args: ${{ inputs.args }}
+1 -1
View File
@@ -87,7 +87,7 @@ jobs:
bash ci/update_lockfiles.sh --amend
- name: Push new version tag
if: ${{ !inputs.dry_run }}
uses: ad-m/github-push-action@881a6320fdb16eb5318c5054f31c218aec2b324c # v1.3.0
uses: ad-m/github-push-action@master
with:
# Need to use PAT here too to trigger next workflow. See comment above.
github_token: ${{ secrets.LANCEDB_RELEASE_TOKEN }}
+4 -23
View File
@@ -22,7 +22,7 @@ permissions:
jobs:
linux:
name: Python ${{ matrix.config.package_name }} ${{ matrix.config.platform }} manylinux${{ matrix.config.manylinux }}
name: Python ${{ matrix.config.platform }} manylinux${{ matrix.config.manylinux }}
timeout-minutes: 60
strategy:
matrix:
@@ -31,28 +31,11 @@ jobs:
manylinux: "2_28"
extra_args: "--features fp16kernels"
runner: ubuntu-22.04
package_name: "lancedb"
rustflags: ""
# For successful fat LTO builds, we need a large runner to avoid OOM errors.
- platform: aarch64
manylinux: "2_28"
extra_args: "--features fp16kernels"
runner: ubuntu-2404-8x-arm64
package_name: "lancedb"
rustflags: ""
# `lancedb-compat`: pre-Haswell-friendly variant for x86_64 hosts
# without AVX2 (Sandy Bridge / Ivy Bridge / Westmere on Intel,
# Bulldozer / Piledriver / Steamroller on AMD). Compiled at the
# `x86-64-v2` baseline; runtime SIMD dispatch in lance-linalg
# picks the appropriate tier (scalar / AVX / AVX+FMA / AVX2+FMA
# / AVX-512) at load time. Same import as `lancedb` -- conflicts
# at install time, so users pick one.
- platform: x86_64
manylinux: "2_28"
extra_args: ""
runner: ubuntu-22.04
package_name: "lancedb-compat"
rustflags: "-Ctarget-cpu=x86-64-v2"
runs-on: ${{ matrix.config.runner }}
steps:
- uses: actions/checkout@v6
@@ -69,13 +52,11 @@ jobs:
args: "--release --strip ${{ matrix.config.extra_args }}"
arm-build: ${{ matrix.config.platform == 'aarch64' }}
manylinux: ${{ matrix.config.manylinux }}
package-name: ${{ matrix.config.package_name }}
rustflags: ${{ matrix.config.rustflags }}
- uses: actions/upload-artifact@v7
if: startsWith(github.ref, 'refs/tags/python-v')
with:
name: wheels-linux-${{ matrix.config.package_name }}-${{ matrix.config.platform }}-${{ matrix.config.manylinux }}
path: target/wheels/*.whl
name: wheels-linux-${{ matrix.config.platform }}-${{ matrix.config.manylinux }}
path: target/wheels/lancedb-*.whl
if-no-files-found: error
mac:
timeout-minutes: 90
@@ -164,7 +145,7 @@ jobs:
FURY_TOKEN: ${{ secrets.FURY_TOKEN }}
run: |
shopt -s nullglob
WHEELS=(target/wheels/*.whl)
WHEELS=(target/wheels/lancedb-*.whl)
if [[ ${#WHEELS[@]} -eq 0 ]]; then
echo "No wheels found in target/wheels/" >&2
exit 1
+6 -22
View File
@@ -98,7 +98,7 @@ jobs:
cargo build --profile ci --benches --all-features --tests
linux:
timeout-minutes: 60
timeout-minutes: 30
# To build all features, we need more disk space than is available
# on the free OSS github runner. This is mostly due to the the
# sentence-transformers feature.
@@ -125,26 +125,10 @@ jobs:
- uses: rui314/setup-mold@v1
- name: Make Swap
run: |
swapfile=/swapfile
min_swap_bytes=$((15 * 1024 * 1024 * 1024))
active_swap_bytes="$(sudo swapon --show=NAME,SIZE --bytes --noheadings | awk '$1 == "/swapfile" { print $2 }')"
if [ -n "$active_swap_bytes" ]; then
if [ "$active_swap_bytes" -ge "$min_swap_bytes" ]; then
echo "/swapfile is already active with enough space; skipping swap creation"
exit 0
fi
echo "/swapfile is already active but smaller than 16G; using /mnt/lancedb-swapfile"
swapfile=/mnt/lancedb-swapfile
fi
if sudo swapon --show=NAME --noheadings | grep -Fxq "$swapfile"; then
echo "$swapfile is already active; skipping swap creation"
exit 0
fi
sudo rm -f "$swapfile"
sudo fallocate -l 16G "$swapfile"
sudo chmod 600 "$swapfile"
sudo mkswap "$swapfile"
sudo swapon "$swapfile"
sudo fallocate -l 16G /swapfile
sudo chmod 600 /swapfile
sudo mkswap /swapfile
sudo swapon /swapfile
- name: Build
run: cargo build --profile ci --all-features --tests --locked --examples
- name: Run feature tests
@@ -158,7 +142,7 @@ jobs:
run: CARGO_ARGS="--profile ci" make -C ./lancedb remote-tests
macos:
timeout-minutes: 60
timeout-minutes: 30
strategy:
matrix:
mac-runner: ["macos-14", "macos-15"]
Generated
+229 -263
View File
File diff suppressed because it is too large Load Diff
+24 -23
View File
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0"
[workspace.dependencies]
lance = { path = "../lance/rust/lance", default-features = false }
lance-core = { path = "../lance/rust/lance-core" }
lance-datagen = { path = "../lance/rust/lance-datagen" }
lance-file = { path = "../lance/rust/lance-file" }
lance-io = { path = "../lance/rust/lance-io", default-features = false }
lance-index = { path = "../lance/rust/lance-index" }
lance-linalg = { path = "../lance/rust/lance-linalg" }
lance-namespace = { path = "../lance/rust/lance-namespace" }
lance-namespace-impls = { path = "../lance/rust/lance-namespace-impls", default-features = false }
lance-table = { path = "../lance/rust/lance-table" }
lance-testing = { path = "../lance/rust/lance-testing" }
lance-datafusion = { path = "../lance/rust/lance-datafusion" }
lance-encoding = { path = "../lance/rust/lance-encoding" }
lance-arrow = { path = "../lance/rust/lance-arrow" }
lance = { "version" = "=9.0.0-beta.19", default-features = false, "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=9.0.0-beta.19", default-features = false, "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=9.0.0-beta.19", default-features = false, "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
ahash = "0.8"
# Note that this one does not include pyarrow
arrow = { version = "58.0.0", optional = false }
@@ -39,15 +39,15 @@ arrow-schema = "58.0.0"
arrow-select = "58.0.0"
arrow-cast = "58.0.0"
async-trait = "0"
datafusion = { version = "54.0.0", default-features = false }
datafusion-catalog = "54.0.0"
datafusion-common = { version = "54.0.0", default-features = false }
datafusion-execution = "54.0.0"
datafusion-expr = "54.0.0"
datafusion-functions = "54.0.0"
datafusion-physical-plan = "54.0.0"
datafusion-physical-expr = "54.0.0"
datafusion-sql = "54.0.0"
datafusion = { version = "53.0.0", default-features = false }
datafusion-catalog = "53.0.0"
datafusion-common = { version = "53.0.0", default-features = false }
datafusion-execution = "53.0.0"
datafusion-expr = "53.0.0"
datafusion-functions = "53.0.0"
datafusion-physical-plan = "53.0.0"
datafusion-physical-expr = "53.0.0"
datafusion-sql = "53.0.0"
env_logger = "0.11"
half = { "version" = "2.7.1", default-features = false, features = [
"num-traits",
@@ -64,6 +64,7 @@ snafu = "0.8"
url = "2"
num-traits = "0.2"
regex = "1.10"
lazy_static = "1"
semver = "1.0.25"
chrono = "0.4"
+30 -165
View File
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
<dependency>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-core</artifactId>
<version>0.32.0-beta.2</version>
<version>0.32.0-beta.1</version>
</dependency>
```
@@ -249,57 +249,6 @@ try (BufferAllocator allocator = new RootAllocator();
}
```
### Creating an Empty Table
To create an empty table, send an Arrow IPC stream that contains the table schema and no record batches.
The schema in the IPC stream becomes the table schema, and rows can be inserted later.
```java
import org.lance.namespace.model.CreateTableRequest;
import org.lance.namespace.model.CreateTableResponse;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowStreamWriter;
import org.apache.arrow.vector.types.FloatingPointPrecision;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;
import java.io.ByteArrayOutputStream;
import java.nio.channels.Channels;
import java.util.Arrays;
Schema schema = new Schema(Arrays.asList(
new Field("id", FieldType.nullable(new ArrowType.Int(32, true)), null),
new Field("name", FieldType.nullable(new ArrowType.Utf8()), null),
new Field("embedding",
FieldType.nullable(new ArrowType.FixedSizeList(128)),
Arrays.asList(new Field("item",
FieldType.nullable(new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE)),
null)))
));
byte[] emptyTableData;
try (BufferAllocator allocator = new RootAllocator();
VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
root.setRowCount(0);
ByteArrayOutputStream out = new ByteArrayOutputStream();
try (ArrowStreamWriter writer = new ArrowStreamWriter(root, null, Channels.newChannel(out))) {
writer.start();
writer.end();
}
emptyTableData = out.toByteArray();
}
CreateTableRequest request = new CreateTableRequest();
request.setId(Arrays.asList("my_namespace", "empty_table"));
CreateTableResponse response = namespaceClient.createTable(request, emptyTableData);
```
### Insert
```java
@@ -482,88 +431,9 @@ query.setVector(vector);
byte[] result = namespaceClient.queryTable(query);
```
## Indexing
### Reading Query Results
The Java SDK exposes the REST namespace index operations through the same `LanceNamespace` client.
Index creation runs asynchronously, so use `listTableIndices` or `describeTableIndexStats` to check progress.
### Creating a Vector Index
```java
import org.lance.namespace.model.CreateTableIndexRequest;
import org.lance.namespace.model.CreateTableIndexResponse;
CreateTableIndexRequest request = new CreateTableIndexRequest();
request.setId(Arrays.asList("my_namespace", "my_table"));
request.setColumn("embedding");
request.setIndexType("IVF_PQ");
request.setDistanceType("cosine");
request.setName("embedding_idx");
CreateTableIndexResponse response = namespaceClient.createTableIndex(request);
System.out.println("Index transaction: " + response.getTransactionId());
```
### Creating a Scalar Index
```java
import org.lance.namespace.model.CreateTableIndexRequest;
import org.lance.namespace.model.CreateTableScalarIndexResponse;
CreateTableIndexRequest request = new CreateTableIndexRequest();
request.setId(Arrays.asList("my_namespace", "my_table"));
request.setColumn("category");
request.setIndexType("BTREE");
request.setName("category_idx");
CreateTableScalarIndexResponse response = namespaceClient.createTableScalarIndex(request);
System.out.println("Index transaction: " + response.getTransactionId());
```
### Creating a Full Text Search Index
```java
import org.lance.namespace.model.CreateTableIndexRequest;
import org.lance.namespace.model.CreateTableScalarIndexResponse;
CreateTableIndexRequest request = new CreateTableIndexRequest();
request.setId(Arrays.asList("my_namespace", "my_table"));
request.setColumn("text_column");
request.setIndexType("FTS");
request.setName("text_idx");
request.setBaseTokenizer("simple");
request.setLowerCase(true);
request.setWithPosition(true);
CreateTableScalarIndexResponse response = namespaceClient.createTableScalarIndex(request);
System.out.println("Index transaction: " + response.getTransactionId());
```
### Listing Indexes
```java
import org.lance.namespace.model.IndexContent;
import org.lance.namespace.model.ListTableIndicesRequest;
import org.lance.namespace.model.ListTableIndicesResponse;
ListTableIndicesRequest request = new ListTableIndicesRequest();
request.setId(Arrays.asList("my_namespace", "my_table"));
ListTableIndicesResponse response = namespaceClient.listTableIndices(request);
for (IndexContent index : response.getIndexes()) {
System.out.println(index.getIndexName() + ": " + index.getStatus());
}
```
!!! note
The current Java namespace API exposes index type, index name, distance type, and full text search tokenizer options.
IVF training parameters such as `num_partitions` are not exposed by `CreateTableIndexRequest` yet.
To make those configurable from Java, the namespace API must add those fields first.
## Reading Query Results
Query results are returned as bytes in Apache Arrow IPC file format. Put the byte-channel
adapter behind a small helper so query code can work with `ArrowFileReader` directly:
Query results are returned in Apache Arrow IPC file format. Here's how to read them:
```java
import org.apache.arrow.vector.ipc.ArrowFileReader;
@@ -571,50 +441,45 @@ import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SeekableByteChannel;
final class ArrowIpc {
static ArrowFileReader openFileReader(byte[] data, BufferAllocator allocator) throws IOException {
return new ArrowFileReader(new ByteArraySeekableByteChannel(data), allocator);
// Helper class to read Arrow data from byte array
class ByteArraySeekableByteChannel implements SeekableByteChannel {
private final byte[] data;
private long position = 0;
private boolean isOpen = true;
public ByteArraySeekableByteChannel(byte[] data) {
this.data = data;
}
private static final class ByteArraySeekableByteChannel implements SeekableByteChannel {
private final byte[] data;
private long position = 0;
private boolean isOpen = true;
private ByteArraySeekableByteChannel(byte[] data) {
this.data = data;
}
@Override
public int read(ByteBuffer dst) {
int remaining = dst.remaining();
int available = (int) (data.length - position);
if (available <= 0) return -1;
int toRead = Math.min(remaining, available);
dst.put(data, (int) position, toRead);
position += toRead;
return toRead;
}
@Override public long position() { return position; }
@Override public SeekableByteChannel position(long newPosition) { position = newPosition; return this; }
@Override public long size() { return data.length; }
@Override public boolean isOpen() { return isOpen; }
@Override public void close() { isOpen = false; }
@Override public int write(ByteBuffer src) { throw new UnsupportedOperationException(); }
@Override public SeekableByteChannel truncate(long size) { throw new UnsupportedOperationException(); }
@Override
public int read(ByteBuffer dst) {
int remaining = dst.remaining();
int available = (int) (data.length - position);
if (available <= 0) return -1;
int toRead = Math.min(remaining, available);
dst.put(data, (int) position, toRead);
position += toRead;
return toRead;
}
@Override public long position() { return position; }
@Override public SeekableByteChannel position(long newPosition) { position = newPosition; return this; }
@Override public long size() { return data.length; }
@Override public boolean isOpen() { return isOpen; }
@Override public void close() { isOpen = false; }
@Override public int write(ByteBuffer src) { throw new UnsupportedOperationException(); }
@Override public SeekableByteChannel truncate(long size) { throw new UnsupportedOperationException(); }
}
// Read query results
byte[] queryResult = namespaceClient.queryTable(query);
try (BufferAllocator allocator = new RootAllocator();
ArrowFileReader reader = ArrowIpc.openFileReader(queryResult, allocator)) {
ArrowFileReader reader = new ArrowFileReader(
new ByteArraySeekableByteChannel(queryResult), allocator)) {
for (int i = 0; i < reader.getRecordBlocks().size(); i++) {
reader.loadRecordBatch(reader.getRecordBlocks().get(i));
-43
View File
@@ -83,24 +83,6 @@ Delete a branch.
***
### diff()
```ts
diff(fromBranch): Promise<BranchDiff>
```
Compare a branch against main without modifying either branch.
#### Parameters
* **fromBranch**: `string`
#### Returns
`Promise`&lt;[`BranchDiff`](../interfaces/BranchDiff.md)&gt;
***
### list()
```ts
@@ -112,28 +94,3 @@ List all branches, mapping name to branch metadata.
#### Returns
`Promise`&lt;`Record`&lt;`string`, [`BranchContents`](BranchContents.md)&gt;&gt;
***
### merge()
```ts
merge(fromBranch, dryRun): Promise<MergeBranchResult>
```
Merge a branch into main.
Set `dryRun` to `true` to preview the merge. A rejected merge resolves
with `status: "rejected"` instead of throwing.
#### Parameters
* **fromBranch**: `string`
Branch to merge from.
* **dryRun**: `boolean` = `false`
When true, only preview the merge. Defaults to false.
#### Returns
`Promise`&lt;[`MergeBranchResult`](../interfaces/MergeBranchResult.md)&gt;
+1 -7
View File
@@ -33,7 +33,7 @@ protected inner: Query | Promise<Query>;
### analyzePlan()
```ts
analyzePlan(distributedMetrics?): Promise<string>
analyzePlan(): Promise<string>
```
Executes the query and returns the physical query plan annotated with runtime metrics.
@@ -41,12 +41,6 @@ Executes the query and returns the physical query plan annotated with runtime me
This is useful for debugging and performance analysis, as it shows how the query was executed
and includes metrics such as elapsed time, rows processed, and I/O statistics.
#### Parameters
* **distributedMetrics?**: [`AnalyzePlanDistributedMetrics`](../type-aliases/AnalyzePlanDistributedMetrics.md)
How distributed worker metrics are displayed for remote query plans.
Defaults to `"aggregate"`.
#### Returns
`Promise`&lt;`string`&gt;
+1 -7
View File
@@ -38,7 +38,7 @@ protected inner: NativeQueryType | Promise<NativeQueryType>;
### analyzePlan()
```ts
analyzePlan(distributedMetrics?): Promise<string>
analyzePlan(): Promise<string>
```
Executes the query and returns the physical query plan annotated with runtime metrics.
@@ -46,12 +46,6 @@ Executes the query and returns the physical query plan annotated with runtime me
This is useful for debugging and performance analysis, as it shows how the query was executed
and includes metrics such as elapsed time, rows processed, and I/O statistics.
#### Parameters
* **distributedMetrics?**: [`AnalyzePlanDistributedMetrics`](../type-aliases/AnalyzePlanDistributedMetrics.md)
How distributed worker metrics are displayed for remote query plans.
Defaults to `"aggregate"`.
#### Returns
`Promise`&lt;`string`&gt;
-26
View File
@@ -934,32 +934,6 @@ Return the table as an arrow table
***
### tokenize()
```ts
abstract tokenize(query, options): Promise<FtsToken[]>
```
Tokenize a full-text search query using the tokenizer configured on an FTS index.
Specify exactly one of `column` or `indexName`.
Model-backed tokenizers such as `jieba/*` and `lindera/*` are rebuilt in
the client process from index metadata. For remote tables, this means the
same tokenizer model files must also exist locally.
#### Parameters
* **query**: `string`
* **options**: [`TokenizeTableOptions`](../type-aliases/TokenizeTableOptions.md)
#### Returns
`Promise`&lt;[`FtsToken`](../interfaces/FtsToken.md)[]&gt;
***
### unsetLsmWriteSpec()
```ts
+1 -7
View File
@@ -29,7 +29,7 @@ protected inner: TakeQuery | Promise<TakeQuery>;
### analyzePlan()
```ts
analyzePlan(distributedMetrics?): Promise<string>
analyzePlan(): Promise<string>
```
Executes the query and returns the physical query plan annotated with runtime metrics.
@@ -37,12 +37,6 @@ Executes the query and returns the physical query plan annotated with runtime me
This is useful for debugging and performance analysis, as it shows how the query was executed
and includes metrics such as elapsed time, rows processed, and I/O statistics.
#### Parameters
* **distributedMetrics?**: [`AnalyzePlanDistributedMetrics`](../type-aliases/AnalyzePlanDistributedMetrics.md)
How distributed worker metrics are displayed for remote query plans.
Defaults to `"aggregate"`.
#### Returns
`Promise`&lt;`string`&gt;
+1 -7
View File
@@ -51,7 +51,7 @@ addQueryVector(vector): VectorQuery
### analyzePlan()
```ts
analyzePlan(distributedMetrics?): Promise<string>
analyzePlan(): Promise<string>
```
Executes the query and returns the physical query plan annotated with runtime metrics.
@@ -59,12 +59,6 @@ Executes the query and returns the physical query plan annotated with runtime me
This is useful for debugging and performance analysis, as it shows how the query was executed
and includes metrics such as elapsed time, rows processed, and I/O statistics.
#### Parameters
* **distributedMetrics?**: [`AnalyzePlanDistributedMetrics`](../type-aliases/AnalyzePlanDistributedMetrics.md)
How distributed worker metrics are displayed for remote query plans.
Defaults to `"aggregate"`.
#### Returns
`Promise`&lt;`string`&gt;
-26
View File
@@ -1,26 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / tokenize
# Function: tokenize()
```ts
function tokenize(query, options?): Promise<FtsToken[]>
```
Tokenize a full-text search query using an explicit tokenizer.
This does not require a table or FTS index. The tokenizer options match
[Index.fts](../classes/Index.md#fts).
## Parameters
* **query**: `string`
* **options?**: `Partial`&lt;[`TokenizeOptions`](../interfaces/TokenizeOptions.md)&gt;
## Returns
`Promise`&lt;[`FtsToken`](../interfaces/FtsToken.md)[]&gt;
-14
View File
@@ -52,11 +52,6 @@
- [AddDataOptions](interfaces/AddDataOptions.md)
- [AddResult](interfaces/AddResult.md)
- [AlterColumnsResult](interfaces/AlterColumnsResult.md)
- [BranchColumnChange](interfaces/BranchColumnChange.md)
- [BranchColumnSummary](interfaces/BranchColumnSummary.md)
- [BranchDiff](interfaces/BranchDiff.md)
- [BranchIndexSummary](interfaces/BranchIndexSummary.md)
- [BranchRowCountSummary](interfaces/BranchRowCountSummary.md)
- [ClientConfig](interfaces/ClientConfig.md)
- [ColumnAlteration](interfaces/ColumnAlteration.md)
- [ColumnOrdering](interfaces/ColumnOrdering.md)
@@ -77,7 +72,6 @@
- [FragmentStatistics](interfaces/FragmentStatistics.md)
- [FragmentSummaryStats](interfaces/FragmentSummaryStats.md)
- [FtsOptions](interfaces/FtsOptions.md)
- [FtsToken](interfaces/FtsToken.md)
- [FullTextQuery](interfaces/FullTextQuery.md)
- [FullTextSearchOptions](interfaces/FullTextSearchOptions.md)
- [HnswPqOptions](interfaces/HnswPqOptions.md)
@@ -91,9 +85,6 @@
- [ListNamespacesOptions](interfaces/ListNamespacesOptions.md)
- [ListNamespacesResponse](interfaces/ListNamespacesResponse.md)
- [LsmWriteSpec](interfaces/LsmWriteSpec.md)
- [MergeBlocker](interfaces/MergeBlocker.md)
- [MergeBranchResult](interfaces/MergeBranchResult.md)
- [MergePreview](interfaces/MergePreview.md)
- [MergeResult](interfaces/MergeResult.md)
- [NativeOAuthConfig](interfaces/NativeOAuthConfig.md)
- [OAuthConfig](interfaces/OAuthConfig.md)
@@ -116,7 +107,6 @@
- [TimeoutConfig](interfaces/TimeoutConfig.md)
- [TlsConfig](interfaces/TlsConfig.md)
- [TokenResponse](interfaces/TokenResponse.md)
- [TokenizeOptions](interfaces/TokenizeOptions.md)
- [UpdateFieldMetadataResult](interfaces/UpdateFieldMetadataResult.md)
- [UpdateOptions](interfaces/UpdateOptions.md)
- [UpdateResult](interfaces/UpdateResult.md)
@@ -126,8 +116,6 @@
## Type Aliases
- [AnalyzePlanDistributedMetrics](type-aliases/AnalyzePlanDistributedMetrics.md)
- [BaseTokenizer](type-aliases/BaseTokenizer.md)
- [Data](type-aliases/Data.md)
- [DataLike](type-aliases/DataLike.md)
- [FieldLike](type-aliases/FieldLike.md)
@@ -137,7 +125,6 @@
- [RecordBatchLike](type-aliases/RecordBatchLike.md)
- [SchemaLike](type-aliases/SchemaLike.md)
- [TableLike](type-aliases/TableLike.md)
- [TokenizeTableOptions](type-aliases/TokenizeTableOptions.md)
## Functions
@@ -148,4 +135,3 @@
- [makeArrowTable](functions/makeArrowTable.md)
- [packBits](functions/packBits.md)
- [permutationBuilder](functions/permutationBuilder.md)
- [tokenize](functions/tokenize.md)
@@ -1,33 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BranchColumnChange
# Interface: BranchColumnChange
A column whose definition differs between main and the branch.
## Properties
### branch
```ts
branch: BranchColumnSummary;
```
***
### main
```ts
main: BranchColumnSummary;
```
***
### name
```ts
name: string;
```
@@ -1,33 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BranchColumnSummary
# Interface: BranchColumnSummary
Summary of a column in a branch diff.
## Properties
### dataType
```ts
dataType: string;
```
***
### name
```ts
name: string;
```
***
### nullable
```ts
nullable: boolean;
```
-129
View File
@@ -1,129 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BranchDiff
# Interface: BranchDiff
Read-only comparison of a branch against main.
## Properties
### addedColumns
```ts
addedColumns: BranchColumnSummary[];
```
***
### addedIndexes
```ts
addedIndexes: BranchIndexSummary[];
```
***
### baseMoved
```ts
baseMoved: boolean;
```
***
### branchVersion
```ts
branchVersion: number;
```
***
### changedColumns
```ts
changedColumns: BranchColumnChange[];
```
***
### fromBranch
```ts
fromBranch: string;
```
***
### mainVersion
```ts
mainVersion: number;
```
***
### mergeBlockers
```ts
mergeBlockers: MergeBlocker[];
```
***
### mergeable
```ts
mergeable: boolean;
```
***
### parentVersion
```ts
parentVersion: number;
```
***
### removedColumns
```ts
removedColumns: BranchColumnSummary[];
```
***
### removedIndexes
```ts
removedIndexes: BranchIndexSummary[];
```
***
### rowCountBranch
```ts
rowCountBranch: number;
```
***
### rowCountMain
```ts
rowCountMain: number;
```
***
### rowSummary
```ts
rowSummary: BranchRowCountSummary;
```
@@ -1,41 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BranchIndexSummary
# Interface: BranchIndexSummary
Summary of an index in a branch diff.
## Properties
### columns
```ts
columns: string[];
```
***
### indexName
```ts
indexName: string;
```
***
### indexType?
```ts
optional indexType: string;
```
***
### status
```ts
status: string;
```
@@ -1,57 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BranchRowCountSummary
# Interface: BranchRowCountSummary
Row-level comparison between main and the branch.
## Properties
### deltaAvailable
```ts
deltaAvailable: boolean;
```
***
### inputsChanged
```ts
inputsChanged: number;
```
***
### newOnBase
```ts
newOnBase: number;
```
***
### newOnBranch
```ts
newOnBranch: number;
```
***
### staleRecompute
```ts
staleRecompute: number;
```
***
### unchanged
```ts
unchanged: number;
```
+1 -5
View File
@@ -23,7 +23,7 @@ whether to remove punctuation
### baseTokenizer?
```ts
optional baseTokenizer: BaseTokenizer;
optional baseTokenizer: "raw" | "simple" | "whitespace" | "ngram";
```
The tokenizer to use when building the index.
@@ -37,10 +37,6 @@ The following tokenizers are available:
"raw" - Raw tokenizer. This tokenizer does not split the text into tokens and indexes the entire text as a single token.
"icu" - ICU dictionary-based word segmentation.
"icu/split" - ICU segmentation with simple-style delimiter splitting.
***
### language?
-29
View File
@@ -1,29 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / FtsToken
# Interface: FtsToken
Token produced by the tokenizer configured on a full-text search index.
## Properties
### position
```ts
position: number;
```
Token position used by full-text query matching.
***
### text
```ts
text: string;
```
Token text after tokenizer filters have been applied.
-25
View File
@@ -1,25 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / MergeBlocker
# Interface: MergeBlocker
A reason why a branch cannot currently be merged.
## Properties
### code
```ts
code: string;
```
***
### message
```ts
message: string;
```
@@ -1,46 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / MergeBranchResult
# Interface: MergeBranchResult
Result of previewing or attempting a branch merge.
## Properties
### diff
```ts
diff: BranchDiff;
```
***
### mainVersionAfter?
```ts
optional mainVersionAfter: number;
```
***
### preview
```ts
preview: MergePreview;
```
***
### status
```ts
status:
| "unknown"
| "rejected"
| "ready"
| "notImplemented"
| "merged";
```
-17
View File
@@ -1,17 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / MergePreview
# Interface: MergePreview
Changes that would be, or were, promoted by a branch merge.
## Properties
### promotedColumns
```ts
promotedColumns: string[];
```
-109
View File
@@ -1,109 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / TokenizeOptions
# Interface: TokenizeOptions
Options for tokenizing a full-text search query without a table index.
## Properties
### asciiFolding?
```ts
optional asciiFolding: boolean;
```
Whether to fold ASCII characters.
***
### baseTokenizer?
```ts
optional baseTokenizer: BaseTokenizer;
```
The tokenizer to use. The default is "simple".
***
### language?
```ts
optional language: string;
```
Language for stemming and stop words.
***
### lowercase?
```ts
optional lowercase: boolean;
```
Whether to lowercase tokens.
***
### maxTokenLength?
```ts
optional maxTokenLength: number;
```
Maximum token length; tokens longer than this are ignored.
***
### ngramMaxLength?
```ts
optional ngramMaxLength: number;
```
N-gram maximum length.
***
### ngramMinLength?
```ts
optional ngramMinLength: number;
```
N-gram minimum length.
***
### prefixOnly?
```ts
optional prefixOnly: boolean;
```
Whether to only emit token prefixes for the n-gram tokenizer.
***
### removeStopWords?
```ts
optional removeStopWords: boolean;
```
Whether to remove stop words.
***
### stem?
```ts
optional stem: boolean;
```
Whether to stem tokens.
@@ -1,11 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / AnalyzePlanDistributedMetrics
# Type Alias: AnalyzePlanDistributedMetrics
```ts
type AnalyzePlanDistributedMetrics: "aggregate" | "per_worker" | "full";
```
-19
View File
@@ -1,19 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BaseTokenizer
# Type Alias: BaseTokenizer
```ts
type BaseTokenizer:
| "simple"
| "whitespace"
| "raw"
| "ngram"
| "icu"
| "icu/split"
| `jieba/${string}`
| `lindera/${string}`;
```
@@ -1,11 +0,0 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / TokenizeTableOptions
# Type Alias: TokenizeTableOptions
```ts
type TokenizeTableOptions: object | object;
```
+1 -1
View File
@@ -8,7 +8,7 @@
<parent>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.32.0-beta.2</version>
<version>0.32.0-beta.1</version>
<relativePath>../pom.xml</relativePath>
</parent>
+2 -2
View File
@@ -6,7 +6,7 @@
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.32.0-beta.2</version>
<version>0.32.0-beta.1</version>
<packaging>pom</packaging>
<name>${project.artifactId}</name>
<description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<arrow.version>15.0.0</arrow.version>
<lance-core.version>10.0.0-beta.2</lance-core.version>
<lance-core.version>9.0.0-beta.19</lance-core.version>
<spotless.skip>false</spotless.skip>
<spotless.version>2.30.0</spotless.version>
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "lancedb-nodejs"
edition.workspace = true
version = "0.32.0-beta.2"
version = "0.32.0-beta.1"
publish = false
license.workspace = true
description.workspace = true
-53
View File
@@ -52,7 +52,6 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
Float64,
Struct,
List,
Map_,
Int16,
Int32,
Int64,
@@ -70,30 +69,6 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
type Schema = ApacheArrow["Schema"];
type Table = ApacheArrow["Table"];
function expectValidMapField(
// biome-ignore lint/suspicious/noExplicitAny: Arrow Field types vary across supported versions
field: any,
): void {
expect(DataType.isMap(field.type)).toBe(true);
expect(field.type.keysSorted).toBe(true);
expect(field.type.children).toHaveLength(1);
const entries = field.type.children[0];
expect(entries.name).toBe("entries");
expect(entries.nullable).toBe(false);
expect(DataType.isStruct(entries.type)).toBe(true);
expect(entries.type.children).toHaveLength(2);
const [key, value] = entries.type.children;
expect([key.name, value.name]).toEqual(["key", "value"]);
expect(key.nullable).toBe(false);
expect(DataType.isUtf8(key.type)).toBe(true);
expect(value.nullable).toBe(true);
expect(DataType.isInt(value.type)).toBe(true);
expect(value.type.bitWidth).toBe(32);
expect(value.type.isSigned).toBe(true);
}
// Helper method to verify various ways to create a table
async function checkTableCreation(
tableCreationMethod: (
@@ -963,34 +938,6 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
false,
);
});
it("will make an empty table with a Map field", async function () {
const schema = new Schema([
new Field(
"attributes",
new Map_(
new Field(
"entries",
new Struct([
new Field("key", new Utf8(), false),
new Field("value", new Int32(), true),
]),
false,
),
true,
),
),
]);
const table = makeEmptyTable(schema);
expectValidMapField(table.schema.fields[0]);
const buffer = await fromTableToBuffer(table);
const roundTripped = tableFromIPC(buffer);
expectValidMapField(roundTripped.schema.fields[0]);
});
});
describe("when using two versions of arrow", function () {
-107
View File
@@ -225,113 +225,6 @@ describe("remote connection", () => {
);
});
it("diffs and merges remote branches", async () => {
const sampleDiff = {
fromBranch: "exp",
parentVersion: 1,
mainVersion: 2,
branchVersion: 3,
baseMoved: false,
rowCountMain: 3,
rowCountBranch: 3,
rowSummary: {
unchanged: 3,
newOnBase: 0,
newOnBranch: 0,
staleRecompute: 0,
inputsChanged: 0,
deltaAvailable: false,
},
addedColumns: [{ name: "tag", dataType: "utf8", nullable: true }],
removedColumns: [],
changedColumns: [],
addedIndexes: [],
removedIndexes: [],
mergeable: true,
mergeBlockers: [],
};
const mergeBodies: Record<string, unknown>[] = [];
await withMockDatabase(
(req, res) => {
const path = req.url ?? "";
if (path.endsWith("/describe/")) {
res.writeHead(200, { "Content-Type": "application/json" }).end(
JSON.stringify({
name: "t",
version: 2,
schema: { fields: [] },
}),
);
return;
}
let raw = "";
req.on("data", (chunk) => {
raw += chunk;
});
req.on("end", () => {
const body = raw ? JSON.parse(raw) : {};
if (path.endsWith("/branches/diff/")) {
// biome-ignore lint/style/useNamingConvention: snake_case mandated by the server wire format
expect(body).toEqual({ from_branch: "exp" });
res
.writeHead(200, { "Content-Type": "application/json" })
.end(JSON.stringify(sampleDiff));
return;
}
if (path.endsWith("/branches/merge/")) {
mergeBodies.push(body);
const dryRun = body["dry_run"] === true;
const response = {
status: dryRun ? "ready" : "rejected",
diff: dryRun
? sampleDiff
: {
...sampleDiff,
mergeable: false,
mergeBlockers: [
{ code: "baseMoved", message: "main has advanced" },
],
},
preview: { promotedColumns: dryRun ? ["tag"] : [] },
};
res
.writeHead(dryRun ? 200 : 409, {
"Content-Type": "application/json",
})
.end(JSON.stringify(response));
return;
}
res.writeHead(404).end();
});
},
async (db) => {
const table = await db.openTable("t");
const branches = await table.branches();
await expect(branches.diff("exp")).resolves.toEqual(sampleDiff);
const rejected = await branches.merge("exp");
expect(rejected.status).toBe("rejected");
expect(rejected.diff.mergeBlockers).toEqual([
{ code: "baseMoved", message: "main has advanced" },
]);
const preview = await branches.merge("exp", true);
expect(preview.status).toBe("ready");
expect(preview.preview.promotedColumns).toEqual(["tag"]);
},
);
expect(mergeBodies).toEqual([
// biome-ignore lint/style/useNamingConvention: snake_case mandated by the server wire format
{ from_branch: "exp", dry_run: false },
// biome-ignore lint/style/useNamingConvention: snake_case mandated by the server wire format
{ from_branch: "exp", dry_run: true },
]);
});
describe("TlsConfig", () => {
it("should create TlsConfig with all fields", () => {
const tlsConfig: TlsConfig = {
+1 -12
View File
@@ -2,7 +2,7 @@
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import * as arrow from "../lancedb/arrow";
import { sanitizeField, sanitizeMap, sanitizeType } from "../lancedb/sanitize";
import { sanitizeField, sanitizeType } from "../lancedb/sanitize";
describe("sanitize", function () {
describe("sanitizeType function", function () {
@@ -181,15 +181,4 @@ describe("sanitize", function () {
);
});
});
describe("sanitizeMap function", function () {
it.each([
["no children", []],
["two children", [{}, {}]],
])("should reject a Map type with %s", function (_, children) {
expect(() => sanitizeMap({ children, keysSorted: false })).toThrow(
"Expected a Map type to have exactly one child",
);
});
});
});
+1 -76
View File
@@ -16,7 +16,6 @@ import {
PhraseQuery,
Table,
connect,
tokenize,
} from "../lancedb";
import {
Table as ArrowTable,
@@ -2308,75 +2307,6 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
expect(results2[0].text).toBe(data[1].text);
});
test("tokenizes FTS queries by column or index name", async () => {
const db = await connect(tmpDir.name);
const data = [
{
text: "Running in cafés",
japanese: "Hello, こんにちは世界!",
vector: [0.1, 0.2, 0.3],
},
];
const table = await db.createTable("test", data);
await table.createIndex("text", {
config: Index.fts({ baseTokenizer: "simple" }),
});
await table.createIndex("japanese", {
config: Index.fts({
baseTokenizer: "icu",
stem: false,
removeStopWords: false,
}),
name: "japanese_icu_idx",
});
await expect(table.tokenize("hello", {} as never)).rejects.toThrow(
"Specify exactly one",
);
await expect(
table.tokenize("hello", {
column: "text",
indexName: "text_idx",
} as never),
).rejects.toThrow("Specify exactly one");
const simpleTokens = await table.tokenize("Running in cafés", {
column: "text",
});
expect(simpleTokens).toEqual([
{ text: "run", position: 0 },
{ text: "cafe", position: 2 },
]);
const icuTokens = await table.tokenize("Hello, こんにちは世界!", {
indexName: "japanese_icu_idx",
});
expect(icuTokens).toEqual([
{ text: "hello", position: 0 },
{ text: "こんにちは", position: 1 },
{ text: "世界", position: 2 },
]);
const directSimpleTokens = await tokenize("Running in cafés", {
baseTokenizer: "simple",
});
expect(directSimpleTokens).toEqual([
{ text: "run", position: 0 },
{ text: "cafe", position: 2 },
]);
const directIcuTokens = await tokenize("Hello, こんにちは世界!", {
baseTokenizer: "icu",
stem: false,
removeStopWords: false,
});
expect(directIcuTokens).toEqual([
{ text: "hello", position: 0 },
{ text: "こんにちは", position: 1 },
{ text: "世界", position: 2 },
]);
});
test("full text search fast search", async () => {
const db = await connect(tmpDir.name);
const data = [{ text: "hello world", vector: [0.1, 0.2, 0.3], id: 1 }];
@@ -2775,13 +2705,8 @@ describe("when calling analyzePlan", () => {
.fill(1)
.map(() => Math.random());
const plan = await table.query().nearestTo(queryVec).analyzePlan();
console.log("Query Plan:\n", plan); // <--- Print the plan
expect(plan).toMatch("AnalyzeExec");
const fullPlan = await table
.query()
.nearestTo(queryVec)
.analyzePlan("full");
expect(fullPlan).toMatch("AnalyzeExec");
});
});
-77
View File
@@ -13,12 +13,9 @@ import {
Connection as LanceDbConnection,
JsHeaderProvider as NativeJsHeaderProvider,
Session,
tokenize as nativeTokenize,
} from "./native.js";
import { HeaderProvider } from "./header";
import type { BaseTokenizer } from "./indices";
import type { FtsToken } from "./table";
// Re-export native header provider for use with connectWithHeaderProvider
export { JsHeaderProvider as NativeJsHeaderProvider } from "./native.js";
@@ -93,7 +90,6 @@ export {
QueryBase,
VectorQuery,
TakeQuery,
AnalyzePlanDistributedMetrics,
QueryExecutionOptions,
ColumnOrdering,
FullTextSearchOptions,
@@ -118,27 +114,16 @@ export {
HnswPqOptions,
HnswSqOptions,
FtsOptions,
BaseTokenizer,
} from "./indices";
export {
Table,
Branches,
BranchColumnSummary,
BranchColumnChange,
BranchIndexSummary,
BranchRowCountSummary,
MergeBlocker,
BranchDiff,
MergePreview,
MergeBranchResult,
AddDataOptions,
UpdateOptions,
OptimizeOptions,
Version,
WriteProgress,
FtsToken,
TokenizeTableOptions,
LsmWriteSpec,
ColumnAlteration,
FieldMetadataUpdate,
@@ -170,68 +155,6 @@ export {
} from "./arrow";
export { IntoSql, packBits } from "./util";
/**
* Options for tokenizing a full-text search query without a table index.
*/
export interface TokenizeOptions {
/**
* The tokenizer to use. The default is "simple".
*/
baseTokenizer?: BaseTokenizer;
/** Language for stemming and stop words. */
language?: string;
/** Maximum token length; tokens longer than this are ignored. */
maxTokenLength?: number;
/** Whether to lowercase tokens. */
lowercase?: boolean;
/** Whether to stem tokens. */
stem?: boolean;
/** Whether to remove stop words. */
removeStopWords?: boolean;
/** Whether to fold ASCII characters. */
asciiFolding?: boolean;
/** N-gram minimum length. */
ngramMinLength?: number;
/** N-gram maximum length. */
ngramMaxLength?: number;
/** Whether to only emit token prefixes for the n-gram tokenizer. */
prefixOnly?: boolean;
}
/**
* Tokenize a full-text search query using an explicit tokenizer.
*
* This does not require a table or FTS index. The tokenizer options match
* {@link Index.fts}.
*/
export async function tokenize(
query: string,
options?: Partial<TokenizeOptions>,
): Promise<FtsToken[]> {
return await nativeTokenize(
query,
options?.baseTokenizer,
options?.language,
options?.maxTokenLength,
options?.lowercase,
options?.stem,
options?.removeStopWords,
options?.asciiFolding,
options?.ngramMinLength,
options?.ngramMaxLength,
options?.prefixOnly,
);
}
/**
* Connect to a LanceDB instance at the given URI.
*
+1 -15
View File
@@ -486,16 +486,6 @@ export interface IvfFlatOptions {
sampleRate?: number;
}
export type BaseTokenizer =
| "simple"
| "whitespace"
| "raw"
| "ngram"
| "icu"
| "icu/split"
| `jieba/${string}`
| `lindera/${string}`;
/**
* Options to create a full text search index
*/
@@ -519,12 +509,8 @@ export interface FtsOptions {
* "whitespace" - Whitespace tokenizer. This tokenizer splits the text into tokens using whitespace as a delimiter.
*
* "raw" - Raw tokenizer. This tokenizer does not split the text into tokens and indexes the entire text as a single token.
*
* "icu" - ICU dictionary-based word segmentation.
*
* "icu/split" - ICU segmentation with simple-style delimiter splitting.
*/
baseTokenizer?: BaseTokenizer;
baseTokenizer?: "simple" | "whitespace" | "raw" | "ngram";
/**
* language for stemming and stop words
+3 -12
View File
@@ -79,8 +79,6 @@ export interface QueryExecutionOptions {
timeoutMs?: number;
}
export type AnalyzePlanDistributedMetrics = "aggregate" | "per_worker" | "full";
export interface ColumnOrdering {
columnName: string;
ascending?: boolean;
@@ -313,20 +311,13 @@ export class QueryBase<
* KNNVectorDistance: metric=l2, metrics=[output_rows=1, elapsed_compute=114.333µs, output_batches=1]
* LanceScan: uri=/path/to/data, projection=[vector], row_id=true, row_addr=false, ordered=false, metrics=[output_rows=1, elapsed_compute=103.626µs, bytes_read=549, iops=2, requests=2]
*
* @param distributedMetrics - How distributed worker metrics are displayed for remote query plans.
* Defaults to `"aggregate"`.
* @returns A query execution plan with runtime metrics for each step.
*/
async analyzePlan(
distributedMetrics?: AnalyzePlanDistributedMetrics,
): Promise<string> {
const distributedMetricsMode = distributedMetrics ?? "aggregate";
async analyzePlan(): Promise<string> {
if (this.inner instanceof Promise) {
return this.inner.then((inner) =>
inner.analyzePlan(distributedMetricsMode),
);
return this.inner.then((inner) => inner.analyzePlan());
} else {
return this.inner.analyzePlan(distributedMetricsMode);
return this.inner.analyzePlan();
}
}
+5 -4
View File
@@ -288,11 +288,12 @@ export function sanitizeMap(typeLike: object) {
if (!("keysSorted" in typeLike) || typeof typeLike.keysSorted !== "boolean") {
throw Error("Expected a Map type to have a `keysSorted` property");
}
if (typeLike.children.length !== 1) {
throw Error("Expected a Map type to have exactly one child");
}
return new Map_(sanitizeField(typeLike.children[0]), typeLike.keysSorted);
return new Map_(
// biome-ignore lint/suspicious/noExplicitAny: skip
typeLike.children.map((field) => sanitizeField(field)) as any,
typeLike.keysSorted,
);
}
export function sanitizeDuration(typeLike: object) {
-138
View File
@@ -158,26 +158,6 @@ export interface Version {
metadata: Record<string, string>;
}
/** Token produced by the tokenizer configured on a full-text search index. */
export interface FtsToken {
/** Token text after tokenizer filters have been applied. */
text: string;
/** Token position used by full-text query matching. */
position: number;
}
export type TokenizeTableOptions =
| {
/** FTS-indexed column whose tokenizer should be used. */
column: string;
indexName?: never;
}
| {
/** Name of the FTS index whose tokenizer should be used. */
indexName: string;
column?: never;
};
/**
* Specification selecting Lance's MemWAL LSM-style write path for
* `mergeInsert`.
@@ -736,19 +716,6 @@ export abstract class Table {
abstract optimize(options?: Partial<OptimizeOptions>): Promise<OptimizeStats>;
/** List all indices that have been created with {@link Table.createIndex} */
abstract listIndices(): Promise<IndexConfig[]>;
/**
* Tokenize a full-text search query using the tokenizer configured on an FTS index.
*
* Specify exactly one of `column` or `indexName`.
*
* Model-backed tokenizers such as `jieba/*` and `lindera/*` are rebuilt in
* the client process from index metadata. For remote tables, this means the
* same tokenizer model files must also exist locally.
*/
abstract tokenize(
query: string,
options: TokenizeTableOptions,
): Promise<FtsToken[]>;
/** Return the table as an arrow table */
abstract toArrow(): Promise<ArrowTable>;
@@ -1206,17 +1173,6 @@ export class LocalTable extends Table {
return await this.inner.listIndices();
}
async tokenize(
query: string,
options: TokenizeTableOptions,
): Promise<FtsToken[]> {
return await this.inner.tokenize(
query,
options?.column,
options?.indexName,
);
}
async toArrow(): Promise<ArrowTable> {
return await this.query().toArrow();
}
@@ -1329,76 +1285,6 @@ export interface FieldMetadataUpdate {
replace?: boolean;
}
/** Summary of a column in a branch diff. */
export interface BranchColumnSummary {
name: string;
dataType: string;
nullable: boolean;
}
/** A column whose definition differs between main and the branch. */
export interface BranchColumnChange {
name: string;
main: BranchColumnSummary;
branch: BranchColumnSummary;
}
/** Summary of an index in a branch diff. */
export interface BranchIndexSummary {
indexName: string;
columns: string[];
indexType?: string;
status: string;
}
/** Row-level comparison between main and the branch. */
export interface BranchRowCountSummary {
unchanged: number;
newOnBase: number;
newOnBranch: number;
staleRecompute: number;
inputsChanged: number;
deltaAvailable: boolean;
}
/** A reason why a branch cannot currently be merged. */
export interface MergeBlocker {
code: string;
message: string;
}
/** Read-only comparison of a branch against main. */
export interface BranchDiff {
fromBranch: string;
parentVersion: number;
mainVersion: number;
branchVersion: number;
baseMoved: boolean;
rowCountMain: number;
rowCountBranch: number;
rowSummary: BranchRowCountSummary;
addedColumns: BranchColumnSummary[];
removedColumns: BranchColumnSummary[];
changedColumns: BranchColumnChange[];
addedIndexes: BranchIndexSummary[];
removedIndexes: BranchIndexSummary[];
mergeable: boolean;
mergeBlockers: MergeBlocker[];
}
/** Changes that would be, or were, promoted by a branch merge. */
export interface MergePreview {
promotedColumns: string[];
}
/** Result of previewing or attempting a branch merge. */
export interface MergeBranchResult {
status: "ready" | "rejected" | "notImplemented" | "merged" | "unknown";
diff: BranchDiff;
preview: MergePreview;
mainVersionAfter?: number;
}
/**
* Branch manager for a {@link Table}.
*
@@ -1451,28 +1337,4 @@ export class Branches {
async delete(name: string): Promise<void> {
return await this.#inner.delete(name);
}
/** Compare a branch against main without modifying either branch. */
async diff(fromBranch: string): Promise<BranchDiff> {
return (await this.#inner.diff(fromBranch)) as unknown as BranchDiff;
}
/**
* Merge a branch into main.
*
* Set `dryRun` to `true` to preview the merge. A rejected merge resolves
* with `status: "rejected"` instead of throwing.
*
* @param fromBranch Branch to merge from.
* @param dryRun When true, only preview the merge. Defaults to false.
*/
async merge(
fromBranch: string,
dryRun: boolean = false,
): Promise<MergeBranchResult> {
return (await this.#inner.merge(
fromBranch,
dryRun,
)) as unknown as MergeBranchResult;
}
}
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-darwin-arm64",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"os": ["darwin"],
"cpu": ["arm64"],
"main": "lancedb.darwin-arm64.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-gnu",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-musl",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-gnu",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-musl",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-arm64-msvc",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"os": [
"win32"
],
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-x64-msvc",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"os": ["win32"],
"cpu": ["x64"],
"main": "lancedb.win32-x64-msvc.node",
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "@lancedb/lancedb",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.0",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "@lancedb/lancedb",
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.0",
"cpu": [
"x64",
"arm64"
+1 -1
View File
@@ -11,7 +11,7 @@
"ann"
],
"private": false,
"version": "0.32.0-beta.2",
"version": "0.32.0-beta.1",
"main": "dist/index.js",
"exports": {
".": "./dist/index.js",
-62
View File
@@ -9,11 +9,8 @@ use lancedb::index::vector::{
IvfFlatIndexBuilder, IvfHnswPqIndexBuilder, IvfHnswSqIndexBuilder, IvfPqIndexBuilder,
IvfRqIndexBuilder,
};
use lancedb::tokenize as lancedb_tokenize;
use napi_derive::napi;
use crate::error::NapiErrorExt;
use crate::table::FtsToken;
use crate::util::parse_distance_type;
#[napi]
@@ -33,65 +30,6 @@ impl Index {
}
}
#[napi(catch_unwind)]
#[allow(dead_code, clippy::too_many_arguments)]
pub fn tokenize(
query: String,
base_tokenizer: Option<String>,
language: Option<String>,
max_token_length: Option<u32>,
lower_case: Option<bool>,
stem: Option<bool>,
remove_stop_words: Option<bool>,
ascii_folding: Option<bool>,
ngram_min_length: Option<u32>,
ngram_max_length: Option<u32>,
prefix_only: Option<bool>,
) -> napi::Result<Vec<FtsToken>> {
let mut opts = FtsIndexBuilder::default();
if let Some(base_tokenizer) = base_tokenizer {
opts = opts.base_tokenizer(base_tokenizer);
}
if let Some(language) = language {
opts = opts.language(&language).map_err(|_| {
napi::Error::from_reason(format!(
"LanceDB does not support the requested language: '{}'",
language
))
})?;
}
if let Some(max_token_length) = max_token_length {
opts = opts.max_token_length(Some(max_token_length as usize));
}
if let Some(lower_case) = lower_case {
opts = opts.lower_case(lower_case);
}
if let Some(stem) = stem {
opts = opts.stem(stem);
}
if let Some(remove_stop_words) = remove_stop_words {
opts = opts.remove_stop_words(remove_stop_words);
}
if let Some(ascii_folding) = ascii_folding {
opts = opts.ascii_folding(ascii_folding);
}
if let Some(ngram_min_length) = ngram_min_length {
opts = opts.ngram_min_length(ngram_min_length);
}
if let Some(ngram_max_length) = ngram_max_length {
opts = opts.ngram_max_length(ngram_max_length);
}
if let Some(prefix_only) = prefix_only {
opts = opts.ngram_prefix_only(prefix_only);
}
Ok(lancedb_tokenize(&query, &opts)
.default_error()?
.into_iter()
.map(FtsToken::from)
.collect())
}
#[napi]
impl Index {
#[napi(factory)]
+21 -56
View File
@@ -19,7 +19,6 @@ use lancedb::index::scalar::{
BooleanQuery, BoostQuery, FtsQuery, FullTextSearchQuery, MatchQuery, MultiMatchQuery, Occur,
Operator, PhraseQuery,
};
use lancedb::query::AnalyzePlanDistributedMetrics;
use lancedb::query::ExecutableQuery;
use lancedb::query::Query as LanceDbQuery;
use lancedb::query::QueryBase;
@@ -48,28 +47,6 @@ impl From<ColumnOrdering> for LanceDbColumnOrdering {
}
}
fn analyze_plan_options(
distributed_metrics: Option<String>,
) -> napi::Result<QueryExecutionOptions> {
let analyze_plan_distributed_metrics =
match distributed_metrics.as_deref().unwrap_or("aggregate") {
"aggregate" => AnalyzePlanDistributedMetrics::Aggregate,
"per_worker" => AnalyzePlanDistributedMetrics::PerWorker,
"full" => AnalyzePlanDistributedMetrics::Full,
mode => {
return Err(napi::Error::from_reason(format!(
"Invalid distributedMetrics value '{}'. Expected one of: \
'aggregate', 'per_worker', 'full'",
mode
)));
}
};
let mut options = QueryExecutionOptions::default();
options.analyze_plan_distributed_metrics = analyze_plan_distributed_metrics;
Ok(options)
}
fn bytes_to_arrow_array(data: Uint8Array, dtype: String) -> napi::Result<Arc<dyn Array>> {
let buf = arrow_buffer::Buffer::from(data.to_vec());
let num_bytes = buf.len();
@@ -223,17 +200,13 @@ impl Query {
}
#[napi(catch_unwind)]
pub async fn analyze_plan(&self, distributed_metrics: Option<String>) -> napi::Result<String> {
let options = analyze_plan_options(distributed_metrics)?;
self.inner
.analyze_plan_with_options(options)
.await
.map_err(|e| {
napi::Error::from_reason(format!(
"Failed to execute analyze plan: {}",
convert_error(&e)
))
})
pub async fn analyze_plan(&self) -> napi::Result<String> {
self.inner.analyze_plan().await.map_err(|e| {
napi::Error::from_reason(format!(
"Failed to execute analyze plan: {}",
convert_error(&e)
))
})
}
}
@@ -439,17 +412,13 @@ impl VectorQuery {
}
#[napi(catch_unwind)]
pub async fn analyze_plan(&self, distributed_metrics: Option<String>) -> napi::Result<String> {
let options = analyze_plan_options(distributed_metrics)?;
self.inner
.analyze_plan_with_options(options)
.await
.map_err(|e| {
napi::Error::from_reason(format!(
"Failed to execute analyze plan: {}",
convert_error(&e)
))
})
pub async fn analyze_plan(&self) -> napi::Result<String> {
self.inner.analyze_plan().await.map_err(|e| {
napi::Error::from_reason(format!(
"Failed to execute analyze plan: {}",
convert_error(&e)
))
})
}
}
@@ -522,17 +491,13 @@ impl TakeQuery {
}
#[napi(catch_unwind)]
pub async fn analyze_plan(&self, distributed_metrics: Option<String>) -> napi::Result<String> {
let options = analyze_plan_options(distributed_metrics)?;
self.inner
.analyze_plan_with_options(options)
.await
.map_err(|e| {
napi::Error::from_reason(format!(
"Failed to execute analyze plan: {}",
convert_error(&e)
))
})
pub async fn analyze_plan(&self) -> napi::Result<String> {
self.inner.analyze_plan().await.map_err(|e| {
napi::Error::from_reason(format!(
"Failed to execute analyze plan: {}",
convert_error(&e)
))
})
}
}
-4
View File
@@ -232,10 +232,6 @@ impl From<ClientConfig> for lancedb::remote::ClientConfig {
tls_config: config.tls_config.map(Into::into),
header_provider: None, // the header provider is set separately later
user_id: config.user_id,
// Resolved from LANCE_CLIENT_MAX_BYTES_PER_REQUEST or the default.
max_bytes_per_request: None,
// Resolved from LANCE_CLIENT_MAX_REQUEST_DURATION or the read timeout.
max_request_duration: None,
}
}
}
+2 -65
View File
@@ -8,8 +8,8 @@ use chrono::{DateTime, Utc};
use lancedb::ipc::{ipc_file_to_batches, ipc_file_to_schema};
use lancedb::table::{
AddDataMode, ColumnAlteration as LanceColumnAlteration, Duration,
FieldMetadataUpdate as LanceFieldMetadataUpdate, FtsToken as LanceDbFtsToken,
NewColumnTransform, OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable,
FieldMetadataUpdate as LanceFieldMetadataUpdate, NewColumnTransform, OptimizeAction,
OptimizeOptions, Ref, Table as LanceDbTable,
};
use napi::bindgen_prelude::*;
use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode};
@@ -574,27 +574,6 @@ impl Table {
.collect::<Vec<_>>())
}
#[napi(catch_unwind)]
pub async fn tokenize(
&self,
query: String,
column: Option<String>,
index_name: Option<String>,
) -> napi::Result<Vec<FtsToken>> {
let table = self.inner_ref()?;
let tokens = match (column.as_deref(), index_name.as_deref()) {
(Some(_), Some(_)) | (None, None) => {
return Err(napi::Error::from_reason(
"Specify exactly one of 'column' or 'indexName'",
));
}
(Some(column), None) => table.tokenize_with_column(&query, column).await,
(None, Some(index_name)) => table.tokenize(&query, index_name).await,
}
.default_error()?;
Ok(tokens.into_iter().map(FtsToken::from).collect())
}
#[napi(catch_unwind)]
pub async fn index_stats(&self, index_name: String) -> napi::Result<Option<IndexStatistics>> {
let tbl = self.inner_ref()?;
@@ -702,24 +681,6 @@ impl From<lancedb::index::IndexConfig> for IndexConfig {
}
}
#[napi(object)]
/// A token produced by the tokenizer configured on a full-text search index.
pub struct FtsToken {
/// The token text after the index tokenizer has applied its filters.
pub text: String,
/// The token position used by full-text query matching.
pub position: u32,
}
impl From<LanceDbFtsToken> for FtsToken {
fn from(token: LanceDbFtsToken) -> Self {
Self {
text: token.text,
position: token.position,
}
}
}
/// Specification selecting Lance's MemWAL LSM-style write path for
/// `mergeInsert`.
///
@@ -1355,28 +1316,4 @@ impl Branches {
pub async fn delete(&self, name: String) -> napi::Result<()> {
self.inner.delete_branch(&name).await.default_error()
}
#[napi(ts_return_type = "Promise<Record<string, unknown>>")]
pub async fn diff(&self, from_branch: String) -> napi::Result<serde_json::Value> {
let diff = self.inner.diff_branch(&from_branch).await.default_error()?;
serde_json::to_value(diff).map_err(|err| {
napi::Error::from_reason(format!("failed to serialize branch diff: {err}"))
})
}
#[napi(ts_return_type = "Promise<Record<string, unknown>>")]
pub async fn merge(
&self,
from_branch: String,
dry_run: Option<bool>,
) -> napi::Result<serde_json::Value> {
let result = self
.inner
.merge_branch(&from_branch, dry_run.unwrap_or(false))
.await
.default_error()?;
serde_json::to_value(result).map_err(|err| {
napi::Error::from_reason(format!("failed to serialize branch merge result: {err}"))
})
}
}
@@ -1,21 +0,0 @@
{
"name": "lancedb",
"description": "Write, review, debug, and document LanceDB pipelines in Python and TypeScript that work across local LanceDB OSS tables and remote LanceDB Enterprise/Cloud tables, with idiomatic query/search patterns and performance defaults for ingestion, indexing, filtering, and diagnostics.",
"version": "0.1.0",
"author": {
"name": "LanceDB"
},
"homepage": "https://www.lancedb.com",
"keywords": [
"lancedb",
"vector-search",
"full-text-search",
"hybrid-search",
"python",
"typescript",
"pipelines",
"ingestion",
"indexing",
"performance"
]
}
-33
View File
@@ -1,33 +0,0 @@
{
"name": "lancedb",
"version": "0.1.0",
"description": "Codex plugin for building LanceDB pipelines in Python and TypeScript.",
"author": {
"name": "LanceDB"
},
"keywords": [
"lancedb",
"vector-search",
"full-text-search",
"hybrid-search",
"python",
"typescript",
"pipelines"
],
"skills": "./skills/",
"interface": {
"displayName": "LanceDB",
"shortDescription": "Build LanceDB pipelines in Python and TypeScript.",
"longDescription": "Write, review, debug, and document LanceDB pipelines in Python and TypeScript that work across local LanceDB OSS tables and remote LanceDB Enterprise/Cloud tables, with idiomatic query/search patterns and performance defaults for ingestion, indexing, filtering, and diagnostics.",
"developerName": "LanceDB",
"websiteURL": "https://www.lancedb.com",
"category": "Developer Tools",
"capabilities": [
"Developer Tools"
],
"defaultPrompt": "Create a LanceDB table, embed sample text, and run a vector search.",
"composerIcon": "./assets/logo.png",
"logo": "./assets/logo.png",
"logoDark": "./assets/logo-dark.png"
}
}
Binary file not shown.

Before

Width:  |  Height:  |  Size: 10 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 35 KiB

-87
View File
@@ -1,87 +0,0 @@
---
name: lancedb
description: Use when writing, reviewing, debugging, or documenting LanceDB pipelines in Python or TypeScript, especially code that should work across local LanceDB OSS tables and remote LanceDB Enterprise/Cloud tables. Helps avoid non-portable full-table materialization, choose idiomatic query/search patterns, apply LanceDB performance defaults for ingestion, indexing, filtering, and diagnostics, and resolve connections to the remote server for Enterprise-only operations such as jobs.
---
# Building LanceDB Pipelines
Use this skill to produce LanceDB pipelines that are portable between local and remote tables (for LanceDB Enterprise/Cloud) and idiomatic for the selected SDK.
## LanceDB Table Modes
LanceDB has two common execution modes:
- **Local table**: embedded, open source, in-process LanceDB. The client opens data from a local path or object storage URI and executes queries in the application process.
- **Remote table**: LanceDB Enterprise/Cloud table opened through a `db://...` URI. The data may be very large, commonly backed by object storage, and queried through a remote service.
Do NOT assume local-only table helpers exist on remote tables. If the user asks for LanceDB Enterprise, Cloud, `db://...`, production remote access, or a remote table, focus on the remote table path: use `search()` / `query()`, keep reads bounded with `select()` and `limit()`, and avoid table-level full materialization APIs.
## Workflow
1. Identify the SDK: Python, TypeScript, or both.
2. Identify the table mode: local/embedded OSS, remote Enterprise/Cloud, or portable across both. If the user says "LanceDB Enterprise", choose the remote table path. If the task involves jobs in any way (listing, inspecting, creating, or canceling jobs), it is always the remote path and requires a remote server connection — see "Connecting to the LanceDB remote server" below before doing anything else.
3. Read the matching language branch before writing or changing code:
- Python patterns: `references/python/patterns.md`
- Python API quick reference: `references/python/api_reference.md`
- Python performance guidance: `references/python/performance.md`
- TypeScript patterns: `references/typescript/patterns.md`
- TypeScript API quick reference: `references/typescript/api_reference.md`
- TypeScript performance guidance: `references/typescript/performance.md`
- Column metadata authoring (both SDKs): `references/column_metadata.md`
- Branch operations (both SDKs): `references/branch_ops.md`
- Remote server connection resolution (jobs, raw REST): `references/remote_connect.md`
- Job operations REST API (list/describe/cancel/query_events): `references/remote_jobs.md`
4. Start with `patterns.md` for the selected SDK. Read `api_reference.md` when choosing method names or return collectors. Read `performance.md` when the task involves ingestion, indexing, filtering, query tuning, diagnostics, or large datasets. Read `column_metadata.md` when the task is documenting, tagging, classifying, or grouping table columns (field descriptions, `lancedb:tag:*` tags, logical column families). Read `branch_ops.md` when the task involves branch lifecycle (list/create/delete), writing to a non-main branch, or verifying a change stayed off main. Read `remote_connect.md` when the task involves jobs or direct REST access to an Enterprise deployment, and `remote_jobs.md` for the job REST methods themselves (list, describe, cancel, query_events).
5. For Python schemas, favor Pydantic models and validate records before writing. Use PyArrow schemas when Arrow-native, streaming, or highly dynamic data makes them materially better suited.
6. Prefer `search()` or `query()` builders with explicit `select()` and `limit()` for reads.
7. Avoid table-level full materialization in remote or portable code. This is the main local-vs-remote read pitfall.
8. After a successful embedded OSS ingestion, call `table.optimize()`. Do not call it for Enterprise/Cloud; remote maintenance is automatic.
9. For remote Enterprise/Cloud writes, never drop-then-reuse or `mode="overwrite"` the same table name — see "Enterprise: never drop-then-reuse the same table name" below. This is the main local-vs-remote write pitfall.
10. If reviewing an existing file or repo, run `scripts/check_materialization.py` on the relevant paths and inspect each finding before editing.
11. Cross-check unfamiliar or non-trivial API claims against the source tree instead of relying on memory.
## Core Portability Rule
Do not write code that assumes a local table API will exist on a remote table. Remote tables can be very large, so whole-table materialization helpers are intentionally unavailable or unsafe.
This does **not** mean result conversion is forbidden. Bounded query/search result collection is normal:
- Python: `table.search(...).select([...]).limit(10).to_pandas()`
- TypeScript: `await table.search(...).select([...]).limit(10).toArray()`
The unsafe pattern is table-level or unbounded collection, plus local-only dataset escape hatches in remote code:
- Python: `table.to_pandas()`, `table.to_arrow()`, `table.to_polars()`; `table.to_lance()` is local/OSS-only dataset access, not materialization
- TypeScript: `await table.toArrow()`, `await table.query().toArray()` without `limit()`
## Enterprise: never drop-then-reuse the same table name
LanceDB Enterprise/Cloud splits a **control plane** (DDL: create/drop/rename) from a **data plane** (query nodes that serve reads). Query nodes cache the resolved dataset for a table name for up to `table_cache_ttl`**default 300 seconds (5 minutes)**. After you drop or overwrite a table, the control plane updates immediately but the data plane keeps serving the *old* dataset until that cache entry expires. During the window the two planes disagree.
The failure this causes: you `drop_table("t")` then immediately `create_table("t", ...)` (or `create_table("t", ..., mode="overwrite")`). The DDL returns success, but every query against `t` returns **`500 Internal Server Error`** (the query node resolves the stale/deleted dataset), and a fresh `describe` may still show the *old* schema/version. It looks like your write silently failed; it didn't — the name is cached.
**`mode="overwrite"` has the same problem** — it is a drop+create of the same name under the hood.
Rules for portable Enterprise ingestion:
1. **Never reuse a table name you just dropped/overwrote within the cache TTL.** Do not use `mode="overwrite"` to replace an existing Enterprise table in place.
2. To (re)load data, **write to a fresh table name** (e.g. `<table>_v2`, or a run-stamped suffix). A brand-new name has no cached data-plane entry, so writes and reads work immediately.
3. Before creating, `list_tables()` and **fail loudly if the name already exists** rather than overwriting — prompt for a new name.
4. To land on a specific final name that is currently occupied by an old table: drop the old table, **wait out the TTL (~5 min), then `rename_table(fresh_name, final_name)`**. Renaming onto a name whose old dataset is still cached hits the same race, so the wait is mandatory. `rename_table` is a supported control-plane op.
5. When you hand a table name back to a human, tell them which step still needs the propagation wait (usually: "the old `t` was dropped; run the rename in ~5 minutes").
This is Enterprise/Cloud-specific. Local/OSS tables have no separate data plane, so `mode="overwrite"` and immediate same-name reuse are fine there.
## Connecting to the LanceDB remote server
LanceDB Enterprise/Cloud deployments are served by a server implementing the lance-namespace OpenAPI spec (<https://github.com/lance-format/lance-namespace/blob/main/docs/src/spec.yaml>). Every remote (`db://...`) connection talks to such a server, and some operations exist only there. In particular, **all operations around jobs (listing, inspecting, creating, or canceling jobs) run server-side** — there is no local/OSS equivalent. Before any job work, or any direct REST call to an Enterprise deployment, read `references/remote_connect.md` to resolve the base URL, credentials, and database header and to validate the connection. Then use the four job REST methods documented in `references/remote_jobs.md` (list, describe, cancel, query_events).
## Script
Run the scanner when reviewing or modifying an existing codebase:
```bash
python skills/lancedb/scripts/check_materialization.py path/to/file_or_dir
```
The script reports likely unsafe full-table materialization in Python and TypeScript. Treat results as review prompts, not automatic proof of a bug.
@@ -1,6 +0,0 @@
interface:
display_name: "LanceDB"
short_description: "Build LanceDB pipelines in Python and TypeScript"
default_prompt: "Use $lancedb to create a table, embed sample text, and run a vector search."
icon_small: "./assets/icon.png"
icon_large: "./assets/icon.png"
Binary file not shown.

Before

Width:  |  Height:  |  Size: 23 KiB

@@ -1,182 +0,0 @@
# Branch Operations
Manage branches on a LanceDB table: list what exists, create new ones, delete stale ones, and direct read/write operations at a specific branch without touching main. Use for branch lifecycle tasks, experimental/isolated table versions, targeting an operation at a non-main branch, or confirming a mutation did not affect main.
Works on local/OSS and remote Enterprise/Cloud tables, except merging a branch into main, which is Enterprise-only.
## The branch model (important)
Branches are isolated, writable lines of history forked from another branch (or a specific version). Writes on a branch never affect `main`.
There is **no global "switch branch" state** — you never repoint the whole table at a branch. Instead, **operations are scoped by which table handle you use**:
- The handle you got from `open_table(name)` / `openTable(name)` targets `main`.
- `branches.create(...)` and `branches.checkout(...)` return a **new table handle scoped to that branch**. Every read/write on that handle (add, update, `update_field_metadata`, `create_index`, search, …) lands on the branch.
- The original main handle is unaffected — keep it around to verify isolation.
`branches.list()` returns only non-main branches. Main always exists and is not listed.
## Python
`table.branches` is a property returning the branch manager; `table.current_branch()` tells you what a handle is scoped to (`None` = main).
```python
table = db.open_table("products") # scoped to main
# list — dict of name -> metadata (parent_branch, parent_version, ...); {} = only main
table.branches.list()
# create: forks from main by default and returns a handle scoped to the new branch
exp = table.branches.create("experiment-reindex")
exp = table.branches.create("exp2", from_ref="main", from_version=None) # optional fork point
# checkout an existing branch -> branch-scoped handle
wip = table.branches.checkout("wip-branch")
# with version= it pins to that version (read-only detached view); omit to track latest, writable
# operate on the branch simply by using its handle
wip.update_field_metadata(
{"path": "category", "metadata": {"lancedb:description": "Product category label."}}
)
wip.create_scalar_index("category")
# delete: removes only the branch pointer; main and row data remain intact
table.branches.delete("stale-2024")
# alternatively, open a branch handle directly from the connection
wip = db.open_table("products", branch="wip-branch")
exp.current_branch() # "experiment-reindex"
table.current_branch() # None (main)
```
Async: same shape — `table.branches` returns `AsyncBranches`; `await table.branches.create(...)` etc.
## TypeScript
`table.branches()` is an **async method** returning the `Branches` manager; `table.currentBranch()` returns the scoped branch or `null` for main.
```typescript
const table = await db.openTable("products"); // scoped to main
const branches = await table.branches();
// list — Record<string, BranchContents>; {} = only main
await branches.list();
// create: forks from main by default, returns a Table scoped to the new branch
const exp = await branches.create("experiment-reindex");
const exp2 = await branches.create("exp2", "main" /* fromRef */, undefined /* fromVersion */);
// checkout an existing branch -> branch-scoped Table
const wip = await branches.checkout("wip-branch");
// with a version arg it pins (read-only detached view); omit to track latest, writable
// operate on the branch simply by using its handle
await wip.updateFieldMetadata([
{ path: "category", metadata: { "lancedb:description": "Product category label." } },
]);
await wip.createIndex("category");
// delete: removes only the branch pointer; main and row data remain intact
await branches.delete("stale-2024");
// alternatively, open a branch handle directly from the connection
const wip2 = await db.openTable("products", { branch: "wip-branch" });
exp.currentBranch(); // "experiment-reindex"
table.currentBranch(); // null (main)
```
## Verifying isolation
After writing to a branch, confirm the change did NOT land on main by reading through both handles:
```python
wip = table.branches.checkout("wip-branch")
wip.update_field_metadata({"path": "category", "metadata": {"lancedb:description": "..."}})
assert b"lancedb:description" in (wip.schema.field("category").metadata or {})
assert b"lancedb:description" not in (table.schema.field("category").metadata or {}) # main untouched
```
Two handles on the same branch see each other's writes (e.g. `table.branches.create("exp")` and `db.open_table(name, branch="exp")`); main stays isolated.
## Merging a branch into main (Enterprise only)
Merge is available through the SDKs (`table.branches.merge(...)`) on **Enterprise tables only** — it is not supported on Cloud or local/OSS tables, which raise `NotSupported`.
`merge` takes the branch to merge **from** and a `dry_run` flag. Both the SDK method and the underlying REST endpoint **actually merge by default** (`dry_run=False`); pass `dry_run=True` to only preview. A rejected merge is **not an exception** — it returns a result with `status="rejected"` rather than raising, so inspect the return value. Use `branches.diff(from_branch)` to inspect a branch's pending diff without attempting a merge.
```python
exp = "experiment-reindex"
# preview only — returns status="ready" if it would merge cleanly
preview = table.branches.merge(exp, dry_run=True)
# actually merge (default)
result = table.branches.merge(exp)
if result["status"] == "merged":
print("landed at", result["mainVersionAfter"])
elif result["status"] == "rejected":
print(result["diff"]["mergeBlockers"]) # why it was refused
# inspect a branch's pending diff without merging
diff = table.branches.diff(exp)
```
Async: `await table.branches.merge(exp)`, `await table.branches.diff(exp)`.
```typescript
const branches = await table.branches();
const exp = "experiment-reindex";
// preview only (second arg is dryRun)
const preview = await branches.merge(exp, true);
// actually merge (default)
const result = await branches.merge(exp);
if (result.status === "merged") {
console.log("landed at", result.mainVersionAfter);
} else if (result.status === "rejected") {
console.log(result.diff.mergeBlockers);
}
const diff = await branches.diff(exp);
```
The result is the wire JSON, containing `status` (`ready` on a passing dry run, `merged` on success, `rejected` when refused — also `notImplemented`/`unknown`), the branch `diff` (including `mergeBlockers` explaining any rejection), a `preview` of the columns that would be promoted, and — after a real merge — `mainVersionAfter`.
### Merge preconditions
Merge only **promotes newly added columns** onto main; it does not replay arbitrary commits. Practically, a branch is mergeable only if it has **exactly one commit since it was created, and that commit added a column**. The merge is rejected (`status: "rejected"`, with `mergeBlockers` set) if:
- the branch was forked from another branch rather than directly from main
- main has advanced since the branch was forked
- the branch's rows changed since the fork (row counts must match main exactly)
- the branch removed columns or changed a column's type/nullability
- the branch added no columns (index-only changes are not merged)
### Adding a column in a single commit
Because the branch must contain just one column-adding commit, add the column with its values in one operation rather than add-then-backfill:
1. **SQL transformation**`add_columns` with a SQL expression computed from existing columns, so the column lands populated in one commit.
2. **Precompute the values** — compute the column's values externally, then add the fully-populated column in a single operation (e.g. via `merge_insert`/`add_columns` with the data ready).
3. **Lance-format-level data evolution (pylance)** — use Lance's data evolution with backfill, documented at <https://lance.org/guide/data_evolution/#with-data-backfill>.
## Quick reference
| Goal | Python | TypeScript |
|------|--------|------------|
| List branches (non-main) | `table.branches.list()` | `await (await table.branches()).list()` |
| Create branch (off main) | `table.branches.create(name)` → branch handle | `await branches.create(name)` → branch `Table` |
| Create from a fork point | `table.branches.create(name, from_ref=..., from_version=...)` | `await branches.create(name, fromRef, fromVersion)` |
| Get a branch handle | `table.branches.checkout(name)` or `db.open_table(t, branch=name)` | `await branches.checkout(name)` or `await db.openTable(t, { branch: name })` |
| Pin to a branch version (read-only) | `table.branches.checkout(name, version=v)` | `await branches.checkout(name, v)` |
| Delete branch | `table.branches.delete(name)` | `await branches.delete(name)` |
| Which branch is this handle on? | `table.current_branch()` (`None` = main) | `table.currentBranch()` (`null` = main) |
| Target main | use the original (non-branch) handle | use the original (non-branch) handle |
| Merge branch into main (Enterprise only) | `table.branches.merge(from_branch, dry_run=False)` | `await branches.merge(fromBranch, dryRun)` |
| Preview a branch's pending diff (Enterprise only) | `table.branches.diff(from_branch)` | `await branches.diff(fromBranch)` |
Branch names must be non-empty; empty names raise a validation error.
@@ -1,183 +0,0 @@
# Column Metadata Authoring
Write column-level descriptions, tags, and logical groupings onto a LanceDB table's schema. Use this when the user wants to document, annotate, tag, or classify what their table columns ARE (embeddings vs labels vs eval metrics, model provenance, version families, etc.).
Works on local/OSS and remote Enterprise/Cloud tables alike — read the schema through the table handle, write through `update_field_metadata` (Python) / `updateFieldMetadata` (TypeScript).
## Metadata key conventions
All metadata uses namespaced keys:
| Key | Purpose | Example value |
|-----|---------|---------------|
| `lancedb:description` | Human-readable explanation of what the column contains | `"CLIP ViT-L/14 image embedding, L2-normalized (768-dim)"` |
| `lancedb:tag:<name>` | Flexible key-value tag; the suffix names the tag category | `lancedb:tag:field_type: "embedding"`, `lancedb:tag:model: "clip"`, `lancedb:tag:project_id: "foo"` |
| `lancedb:logical-column` | Logical group/family this column belongs to | `"clip_features"` |
Tags are open-ended — use whatever key suffix and value make sense given the user's intent. The tag suffix should describe *what is being classified* (e.g., `field_type`, `model`, `project_id`) and the value describes *how*. Multiple tags on the same column are fine — each is a separate key. All values are strings.
## Step 1: Read the schema and existing metadata
Read existing metadata before writing, to avoid redundant updates.
Python — `table.schema` (sync property; async: `await table.schema()`) returns a `pyarrow.Schema`. **Arrow field metadata is bytes-keyed in Python**:
```python
schema = table.schema
for field in schema:
meta = field.metadata or {} # dict[bytes, bytes], e.g. {b"lancedb:description": b"..."}
print(field.name, field.type, field.nullable, meta)
```
TypeScript — `await table.schema()` returns an Arrow `Schema`; field metadata is a `Map<string, string>`:
```typescript
const schema = await table.schema();
for (const field of schema.fields) {
console.log(field.name, field.type, field.nullable, field.metadata); // Map
// field.metadata.get("lancedb:description")
}
```
For struct/nested fields, recurse into the field's children and address them as dot-paths (e.g., `parent.child`).
If the user hasn't specified which columns to update, work with all columns.
## Step 2: Generate metadata
Decide what to generate based on the user's request.
### Descriptions (`lancedb:description`)
Base descriptions on:
- The column name and Arrow type (e.g., `FixedSizeList` of floats → likely an embedding)
- User-supplied context (upstream pipeline, sample values, domain knowledge)
- Name patterns: `_embedding`/`_vec`/`_embed` → vector; `_label`/`_class` → label; `_score`/`_eval`/`_metric` → evaluation metric
Be specific and concise. Good: `"Sentence-BERT embedding of the query text (768-dim)."` Not: `"An embedding column."`
### Tags (`lancedb:tag:<name>`)
Choose tag key names that match what the user asked to annotate. Common patterns:
- Semantic field type → `lancedb:tag:field_type: "embedding"` / `"text"` / `"image"` / `"label"` / `"eval"` / `"id"` / `"metadata"`
- Model or source → `lancedb:tag:model: "clip"` / `"bert"` / `"vit"`
- Project affiliation → `lancedb:tag:project_id: "<name>"`
- Version → `lancedb:tag:version: "v3"` (and `lancedb:tag:latest: "true"` for the newest)
Use Arrow type as a hint: `FixedSizeList` + float → embedding; `Utf8`/`LargeUtf8` → text; `Binary` → image or blob.
### Logical groupings (`lancedb:logical-column`)
Look for naming patterns across columns:
- `clip_v1`, `clip_v2`, `clip_v3` → logical column `"clip"`, latest is `v3`
- `text_embed_20240101`, `text_embed_20240601` → logical column `"text_embed"`, latest is the most recent date suffix
Write `lancedb:logical-column` on all members of a group. Mark the newest with `lancedb:tag:latest: "true"` (in addition to its version tag).
## Step 3: Write the metadata
Each update names a field by dot-path and carries a metadata map. Semantics (identical in both SDKs):
- **Merge by default** (`replace` omitted/false) — preserves existing metadata the user didn't ask to change
- `replace: true` swaps the field's entire metadata map — only if the user explicitly asks to overwrite
- A value of `None`/`null` deletes that specific key
- Batch all field updates into a single call when possible
- Returns the new table version
Python (sync and async take one dict per field, as varargs):
```python
res = table.update_field_metadata(
{
"path": "clip_v3",
"metadata": {
"lancedb:description": "CLIP ViT-L/14 image embedding, L2-normalized (1024-dim).",
"lancedb:tag:field_type": "embedding",
"lancedb:tag:model": "clip",
"lancedb:tag:version": "v3",
"lancedb:tag:latest": "true",
"lancedb:logical-column": "clip",
},
},
{
"path": "clip_v2",
"metadata": {
"lancedb:description": "CLIP ViT-B/32 image embedding (768-dim), superseded by v3.",
"lancedb:tag:field_type": "embedding",
"lancedb:tag:model": "clip",
"lancedb:tag:version": "v2",
"lancedb:logical-column": "clip",
},
},
)
print(res.version) # new table version
# merge semantics: add a key, delete one via None, keep the rest
table.update_field_metadata(
{"path": "clip_v2", "metadata": {"lancedb:tag:archived": "true", "lancedb:tag:latest": None}}
)
```
(`replace_field_metadata` is deprecated — use `update_field_metadata`.)
TypeScript (takes an array of `FieldMetadataUpdate`):
```typescript
const res = await table.updateFieldMetadata([
{
path: "clip_v3",
metadata: {
"lancedb:description": "CLIP ViT-L/14 image embedding, L2-normalized (1024-dim).",
"lancedb:tag:field_type": "embedding",
"lancedb:tag:model": "clip",
"lancedb:tag:version": "v3",
"lancedb:tag:latest": "true",
"lancedb:logical-column": "clip",
},
},
{
path: "clip_v2",
metadata: {
"lancedb:description": "CLIP ViT-B/32 image embedding (768-dim), superseded by v3.",
"lancedb:tag:field_type": "embedding",
"lancedb:tag:model": "clip",
"lancedb:tag:version": "v2",
"lancedb:logical-column": "clip",
},
},
]);
console.log(res.version); // new table version
// merge semantics: add a key, delete one via null, keep the rest
await table.updateFieldMetadata([
{ path: "clip_v2", metadata: { "lancedb:tag:archived": "true", "lancedb:tag:latest": null } },
]);
```
## Step 4: Confirm
Report back:
- Which columns were updated and what was written
- The new table version number (from the result)
- Any columns skipped (e.g., already had up-to-date metadata)
## Quick examples
**"Write descriptions for all columns in the `product_embeddings` table"**
1. Read `table.schema` → all fields + existing metadata
2. Generate a `lancedb:description` for each column based on name + type
3. One `update_field_metadata` call with all descriptions
4. Report
**"Tag the columns in `model_outputs` with their field type and model"**
1. Read the schema
2. For each field, classify by name + Arrow type → set `lancedb:tag:field_type` and `lancedb:tag:model` where applicable
3. Write in one batched call
4. Report
**"Group the feature columns in `training_features` into logical families and mark the latest version"**
1. Read the schema
2. Find version patterns → assign `lancedb:logical-column` and `lancedb:tag:version`; mark newest with `lancedb:tag:latest: "true"`
3. Write in one batched call
4. Show the grouping
@@ -1,138 +0,0 @@
# Python API Reference
Quick method reference for Python LanceDB code. Cross-check source for non-trivial claims.
## Connect
If you're connecting to a remote database, use this:
```python
import lancedb
db = lancedb.connect("db://my-db", api_key=api_key, host_override=host_override) # remote
```
(values may be found in LANCEDB_API_KEY and LANCEDB_HOST_OVERRIDE, either in env vars or a .env file)
If you're connecting to a local table using OSS LanceDB, use this:
```python
db = lancedb.connect("./camelot-db") # local/OSS
```
If you're not sure which, or if you can't find the api_key or host_override params, ask the user.
**Place the local database directory next to the script/entrypoint that opens it** (i.e. resolve the path relative to the script, `Path(__file__).parent / "camelot-db"`), not buried under a shared `data/` folder. The Lance dataset is the database, not a data file — keeping it beside its code makes ownership obvious and paths stable regardless of the working directory the script is launched from.
**Do not name the directory `lancedb`** (e.g. `./lancedb`, `./data/lancedb`). It collides with the imported `lancedb` package name, which is confusing to read and easy to shadow in scripts. Give it a name derived from the repo or dataset with a clear prefix/suffix — for example `./<dataset>-db`, `./<repo>_lancedb`, or `./vectordb`.
Async:
```python
db = await lancedb.connect_async("./camelot-db")
```
## Table Reads
| Task | Preferred API |
| --- | --- |
| Vector search | `table.search(query_vector).limit(k)` |
| Full scan with filters/projection (sync) | `table.search().where(...).select(...).limit(...)` |
| Full scan with filters/projection (async) | `table.query().where(...).select(...).limit(...)` |
| Filter | `.where("col > 10")` |
| Projection | `.select(["id", "text"])` |
| Bound result count | `.limit(20)` |
| Collect bounded result as Python objects (default, no extra deps) | `.to_list()` on query/search result |
| Collect bounded result as Arrow (default, `pyarrow` always available) | `.to_arrow()` on query/search result |
| Collect bounded result as pandas (only if project uses pandas) | `.to_pandas()` on query/search result |
| Collect bounded result as Polars (only if project uses polars) | `.to_polars()` on query/search result |
## Sync vs Async Scan API
The plain-scan entry point differs between the sync and async clients. **Verified against `lancedb` 0.34.0** — re-check if the pinned version changes:
- **Sync** (`lancedb.connect(...)`): the table has **no `.query()` method**. Use `.search()` with no argument for a plain scan; it returns a query builder that supports `.where()`, `.select()`, `.limit()`, and the `.to_list()` / `.to_arrow()` / `.to_pandas()` / `.to_polars()` collectors.
```python
rows = table.search().where("status = 'ready'").select(["id", "text"]).limit(20).to_list()
```
- **Async** (`lancedb.connect_async(...)`): the table has **both** `.query()` and `.search()`. Use `.query()` for a plain scan.
```python
rows = await async_table.query().where("status = 'ready'").select(["id", "text"]).limit(20).to_list()
```
Do not call `table.query()` on a sync table — it raises `AttributeError`.
## Local vs Remote Table Methods
| API | Local table | Remote table | Agent guidance |
| --- | --- | --- | --- |
| `table.search(...)` | Yes | Yes | Preferred read path (sync + async) |
| `table.query()` | Async only | Async only | Sync scan path is `table.search()`; `.query()` is the async scan builder |
| `table.to_pandas()` | Yes | No / unsafe for portability | Avoid in portable code |
| `table.to_arrow()` | Yes | No / unsafe for portability | Avoid in portable code |
| `table.to_polars()` | Yes | No / unsafe for portability | Avoid in portable code |
| `table.to_lance()` | Yes | No | Local/OSS escape hatch only |
## Indexes
Use `create_index(...)` for vector indexes and modern index configs. Use scalar indexes for filtered or merge keys.
Common calls:
```python
table.create_index("vector")
table.create_scalar_index("status")
table.create_fts_index("text")
```
Check source docs before specifying advanced index config names or parameters.
## Filtering And Recall Knobs
```python
table.search(query_vector).where("status = 'ready'") # pre-filter by default
table.search(query_vector).where("status = 'ready'", prefilter=False)
table.search(query_vector).limit(10).refine_factor(20)
table.search(query_vector).limit(10).nprobes(50)
```
Use post-filtering only when fewer than `limit` results are acceptable.
## Diagnostics
```python
print(table.search(query_vector).where("year > 2000").limit(10).analyze_plan())
print(table.index_stats("vector_idx"))
```
Use these before changing indexes or search tuning.
## Column (Field) Metadata
```python
schema = table.schema # sync property; async: await table.schema()
meta = schema.field("category").metadata # dict[bytes, bytes] — Arrow metadata is bytes-keyed
res = table.update_field_metadata( # varargs: one dict per field; works local + remote
{"path": "category", "metadata": {"lancedb:description": "...", "lancedb:tag:field_type": "label"}}
)
res.version # new table version
```
Merges by default; a `None` value deletes that key; `"replace": True` swaps the whole map. Nested fields use dot-paths (`"a.b.c"`). `replace_field_metadata` is deprecated. See `references/column_metadata.md` for key conventions (`lancedb:description`, `lancedb:tag:<name>`, `lancedb:logical-column`) and the authoring workflow.
## Branches
```python
table.branches.list() # non-main branches; {} = only main
exp = table.branches.create("exp") # fork off main -> handle scoped to the branch
wip = table.branches.checkout("wip") # existing branch -> scoped handle (version= pins read-only)
wip = db.open_table("t", branch="wip") # or open scoped directly
table.branches.delete("stale") # removes only the branch pointer
table.current_branch() # None = main
```
There is no global switch — scoping is per table handle: any read/write on a branch handle lands on that branch; the original handle keeps targeting main. See `references/branch_ops.md` for the model and isolation checks.
## Maintenance
```python
table.optimize()
```
Call this after every successful local/OSS ingestion. It handles compaction, cleanup of old versions according to retention, and index optimization. Do not add this for LanceDB Enterprise/Cloud remote tables; Enterprise handles compaction and cleanup automatically from cluster configuration.
@@ -1,173 +0,0 @@
# Python Patterns
Use these patterns when writing Python code with `lancedb`.
## Before Writing Code
Choose the output type from what the project actually depends on. **Do not assume `pandas` or `polars` is installed** — they are heavy dependencies that many LanceDB projects do not use. `pyarrow`, by contrast, ships as a LanceDB dependency and is always available, so it is a safe default to lean on.
Default output (after applying `select()` and `limit()`):
- **Python objects**: `.to_list()` — a list of dicts, no extra dependencies. Prefer this for scripts, examples, and agent-generated code unless there is a reason to do otherwise.
- **PyArrow**: `.to_arrow()` — a `pyarrow.Table`, when the surrounding code is Arrow-native or you need columnar/zero-copy handoff.
Only reach for a DataFrame when the project *already* declares that dependency:
- Pandas projects (pandas in `pyproject.toml`/requirements): `.to_pandas()`.
- Polars projects (polars declared): `.to_polars()`.
If unsure, check the dependency manifest or the imports in surrounding files. When in doubt, use `.to_list()` or `.to_arrow()`.
## Schema Design and Validation
Favor `LanceModel` and Pydantic validation for Python schemas. They keep field
types readable, validate source records before a write, and map directly to a
LanceDB schema. Use `Vector(dimension)` for fixed-size vectors:
```python
from lancedb.pydantic import LanceModel, Vector
class Document(LanceModel):
id: int
text: str
vector: Vector(384, nullable=False)
rows = [Document.model_validate(row) for row in source_rows]
table = db.create_table("documents", schema=Document)
table.add(rows)
```
Use PyArrow schemas instead when the pipeline is already Arrow-native, needs
record-batch streaming, or has runtime schema requirements that would make a
Pydantic model harder to understand. Declare Pydantic as a direct project
dependency when application code imports it, even if LanceDB also depends on it.
## Recommended Patterns
### Bounded search or query
Use this for application reads, examples, notebooks, and agent-generated scripts:
```python
results = (
table.search(query_vector)
.where("status = 'ready'")
.select(["id", "text"])
.limit(20)
.to_list() # or .to_arrow(); .to_pandas()/.to_polars() only if the project uses them
)
```
Why: `search()` works across local and remote tables and on both the sync and async clients. `select()` avoids fetching unused columns. `limit()` prevents accidental full-table reads. `.to_list()` and `.to_arrow()` avoid assuming pandas/polars is installed (see "Before Writing Code").
For a **plain scan** (no query vector), the entry point differs by client:
```python
# Sync client: no .query() method — use .search() with no argument.
rows = table.search().where("status = 'ready'").select(["id", "text"]).limit(20).to_list()
# Async client: use .query().
rows = await async_table.query().where("status = 'ready'").select(["id", "text"]).limit(20).to_list()
```
`table.query()` on a sync table raises `AttributeError` (verified on `lancedb` 0.34.0). See the "Sync vs Async Scan API" section in `api_reference.md`.
### Bounded query result conversion
It is fine to collect bounded query/search results:
```python
arrow_table = table.search().select(["id"]).limit(100).to_arrow() # sync plain scan
rows = table.search(query_vector).limit(10).to_list()
df = table.search(query_vector).limit(10).to_pandas() # only if pandas is a project dep
```
### Local-only Lance dataset API
`table.to_lance()` does not itself materialize the full dataset. It returns the underlying `lance.LanceDataset`, making the table accessible through the PyLance dataset API. Use it when the task is explicitly local/OSS and needs Lance dataset methods not exposed by LanceDB:
```python
# Local/OSS only: RemoteTable does not expose table.to_lance().
ds = table.to_lance()
for batch in ds.to_batches(columns=["id", "text"], batch_size=10_000):
process(batch)
```
### Async Python
Keep the same shape and bound the result before collecting:
```python
results = await (
async_table.query()
.where("status = 'ready'")
.select(["id", "text"])
.limit(20)
.to_list() # or .to_arrow()
)
```
## Anti-Patterns
**Avoid the following anti-patterns in your code.**
### Table-level full materialization
Avoid whole-table collectors in portable or large-table code:
```python
df = table.to_pandas()
arrow_table = table.to_arrow()
polars_df = table.to_polars()
```
Why: local tables expose these whole-table collectors, but remote tables intentionally do not — a remote production table can be far larger than a local development table, so it is easy to accidentally pull the entire table into memory.
`table.to_lance()` is different: it is not a full materialization call, but it is still local/OSS-only and should not appear in code meant to run against remote Enterprise tables.
### Unbounded result collection
Avoid query/search collection without a meaningful limit:
```python
rows = table.search().to_list() # unbounded plain scan
rows = table.search(query_vector).to_list() # unbounded vector search
```
Prefer `select(...).limit(...)` before collecting; for large reads, stream in batches instead.
### Per-row writes
Avoid loops that write one row per call:
```python
for row in rows:
table.add([row]) # one commit + fragment per row
```
Each `add()` creates a new version and fragment. Pass the whole batch in a single call, or chunk very large inputs:
```python
table.add(rows) # single commit
# for very large inputs, add batches of several thousand rows
```
After the final successful write to an embedded OSS table, call
`table.optimize()`. Skip this for Enterprise/Cloud tables because their
maintenance is automatic.
### Drop-then-reuse the same table name (Enterprise/Cloud)
Avoid dropping or overwriting a remote table and then reusing that name right away:
```python
db.drop_table("my_table")
table = db.create_table("my_table", data=rows) # reads 500 for ~5 min
table = db.create_table("my_table", data=rows, mode="overwrite") # same problem
```
Why: Enterprise/Cloud splits DDL (control plane) from query serving (data plane). The data plane caches the dataset behind a table name for up to `table_cache_ttl` (default 300s / 5 min), so after a drop/overwrite the DDL succeeds but queries against the reused name return `500 Internal Server Error` until the cache expires — and a fresh `describe` may still show the old schema. Instead, write to a **fresh name**, use `list_tables()` and fail if it already exists, then `rename_table(fresh, final)` onto the final name only after the old table's drop has propagated (~5 min). See the "Enterprise: never drop-then-reuse the same table name" section in `SKILL.md`. Local/OSS tables have no separate data plane — overwrite freely there.
### Guessing performance fixes
Avoid changing `nprobes`, `refine_factor`, or index types before checking the query plan and index stats. Diagnose first, then tune one knob at a time.
@@ -1,131 +0,0 @@
# Python Performance Guidance
Use this when writing Python code that ingests data, queries large tables, builds indexes, or investigates latency.
## Ingestion
### Recommended: validate schemas and records with Pydantic
Favor `LanceModel` for readable Python schema definitions and validate source
records before writing. Use PyArrow directly for Arrow-native or streaming
pipelines where it is the clearer representation.
```python
from lancedb.pydantic import LanceModel, Vector
class Document(LanceModel):
id: int
text: str
vector: Vector(384, nullable=False)
rows = [Document.model_validate(row) for row in source_rows]
table = db.create_table("documents", schema=Document)
table.add(rows)
```
### Recommended: bulk ingestion for materialized data
```python
table.add(arrow_table)
table.add(df)
table.add(pa.dataset("data/", format="parquet"))
```
For very large initial loads, create the table empty first, then call `add(...)`. Passing data directly to `create_table(name, data)` can skip the auto-parallel write path.
### Recommended: iterator ingestion for generated or streamed data
```python
def batches():
for raw in source:
vectors = model.encode(raw["text"])
yield pa.RecordBatch.from_pydict({**raw, "vector": vectors})
table.add(batches())
```
Use chunks of several thousand rows or more when practical. Tiny batches and per-row writes create many small fragments.
### Anti-pattern: per-row `add()`
```python
for row in rows:
table.add([row])
```
Each call creates a version and fragment. This slows ingestion and later queries.
## Indexing
- Build a vector index once brute-force vector search becomes too slow. As a rule of thumb, local brute force is fine below roughly 100K vectors; beyond that, build an index.
- Use `IVF_PQ` as the general-purpose default. Enterprise builds this automatically.
- Use scalar indexes for filtered columns and merge/upsert keys.
- Use `BTREE` for mostly distinct numeric/string/temporal columns, `BITMAP` for booleans and low-cardinality columns, and `LABEL_LIST` for list membership queries.
- Keep full-text defaults unless phrase queries require position data.
## Querying
Always be explicit:
```python
table.search(query_vector).select(["id", "title"]).limit(20)
```
- `select()` reduces bytes read and transferred.
- `limit()` prevents accidental full-table materialization.
- Pre-filtering is the default and guarantees returned rows satisfy the predicate.
- Use post-filtering only when fewer than `limit` results are acceptable.
## Recall Tuning
Tune one knob at a time:
- Quantized indexes: raise `refine_factor` to rescore more candidates on full vectors.
- HNSW-backed indexes: raise `ef`; start around `1.5 * k`, increase toward `10 * k` if recall is short.
- IVF candidate breadth: `nprobes` is auto-tuned; override only when a selective pre-filter leaves too few neighbors.
## Maintenance
After every successful embedded OSS/local ingestion, call `table.optimize()`.
Do not add this to LanceDB Enterprise/Cloud remote table code; remote compaction
and cleanup are handled automatically based on the Enterprise cluster
configuration.
Why local maintenance is needed:
- Frequent writes can create many small fragments. Queries then need to scan across more files, which can increase latency.
- Updates, deletes, and appends create new table versions. Old versions are retained for time travel and rollback, which can grow disk usage.
- Indexes may have newly added rows that are not yet fully optimized into the index structure.
For local/OSS tables, run `optimize()` after the final successful ingestion
write. Also run it after later batches of update/delete operations or on a
regular maintenance schedule:
```python
table.optimize()
```
If the user wants more aggressive local disk cleanup, pass a shorter cleanup retention window:
```python
from datetime import timedelta
table.optimize(cleanup_older_than=timedelta(days=1))
```
Do not use very short cleanup windows when the application depends on time travel, rollback, or old versions.
## Diagnostics
Before changing code or indexes, inspect:
```python
print(table.search(query_vector).where("year > 2000").limit(10).analyze_plan())
print(table.index_stats("vector_idx"))
```
Look for high scan bytes, missing indexes, fragmented data, and unindexed rows.
## Python Multiprocessing
When using multiprocessing, use `spawn` rather than `fork`. LanceDB is multi-threaded internally, and `fork` plus a multi-threaded process is unsafe.
@@ -1,45 +0,0 @@
# Connecting to a LanceDB remote server
LanceDB Enterprise/Cloud deployments are served by a server implementing the
lance-namespace OpenAPI spec
(<https://github.com/lance-format/lance-namespace/blob/main/docs/src/spec.yaml>).
Every remote (`db://...`) connection talks to such a server, and some operations
exist only there. In particular, all operations around jobs (listing, inspecting,
creating, or canceling jobs) run server-side — there is no local/OSS equivalent, so
resolve a server connection before attempting any job work. The job REST methods
themselves are documented in `references/remote_jobs.md`.
Every request needs two things:
1. **Base URL** — the server endpoint
2. **Credentials** — an API key (`x-api-key` header over REST), and usually a database name (`x-lancedb-database` header)
## Resolution steps
1. If the user already gave a URL and API key (or said which environment they're working against), use that.
2. Otherwise, look for credentials already available in the environment:
- Env vars like `LANCEDB_URI` / `LANCEDB_HOST` / `LANCEDB_API_KEY`
- A server endpoint already running or port-forwarded locally (the REST default port is 2333, i.e. `http://localhost:2333`)
3. If you didn't find both pieces, ask the user directly: **"What's your LanceDB endpoint's URL, and what's your API key?"** Also ask which database to use if it isn't obvious. Don't guess or probe further — the user knows their deployment.
## Validating the connection
Make a cheap authenticated request and check the status before starting real work:
```bash
curl -s -w "\n%{http_code}" "{base_url}/v1/table/?limit=1" \
-H "x-api-key: <key>" \
-H "x-lancedb-database: <database>"
```
- `200` — connection, key, and database header all good
- `401` — API key missing or wrong
- `400` mentioning a database header — this deployment expects `x-lancedb-database`
## Non-REST equivalents
The same credentials work through the SDKs and CLI:
- Python SDK: `lancedb.connect("db://<database>", api_key="<key>", host_override="<base_url>")`
- TypeScript SDK: `await lancedb.connect("db://<database>", { apiKey: "<key>", hostOverride: "<base_url>" })`
- `lancedb` CLI: a `[profiles.<name>]` entry in `~/.lancedb/config.toml` with `http_server_url`, `api_key`, `database`
@@ -1,151 +0,0 @@
# Job operations over the LanceDB remote server REST API
Jobs are server-side background operations on LanceDB Enterprise/Cloud — index builds,
column backfills, materialized view refreshes, and similar async work. Endpoints that
trigger async work (e.g. the column backfill or materialized view refresh endpoints)
return a `job_id`; these four methods are how you track and manage those jobs.
Resolve the connection first — see `references/remote_connect.md`. All four methods
are **POST** requests under `{base_url}/v1/jobs/` with JSON bodies, and take the usual
`x-api-key` / `x-lancedb-database` headers. If every job call returns `501`, job APIs
are disabled on that deployment (the server has no job registry configured) — report
that rather than retrying.
## 1. List jobs — `POST /v1/jobs/list`
The body is optional; an empty body lists everything. All fields are filters:
```json
{
"limit": 100,
"table_name": "my_table",
"job_type": "...",
"job_subtype": "...",
"state": "...",
"page_token": "..."
}
```
```bash
curl -s -X POST "{base_url}/v1/jobs/list" \
-H "x-api-key: <key>" -H "x-lancedb-database: <database>" \
-H "content-type: application/json" \
-d '{"table_name": "my_table"}'
```
Response:
```json
{
"jobs": [
{
"job_id": "...",
"table": "my_table",
"job_type": "...",
"job_subtype": "...",
"state": "done",
"created_at_millis": 1720000000000
}
],
"page_token": "..."
}
```
A `page_token` in the response means there are more results — pass it back in the next
request to continue. Note list rows use a lowercase `state` string, while describe uses
an uppercase `job_state`.
## 2. Describe a job — `POST /v1/jobs/describe`
Body: `{"job_id": "<id>"}`. Returns full detail for one job:
```json
{
"job_id": "...",
"job_type": "...",
"job_subtype": "...",
"job_state": "IN_PROGRESS",
"creation_ms": 1720000000000,
"spec": {},
"status": {}
}
```
`job_state` is one of `IN_PROGRESS`, `CANCELLED`, `FAILED`, `DONE`. `spec` and `status`
are job-type-specific JSON objects (the job's input specification and its current
progress/status). Returns `404` for an unknown job id.
## 3. Cancel a job — `POST /v1/jobs/cancel`
Body: `{"job_id": "<id>"}`; response echoes `{"job_id": "<id>"}`. Cancellation is a
service-level operation requiring the same administrative authorization as the
`/admin` routes — a database-scoped API key that can list and describe jobs may still
get a permission error here. Other errors: `404` unknown job, `409` state conflict
(e.g. already in a terminal state), `429` too much write contention (safe to retry).
## 4. Query job event history — `POST /v1/jobs/query_events`
Returns the event history (state transitions, progress updates) for one or more jobs.
Body: `{"job_id": "<id>"}` for one job, or `{"job_ids": ["<id>", ...]}` for a batch.
Optional fields: `limit` (max event rows), `limit_per_job` (per job in a batch query),
and `filter` — a SQL-like expression over the columns `state`, `updated_by`,
`owner_component`, and `claim_entity`. (`full_text_search` is reserved and currently
rejected as not implemented.)
The response is **not JSON** — it is an Arrow IPC stream
(`content-type: application/vnd.apache.arrow.stream`). Decode it, e.g. in Python:
```python
import pyarrow.ipc
import requests
resp = requests.post(
f"{base_url}/v1/jobs/query_events",
headers={"x-api-key": key, "x-lancedb-database": database},
json={"job_id": job_id},
)
resp.raise_for_status()
events = pyarrow.ipc.open_stream(resp.content).read_all()
```
## Feature engineering (Geneva) jobs
Feature engineering jobs — UDF column backfills and materialized view refreshes run
through Geneva — are tracked **separately** from the `/v1/jobs` registry above. Their
records live in a `geneva_jobs` table inside the database itself (in the `__system`
namespace), and you access them through a Python `geneva` connection rather than the
REST endpoints above:
```python
import geneva
from geneva.jobs import JobStateManager
# Same credentials as lancedb.connect / the REST API
conn = geneva.connect("db://<database>", api_key="<key>", host_override="<base_url>")
jsm = JobStateManager(conn)
# List jobs. NOTE: status defaults to "RUNNING"; pass status=None for all jobs.
# Statuses: PENDING | RUNNING | DONE | FAILED | CANCELLED
jobs = jsm.list_jobs(table_name="my_table", status=None)
# Fetch one job by id (returns a list of JobRecord)
records = jsm.get("<job_id>")
```
Each `JobRecord` has `table_name`, `column_name`, `job_id`, `job_type`, `status`,
`launched_at`, `completed_at`, `config`, `launched_by`, `manifest_id`, `cluster_name`,
`metrics` (progress counters), `events` (human-readable history), and `updated_at`.
For filters `list_jobs` doesn't support (e.g. time ranges), query the underlying table
directly: `jsm.get_table(True).search().where("launched_at >= TIMESTAMP '...'")`
pass `True` to check out the latest version, since other processes update job state.
Stale-status caveat: nothing reaps dead Geneva jobs, so a job can sit in
`RUNNING`/`PENDING` forever if its worker died. Treat a job as effectively `FAILED`
when it has been running longer than ~36 hours, or its `updated_at` is more than ~2
hours old (this matches the heuristic the Geneva console UI applies on read).
## Workflow tips
- To wait for async work (a backfill, an index build), poll `describe` until
`job_state` leaves `IN_PROGRESS`; on `FAILED`, pull `status` and `query_events` for
the failure detail.
@@ -1,105 +0,0 @@
# TypeScript API Reference
Quick method reference for TypeScript LanceDB code. Cross-check source for non-trivial claims.
## Connect
```typescript
import * as lancedb from "@lancedb/lancedb";
const db = await lancedb.connect("./camelot-db");
```
**Place the local database directory next to the script/entrypoint that opens it** (resolve the path relative to the module, e.g. via `import.meta.dirname` / `__dirname`), not buried under a shared `data/` folder. The Lance dataset is the database, not a data file — keeping it beside its code makes ownership obvious and paths stable regardless of the working directory the script is launched from.
**Do not name the directory `lancedb`** (e.g. `./lancedb`, `./data/lancedb`). It collides with the imported `lancedb` package/namespace, which is confusing to read. Give it a name derived from the repo or dataset with a clear prefix/suffix — for example `./<dataset>-db`, `./<repo>_lancedb`, or `./vectordb`.
Remote connections use `db://...` plus Enterprise/Cloud credentials and deployment settings. Check current source/docs for exact connection options.
## Table Reads
| Task | Preferred API |
| --- | --- |
| Vector search | `table.search(queryVector).limit(k)` |
| Full scan with filters/projection | `table.query().where(...).select(...).limit(...)` |
| Filter | `.where("col > 10")` |
| Projection | `.select(["id", "text"])` |
| Bound result count | `.limit(20)` |
| Collect bounded result as objects | `.toArray()` on query/search result |
| Collect bounded result as Arrow | `.toArrow()` on query/search result |
| Stream result batches | `for await (const batch of table.query()...)` |
## Local vs Remote Safety
| API | Agent guidance |
| --- | --- |
| `table.search(...)` | Preferred read path |
| `table.query()` | Preferred scan/filter path |
| `await table.toArrow()` | Avoid in portable or large-table code |
| `await table.query().toArray()` with no `limit()` | Avoid; unbounded collection |
| `await table.query().toArrow()` with no `limit()` | Avoid; unbounded collection |
## Indexes
```typescript
await table.createIndex("vector");
await table.createIndex("status");
```
Use vector indexes for large vector search workloads and scalar indexes for filtered columns or merge/upsert keys. Check source/docs before specifying advanced index options.
## Filtering And Recall Knobs
```typescript
await table.search(queryVector).where("status = 'ready'").limit(10).toArray();
await table.search(queryVector).limit(10).refineFactor(20).toArray();
await table.search(queryVector).limit(10).nprobes(50).toArray();
await table.search(queryVector).limit(10).ef(100).toArray();
await table.search(queryVector).where("status = 'ready'").postfilter().limit(10).toArray();
```
Use `postfilter()` only when fewer than `limit` results are acceptable.
## Diagnostics
```typescript
console.log(await table.search(queryVector).where("year > 2000").limit(10).analyzePlan());
console.log(await table.indexStats("vector_idx"));
```
Use these before changing indexes or search tuning.
## Column (Field) Metadata
```typescript
const schema = await table.schema();
const meta = schema.fields.find((f) => f.name === "category")?.metadata; // Map<string, string>
const res = await table.updateFieldMetadata([
{ path: "category", metadata: { "lancedb:description": "...", "lancedb:tag:field_type": "label" } },
]);
res.version; // new table version
```
Merges by default; a `null` value deletes that key; `replace: true` swaps the whole map. Nested fields use dot-paths (`"a.b.c"`). See `references/column_metadata.md` for key conventions (`lancedb:description`, `lancedb:tag:<name>`, `lancedb:logical-column`) and the authoring workflow.
## Branches
```typescript
const branches = await table.branches(); // async manager
await branches.list(); // non-main branches; {} = only main
const exp = await branches.create("exp"); // fork off main -> Table scoped to the branch
const wip = await branches.checkout("wip"); // existing branch -> scoped Table (version arg pins read-only)
const wip2 = await db.openTable("t", { branch: "wip" }); // or open scoped directly
await branches.delete("stale"); // removes only the branch pointer
table.currentBranch(); // null = main
```
There is no global switch — scoping is per table handle: any read/write on a branch handle lands on that branch; the original handle keeps targeting main. See `references/branch_ops.md` for the model and isolation checks.
## Maintenance
```typescript
await table.optimize();
```
Call this after every successful local/OSS ingestion. It handles compaction, cleanup of old versions according to retention, and index optimization. Do not add this for LanceDB Enterprise/Cloud remote tables; Enterprise handles compaction and cleanup automatically from cluster configuration.
@@ -1,100 +0,0 @@
# TypeScript Patterns
Use these patterns when writing TypeScript code with `@lancedb/lancedb`.
## Recommended Patterns
### Bounded query
Use this for application reads, scripts, and examples:
```typescript
const rows = await table
.query()
.where("status = 'ready'")
.select(["id", "text"])
.limit(20)
.toArray();
```
### Bounded vector search
```typescript
const rows = await table
.search(queryVector)
.select(["id", "text"])
.limit(20)
.toArray();
```
### Batch streaming for larger reads
When the task needs many rows, avoid collecting everything at once:
```typescript
for await (const batch of table
.query()
.where("status = 'ready'")
.select(["id", "text"])
.limit(10_000)) {
process(batch);
}
```
## Anti-Patterns
**Avoid the following anti-patterns in your code.**
### Table-level full materialization
Avoid whole-table collectors in portable or large-table code:
```typescript
const tableArrow = await table.toArrow();
```
Why: local tables expose these whole-table collectors, but remote tables intentionally do not — a remote production table can be far larger than a local development table, so it is easy to accidentally pull the entire table into memory.
### Unbounded result collection
Avoid query/search collection without a meaningful limit:
```typescript
const rows = await table.query().toArray(); // unbounded plain scan
const rows = await table.search(queryVector).toArray(); // unbounded vector search
```
Prefer `select(...).limit(...)` before collecting; for large reads, stream in batches instead.
### Per-row writes
Avoid loops that write one row per call:
```typescript
for (const row of rows) {
await table.add([row]); // one commit + fragment per row
}
```
Each `add()` creates a new version and fragment. Pass the whole batch in a single call, or chunk very large inputs:
```typescript
await table.add(rows); // single commit
// for very large inputs, add in chunks of several thousand rows
```
### Drop-then-reuse the same table name (Enterprise/Cloud)
Avoid dropping or overwriting a remote table and then reusing that name right away:
```typescript
await db.dropTable("my_table");
const table = await db.createTable("my_table", rows); // reads 500 for ~5 min
const table = await db.createTable("my_table", rows, { mode: "overwrite" }); // same problem
```
Why: Enterprise/Cloud splits DDL (control plane) from query serving (data plane). The data plane caches the dataset behind a table name for up to `table_cache_ttl` (default 300s / 5 min), so after a drop/overwrite the DDL succeeds but queries against the reused name return `500 Internal Server Error` until the cache expires — and a fresh `describe` may still show the old schema. Instead, write to a **fresh name**, use `tableNames()` and fail if it already exists, then `renameTable(fresh, final)` onto the final name only after the old table's drop has propagated (~5 min). See the "Enterprise: never drop-then-reuse the same table name" section in `SKILL.md`. Local/OSS tables have no separate data plane — overwrite freely there.
### Guessing performance fixes
Avoid changing `nprobes`, `refineFactor`, `ef`, or index settings before checking `analyzePlan()` and `indexStats(...)`. Diagnose first, then tune one knob at a time.
@@ -1,78 +0,0 @@
# TypeScript Performance Guidance
Use this when writing TypeScript code that ingests data, queries large tables, builds indexes, or investigates latency.
## Ingestion
- Prefer bulk or batched writes.
- Avoid per-row write loops; they create many small commits/fragments.
- For generated data, accumulate reasonable batches before adding.
- For file-backed data, prefer APIs that stream from Arrow/Parquet-style inputs when available.
## Indexing
- Build a vector index once brute-force vector search becomes too slow. As a rule of thumb, local brute force is fine below roughly 100K vectors; beyond that, build an index.
- Use the general-purpose vector index defaults unless the task has explicit recall/latency requirements.
- Build scalar indexes for filtered columns and merge/upsert keys.
- Use full-text index phrase options only when phrase queries require them.
## Querying
Always be explicit:
```typescript
await table.search(queryVector).select(["id", "title"]).limit(20).toArray();
```
- `select()` reduces bytes read and transferred.
- `limit()` prevents accidental full-table collection.
- Pre-filtering is the default behavior. Use `postfilter()` only when fewer than `limit` results are acceptable.
## Recall Tuning
Tune one knob at a time:
- Quantized indexes: raise `refineFactor(...)` to rescore more candidates on full vectors.
- HNSW-backed indexes: raise `ef(...)`; start around `1.5 * k`, increase toward `10 * k` if recall is short.
- IVF candidate breadth: `nprobes(...)` is usually auto-tuned; override only when a selective pre-filter leaves too few neighbors.
## Maintenance
After every successful embedded OSS/local ingestion, call `table.optimize()`.
Do not add this to LanceDB Enterprise/Cloud remote table code; remote compaction
and cleanup are handled automatically based on the Enterprise cluster
configuration.
Why local maintenance is needed:
- Frequent writes can create many small fragments. Queries then need to scan across more files, which can increase latency.
- Updates, deletes, and appends create new table versions. Old versions are retained for time travel and rollback, which can grow disk usage.
- Indexes may have newly added rows that are not yet fully optimized into the index structure.
For local/OSS tables, run `optimize()` after the final successful ingestion
write. Also run it after later batches of update/delete operations or on a
regular maintenance schedule:
```typescript
await table.optimize();
```
If the user wants more aggressive local disk cleanup, pass a shorter cleanup retention window:
```typescript
const olderThan = new Date(Date.now() - 24 * 60 * 60 * 1000);
await table.optimize({ cleanupOlderThan: olderThan });
```
Do not use very short cleanup windows when the application depends on time travel, rollback, or old versions.
## Diagnostics
Before changing code or indexes, inspect:
```typescript
console.log(await table.search(queryVector).where("year > 2000").limit(10).analyzePlan());
console.log(await table.indexStats("vector_idx"));
```
Look for high scan cost, missing indexes, fragmented data, and unindexed rows.
@@ -1,135 +0,0 @@
#!/usr/bin/env python3
"""Scan Python and TypeScript for likely unsafe LanceDB materialization."""
from __future__ import annotations
import argparse
import re
import sys
from dataclasses import dataclass
from pathlib import Path
PY_FULL_TABLE = re.compile(r"\b\w+\.(to_pandas|to_arrow|to_polars)\s*\(")
TS_TABLE_TO_ARROW = re.compile(r"\b\w+\.toArrow\s*\(")
TS_QUERY_COLLECTOR = re.compile(r"\.query\s*\(\s*\)[\s\S]*?\.to(Array|Arrow)\s*\(")
@dataclass
class Finding:
path: Path
line: int
message: str
text: str
def iter_files(paths: list[Path]) -> list[Path]:
files: list[Path] = []
for path in paths:
if path.is_dir():
files.extend(
p
for p in path.rglob("*")
if p.suffix in {".py", ".ts", ".tsx"} and "node_modules" not in p.parts
)
elif path.suffix in {".py", ".ts", ".tsx"}:
files.append(path)
return sorted(set(files))
def line_number(text: str, offset: int) -> int:
return text.count("\n", 0, offset) + 1
def scan_python(path: Path, text: str) -> list[Finding]:
findings: list[Finding] = []
for match in PY_FULL_TABLE.finditer(text):
line_start = text.rfind("\n", 0, match.start()) + 1
line_end = text.find("\n", match.start())
if line_end == -1:
line_end = len(text)
line = text[line_start:line_end].strip()
if ".search(" in line or ".query(" in line:
continue
findings.append(
Finding(
path,
line_number(text, match.start()),
f"Review Python `{match.group(1)}()` call; table-level materialization is not portable to remote tables.",
line,
)
)
return findings
def statement_around(text: str, start: int, end: int) -> str:
before = max(text.rfind(";", 0, start), text.rfind("\n\n", 0, start))
after_candidates = [pos for pos in (text.find(";", end), text.find("\n\n", end)) if pos != -1]
after = min(after_candidates) if after_candidates else len(text)
return text[before + 1 : after].strip()
def scan_typescript(path: Path, text: str) -> list[Finding]:
findings: list[Finding] = []
for match in TS_TABLE_TO_ARROW.finditer(text):
stmt = statement_around(text, match.start(), match.end())
if ".query(" in stmt or ".search(" in stmt:
continue
findings.append(
Finding(
path,
line_number(text, match.start()),
"Review TypeScript `table.toArrow()`-style call; table-level materialization is not portable for large/remote tables.",
stmt.splitlines()[0].strip(),
)
)
for match in TS_QUERY_COLLECTOR.finditer(text):
stmt = statement_around(text, match.start(), match.end())
if ".limit(" in stmt:
continue
findings.append(
Finding(
path,
line_number(text, match.start()),
"Review unbounded TypeScript query collection; add `limit()` or stream batches.",
stmt.splitlines()[0].strip(),
)
)
return findings
def scan_file(path: Path) -> list[Finding]:
text = path.read_text(encoding="utf-8", errors="replace")
if path.suffix == ".py":
return scan_python(path, text)
if path.suffix in {".ts", ".tsx"}:
return scan_typescript(path, text)
return []
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("paths", nargs="+", type=Path)
parser.add_argument(
"--no-fail", action="store_true", help="Always exit 0 after reporting findings."
)
args = parser.parse_args()
findings: list[Finding] = []
for path in iter_files(args.paths):
findings.extend(scan_file(path))
for finding in findings:
print(f"{finding.path}:{finding.line}: {finding.message}")
print(f" {finding.text}")
if findings:
print(
f"\n{len(findings)} finding(s). Review manually; bounded query result conversion may be OK."
)
return 0 if args.no_fail or not findings else 1
if __name__ == "__main__":
sys.exit(main())
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion]
current_version = "0.35.0-beta.2"
current_version = "0.35.0-beta.1"
parse = """(?x)
(?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\.
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.35.0-beta.2"
version = "0.35.0-beta.1"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
-21
View File
@@ -8,27 +8,6 @@ A Python library for [LanceDB](https://github.com/lancedb/lancedb).
pip install lancedb
```
### Pre-Haswell x86_64 hosts: `lancedb-compat`
The default `lancedb` wheel targets `x86-64-haswell` (AVX2 + FMA + F16C) for full performance on modern hardware. Pre-Haswell hosts — Intel Sandy Bridge / Ivy Bridge / Westmere; AMD Bulldozer / Piledriver / Steamroller — don't have AVX2 and crash with `Illegal instruction` at `import lancedb`.
For those hosts, install the `lancedb-compat` package instead:
```bash
pip install lancedb-compat
```
Same Python API (`import lancedb` works as usual). The compat wheel is compiled at the `x86-64-v2` baseline (Nehalem-class) and uses runtime SIMD dispatch in the embedded lance crate to pick the right kernel tier (scalar / AVX / AVX+FMA / AVX2+FMA / AVX-512) at load time, so it still goes fast on modern hardware while running cleanly on the pre-Haswell silicon. Use `lance.simd_info()` from Python to verify which tier was selected.
`lancedb` and `lancedb-compat` install to the same `lancedb/` namespace and conflict at install time. Pick one. To switch, `pip uninstall lancedb` first, then `pip install lancedb-compat` (or vice-versa).
If you need a custom baseline (or `lancedb-compat` isn't yet published for your platform), build from source with the override:
```bash
RUSTFLAGS="-C target-cpu=x86-64-v2" maturin build --release
pip install ./target/wheels/lancedb-*.whl
```
### Preview Releases
Stable releases are created about every 2 weeks. For the latest features and bug fixes, you can install the preview release. These releases receive the same level of testing as stable releases, but are not guaranteed to be available for more than 6 months after they are released. Once your application is stable, we recommend switching to stable releases.
+2 -3
View File
@@ -61,11 +61,10 @@ tests = [
"duckdb>=0.9.0",
"pytz>=2023.3",
"polars>=0.19, <=1.3.0",
"pyarrow<25",
"pyarrow-stubs>=16.0",
"pylance==9.0.0rc1",
"pylance>=5.0.0b5",
"requests>=2.31.0",
"datafusion>=54,<55",
"datafusion>=52,<53",
"opentelemetry-sdk>=1.30.0",
]
dev = [
+2 -43
View File
@@ -6,22 +6,19 @@ import importlib.metadata
import os
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
from typing import Dict, Optional, Union, Any, List, Iterable
from typing import Dict, Optional, Union, Any, List
__version__ = importlib.metadata.version("lancedb")
from ._lancedb import connect as lancedb_connect
from ._lancedb import FtsToken
from ._lancedb import tokenize as _tokenize
from .common import URI, sanitize_uri
from urllib.parse import urlparse
from .db import AsyncConnection, DBConnection, LanceDBConnection
from .remote import ClientConfig
from .remote.db import RemoteDBConnection
from .expr import Expr, col, lit, func
from .schema import blob, vector, BlobType
from .schema import vector
from .table import AsyncTable, Table
from .types import BaseTokenizerType
from ._lancedb import Session
from .namespace import (
connect_namespace,
@@ -249,40 +246,6 @@ def connect(
)
def tokenize(
query: str,
*,
base_tokenizer: BaseTokenizerType = "simple",
language: str = "English",
max_token_length: Optional[int] = 40,
lower_case: bool = True,
stem: bool = True,
remove_stop_words: bool = True,
ascii_folding: bool = True,
ngram_min_length: int = 3,
ngram_max_length: int = 3,
prefix_only: bool = False,
) -> Iterable[FtsToken]:
"""Tokenize a full-text search query using an explicit tokenizer.
This does not require a table or FTS index. The tokenizer options match
:class:`lancedb.index.FTS`.
"""
return _tokenize(
query,
base_tokenizer=base_tokenizer,
language=language,
max_token_length=max_token_length,
lower_case=lower_case,
stem=stem,
remove_stop_words=remove_stop_words,
ascii_folding=ascii_folding,
ngram_min_length=ngram_min_length,
ngram_max_length=ngram_max_length,
prefix_only=prefix_only,
)
WORKER_PROPERTY_PREFIX = "_lancedb_worker_"
@@ -493,21 +456,17 @@ async def connect_async(
__all__ = [
"connect",
"connect_async",
"tokenize",
"connect_namespace",
"connect_namespace_async",
"AsyncConnection",
"AsyncLanceNamespaceDBConnection",
"AsyncTable",
"FtsToken",
"col",
"Expr",
"func",
"lit",
"URI",
"sanitize_uri",
"blob",
"BlobType",
"vector",
"DBConnection",
"LanceDBConnection",
-420
View File
@@ -1,420 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Blob fetch API and v2 projection helpers."""
from __future__ import annotations
import io
from collections.abc import Awaitable, Callable, Iterable
from typing import TYPE_CHECKING, Optional, Union
import pyarrow as pa
from .expr import Expr
from .schema import blob_v2_column_paths
from .types import BlobMode, QueryProjection, QueryProjectionSpec
from .util import get_uri_scheme
if TYPE_CHECKING:
from _typeshed import WriteableBuffer
from .remote.table import RemoteTable
from .table import AsyncTable, Table
BLOB_MODE_TO_HANDLING = {
"lazy": "blobs_descriptions",
"bytes": "all_binary",
"descriptions": "blobs_descriptions",
}
ROW_ID_FIELD_NAME = "_lance_row_id"
FetchBlobsSync = Callable[[str, pa.Table], pa.Array | pa.ChunkedArray]
FetchBlobsAsync = Callable[[str, pa.Table], Awaitable[pa.Array | pa.ChunkedArray]]
class BlobFile(io.RawIOBase):
"""Seekable lazy handle from :meth:`~lancedb.table.Table.fetch_blob_files`.
Bytes load on ``read`` or ``read_range``, not when the handle is opened.
Use :meth:`aread` from async code.
"""
def __init__(self, inner) -> None:
self._inner = inner
async def aread(self) -> bytes:
return await self._inner.read()
def close(self) -> None:
self._inner.close()
@property
def closed(self) -> bool:
return self._inner.is_closed()
def readable(self) -> bool:
return True
def seekable(self) -> bool:
return True
def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
if whence == io.SEEK_SET:
self._inner.seek(offset)
elif whence == io.SEEK_CUR:
self._inner.seek(self._inner.tell() + offset)
elif whence == io.SEEK_END:
self._inner.seek(self._inner.size() + offset)
else:
raise ValueError(f"invalid whence: {whence}")
return self._inner.tell()
def tell(self) -> int:
return self._inner.tell()
def size(self) -> int:
return self._inner.size()
def readall(self) -> bytes:
return self._inner.read_bytes()
def read(self, size: int = -1) -> bytes:
if size == -1:
return self._inner.read_bytes()
return super().read(size)
def read_range(self, offset: int, length: int) -> bytes:
return self._inner.read_range(offset, length)
def readinto(self, b: WriteableBuffer) -> int:
view = memoryview(b).cast("B")
chunk = self._inner.read_up_to(len(view))
view[: len(chunk)] = chunk
return len(chunk)
def __repr__(self) -> str:
return f"<BlobFile size={self.size()}>"
def validate_blob_mode(blob_mode: BlobMode) -> None:
if blob_mode not in BLOB_MODE_TO_HANDLING:
modes = ", ".join(repr(mode) for mode in BLOB_MODE_TO_HANDLING)
raise ValueError(f"blob_mode must be one of {modes}, got {blob_mode!r}")
def supports_blob_auto_row_id(table: Table | AsyncTable | RemoteTable) -> bool:
"""Blob auto row-id applies to native tables, not LanceDB Cloud."""
from .remote.table import RemoteTable
if isinstance(table, RemoteTable):
return False
inner = getattr(table, "_inner", None)
if inner is not None:
uri = inner.database().uri
if isinstance(uri, str) and get_uri_scheme(uri) == "db":
return False
return True
def projection_includes_blob_column(
projection: QueryProjection,
blob_columns: Iterable[str],
) -> bool:
columns = set(blob_columns)
if not columns:
return False
if projection is None:
return True
for output, source in _iter_projection_pairs(projection):
if output in columns or source in columns:
return True
return False
def blob_v2_projection_sources(
schema: pa.Schema,
projection: QueryProjection,
) -> dict[str, str]:
blob_columns = blob_v2_column_paths(schema)
if not blob_columns:
return {}
columns = set(blob_columns)
if projection is None:
return {column: column for column in blob_columns}
return {
output: source
for output, source in _iter_projection_pairs(projection)
if source in columns
}
def v2_projection_needs_row_id(
schema: pa.Schema,
projection: QueryProjection,
*,
with_row_id: bool,
) -> bool:
if with_row_id:
return False
return projection_includes_blob_column(projection, blob_v2_column_paths(schema))
def blob_auto_row_id_for_scan(
table: Table | AsyncTable | RemoteTable,
schema: pa.Schema,
projection: QueryProjection,
*,
with_row_id: bool | None,
) -> bool:
if with_row_id is not None:
return False
if not supports_blob_auto_row_id(table):
return False
return v2_projection_needs_row_id(schema, projection, with_row_id=False)
def finalize_blob_query_table(
tbl: pa.Table,
*,
user_requested_row_id: bool,
blob_auto_row_id: bool,
blob_paths: Iterable[str] = (),
) -> pa.Table:
if user_requested_row_id or not blob_auto_row_id:
return tbl
return stash_auto_row_ids(tbl, blob_paths)
async def replace_v2_blob_columns_with_bytes(
tbl: pa.Table,
blob_sources: dict[str, str],
fetch_blobs: FetchBlobsAsync,
) -> pa.Table:
for output_name, source_name in blob_sources.items():
if output_name not in tbl.column_names:
continue
blobs = await fetch_blobs(source_name, tbl)
tbl = _set_blob_column(tbl, output_name, blobs)
return tbl
def replace_v2_blob_columns_with_bytes_sync(
tbl: pa.Table,
blob_sources: dict[str, str],
fetch_blobs: FetchBlobsSync,
) -> pa.Table:
for output_name, source_name in blob_sources.items():
if output_name not in tbl.column_names:
continue
blobs = fetch_blobs(source_name, tbl)
tbl = _set_blob_column(tbl, output_name, blobs)
return tbl
def stash_auto_row_ids(tbl: pa.Table, blob_paths: Iterable[str]) -> pa.Table:
if "_rowid" not in tbl.column_names:
raise ValueError("query result has no '_rowid' column to hide")
present_paths = [p for p in blob_paths if p.split(".")[0] in tbl.column_names]
if not present_paths:
raise ValueError("query result has no blob v2 column to carry a row id")
row_ids = tbl["_rowid"]
if isinstance(row_ids, pa.ChunkedArray):
row_ids = row_ids.combine_chunks()
row_ids = row_ids.cast(pa.uint64())
for path in present_paths:
tbl = _embed_row_id_in_column(tbl, path, row_ids)
return tbl.drop_columns(["_rowid"])
def read_row_ids_from_hits(hits: pa.Table, blob_column: str) -> list[int]:
if "_rowid" in hits.column_names:
return hits["_rowid"].to_pylist()
try:
leaf = _leaf_struct_column(hits, blob_column)
if ROW_ID_FIELD_NAME in leaf.type.names:
return leaf.field(ROW_ID_FIELD_NAME).to_pylist()
except KeyError:
pass
# blob_column is the source name; aliased projections use the output name in hits.
row_ids = _find_row_id_in_any_column(hits)
if row_ids is not None:
return row_ids
raise ValueError(
f"query result has no '_rowid' column and no '{ROW_ID_FIELD_NAME}' "
f"field on blob column '{blob_column}'. Pass fresh blob query "
"results, call .with_row_id(True), or pass a list of row ids."
)
def _find_row_id_in_any_column(tbl: pa.Table) -> Optional[list[int]]:
for name in tbl.column_names:
column = tbl.column(name)
if isinstance(column, pa.ChunkedArray):
column = column.combine_chunks()
row_ids = _find_row_id_in_struct(column)
if row_ids is not None:
return row_ids
return None
def _find_row_id_in_struct(array: pa.Array) -> Optional[list[int]]:
if not pa.types.is_struct(array.type):
return None
if ROW_ID_FIELD_NAME in array.type.names:
return array.field(ROW_ID_FIELD_NAME).to_pylist()
for i in range(array.type.num_fields):
row_ids = _find_row_id_in_struct(array.field(i))
if row_ids is not None:
return row_ids
return None
def _iter_projection_pairs(
projection: QueryProjectionSpec,
) -> Iterable[tuple[str, str]]:
if isinstance(projection, dict):
for name, expr in projection.items():
if isinstance(expr, str):
yield name, expr
elif isinstance(expr, Expr):
yield name, expr.to_sql()
return
for column in projection:
if isinstance(column, str):
yield column, column
elif isinstance(column, tuple) and len(column) == 2:
name, expr = column
if isinstance(expr, str):
yield name, expr
elif isinstance(expr, Expr):
yield name, expr.to_sql()
def _set_blob_column(tbl: pa.Table, output_name: str, blobs: pa.Array) -> pa.Table:
index = tbl.schema.get_field_index(output_name)
return tbl.set_column(index, pa.field(output_name, blobs.type), [blobs])
def _embed_row_id_in_column(tbl: pa.Table, path: str, row_ids: pa.Array) -> pa.Table:
def add_row_id(children: list, child_fields: list) -> None:
children.append(row_ids)
child_fields.append(pa.field(ROW_ID_FIELD_NAME, pa.uint64(), nullable=False))
return _transform_struct_column(tbl, path, add_row_id)
def strip_auto_row_ids(tbl: pa.Table, blob_paths: Iterable[str]) -> pa.Table:
"""Remove any `_lance_row_id` field embedded in blob descriptor structs.
For read-only descriptor views (`blob_mode="descriptions"`) that never
fetch bytes, so have no use for the row id.
"""
def drop_row_id(children: list, child_fields: list) -> None:
for i, field in enumerate(child_fields):
if field.name == ROW_ID_FIELD_NAME:
del children[i], child_fields[i]
return
for path in blob_paths:
if path.split(".")[0] not in tbl.column_names:
continue
tbl = _transform_struct_column(tbl, path, drop_row_id)
return tbl
def _transform_struct_column(
tbl: pa.Table, path: str, leaf_transform: Callable[[list, list], None]
) -> pa.Table:
top_name, *rest = path.split(".")
top_index = tbl.schema.get_field_index(top_name)
top_field = tbl.schema.field(top_index)
top_array = tbl.column(top_name)
if isinstance(top_array, pa.ChunkedArray):
top_array = top_array.combine_chunks()
new_array, new_field = _rebuild_struct(top_array, top_field, rest, leaf_transform)
return tbl.set_column(top_index, new_field, new_array)
def _rebuild_struct(
struct_array: pa.StructArray,
struct_field: pa.Field,
remaining_path: list[str],
leaf_transform: Callable[[list, list], None],
) -> tuple[pa.StructArray, pa.Field]:
null_mask = struct_array.is_null()
if not remaining_path:
children = [struct_array.field(i) for i in range(struct_array.type.num_fields)]
child_fields = list(struct_array.type)
leaf_transform(children, child_fields)
new_array = pa.StructArray.from_arrays(
children, fields=child_fields, mask=null_mask
)
else:
child_name = remaining_path[0]
child_index = struct_array.type.get_field_index(child_name)
child_array = struct_array.field(child_index)
child_field = struct_array.type.field(child_index)
new_child_array, new_child_field = _rebuild_struct(
child_array, child_field, remaining_path[1:], leaf_transform
)
children = []
child_fields = []
for i in range(struct_array.type.num_fields):
field = struct_array.type.field(i)
if field.name == child_name:
children.append(new_child_array)
child_fields.append(new_child_field)
else:
children.append(struct_array.field(i))
child_fields.append(field)
new_array = pa.StructArray.from_arrays(
children, fields=child_fields, mask=null_mask
)
new_field = pa.field(
struct_field.name,
new_array.type,
nullable=struct_field.nullable,
metadata=struct_field.metadata,
)
return new_array, new_field
def _leaf_struct_column(tbl: pa.Table, path: str) -> pa.StructArray:
parts = path.split(".")
column = tbl.column(parts[0])
if isinstance(column, pa.ChunkedArray):
column = column.combine_chunks()
for part in parts[1:]:
column = column.field(part)
return column
def _normalize_blob_row_ids(
row_ids: Union[list[int], pa.Table], blob_column: str
) -> list[int]:
if isinstance(row_ids, pa.Table):
return read_row_ids_from_hits(row_ids, blob_column)
if isinstance(row_ids, (pa.Array, pa.ChunkedArray)):
raise ValueError(
"pass a query table with _rowid, not a column array "
"(use fetch_blobs('image', hits), not fetch_blobs('image', hits['image']))"
)
return list(row_ids)
def _wrap_blob_files(handles: Iterable[object]) -> list[Optional[BlobFile]]:
return [BlobFile(handle) if handle is not None else None for handle in handles]
+1 -65
View File
@@ -25,12 +25,10 @@ from lance_namespace import (
ListTablesResponse,
)
from .remote import ClientConfig
from .types import BaseTokenizerType
IvfHnswPq: type[HnswPq] = HnswPq
IvfHnswSq: type[HnswSq] = HnswSq
IvfHnswFlat: type[HnswFlat] = HnswFlat
AnalyzePlanDistributedMetrics = Literal["aggregate", "per_worker", "full"]
class MetricPoint:
name: str
@@ -50,20 +48,6 @@ class MetricDescription:
def register_lancedb_metrics_recorder() -> bool: ...
def lancedb_metrics_catalog() -> List[MetricDescription]: ...
def snapshot_lancedb_metrics() -> List[MetricPoint]: ...
def tokenize(
query: str,
*,
base_tokenizer: BaseTokenizerType = "simple",
language: str = "English",
max_token_length: Optional[int] = 40,
lower_case: bool = True,
stem: bool = True,
remove_stop_words: bool = True,
ascii_folding: bool = True,
ngram_min_length: int = 3,
ngram_max_length: int = 3,
prefix_only: bool = False,
) -> List["FtsToken"]: ...
class PyExpr:
"""A type-safe DataFusion expression node (Rust-side handle)."""
@@ -197,17 +181,6 @@ class Connection(object):
self,
) -> Dict[str, Any]: ...
class BlobFile:
async def read(self) -> bytes: ...
def read_bytes(self) -> bytes: ...
def close(self) -> None: ...
def is_closed(self) -> bool: ...
def seek(self, position: int) -> None: ...
def tell(self) -> int: ...
def size(self) -> int: ...
def read_range(self, offset: int, length: int) -> bytes: ...
def read_up_to(self, length: int) -> bytes: ...
class Table:
def name(self) -> str: ...
def __repr__(self) -> str: ...
@@ -219,7 +192,6 @@ class Table:
data: pa.RecordBatchReader,
mode: Literal["append", "overwrite"],
progress: Optional[Any] = None,
write_parallelism: Optional[int] = None,
) -> AddResult: ...
async def update(
self, updates: Dict[str, str], where: Optional[str]
@@ -255,13 +227,6 @@ class Table:
async def prewarm_index(self, index_name: str) -> None: ...
async def prewarm_data(self, columns: Optional[List[str]] = None) -> None: ...
async def list_indices(self) -> list[IndexConfig]: ...
async def tokenize(
self,
query: str,
*,
column: Optional[str] = None,
index_name: Optional[str] = None,
) -> list[FtsToken]: ...
async def delete(self, filter: Union[str, PyExpr]) -> DeleteResult: ...
async def add_columns(self, columns: list[tuple[str, str]]) -> AddColumnsResult: ...
async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
@@ -293,13 +258,6 @@ class Table:
def query(self) -> Query: ...
def take_offsets(self, offsets: list[int]) -> TakeQuery: ...
def take_row_ids(self, row_ids: list[int]) -> TakeQuery: ...
async def blob_columns(self) -> list[str]: ...
async def fetch_blobs(
self, column: str, row_ids: list[int]
) -> pa.LargeBinaryArray: ...
async def fetch_blob_files(
self, column: str, row_ids: list[int]
) -> list[Optional[BlobFile]]: ...
def vector_search(self) -> VectorQuery: ...
class Tags:
@@ -319,10 +277,6 @@ class Branches:
) -> Table: ...
async def checkout(self, name: str, version: Optional[int] = None) -> Table: ...
async def delete(self, name: str) -> None: ...
async def diff(self, from_branch: str) -> Dict[str, Any]: ...
async def merge(
self, from_branch: str, dry_run: bool = False
) -> Dict[str, Any]: ...
class IndexConfig:
name: str
@@ -399,9 +353,7 @@ class Query:
self, max_batch_length: Optional[int], timeout: Optional[timedelta]
) -> RecordBatchStream: ...
async def explain_plan(self, verbose: Optional[bool]) -> str: ...
async def analyze_plan(
self, distributed_metrics: Optional[AnalyzePlanDistributedMetrics] = None
) -> str: ...
async def analyze_plan(self) -> str: ...
def to_query_request(self) -> PyQueryRequest: ...
class TakeQuery:
@@ -409,10 +361,6 @@ class TakeQuery:
def with_row_id(self): ...
async def output_schema(self) -> pa.Schema: ...
async def execute(self) -> RecordBatchStream: ...
async def explain_plan(self, verbose: Optional[bool]) -> str: ...
async def analyze_plan(
self, distributed_metrics: Optional[AnalyzePlanDistributedMetrics] = None
) -> str: ...
def to_query_request(self) -> PyQueryRequest: ...
class FTSQuery:
@@ -433,10 +381,6 @@ class FTSQuery:
async def execute(
self, max_batch_length: Optional[int], timeout: Optional[timedelta]
) -> RecordBatchStream: ...
async def explain_plan(self, verbose: Optional[bool]) -> str: ...
async def analyze_plan(
self, distributed_metrics: Optional[AnalyzePlanDistributedMetrics] = None
) -> str: ...
def to_query_request(self) -> PyQueryRequest: ...
class VectorQuery:
@@ -459,10 +403,6 @@ class VectorQuery:
def bypass_vector_index(self): ...
def nearest_to_text(self, query: dict) -> HybridQuery: ...
def order_by(self, ordering: Optional[List[ColumnOrdering]]): ...
async def explain_plan(self, verbose: Optional[bool]) -> str: ...
async def analyze_plan(
self, distributed_metrics: Optional[AnalyzePlanDistributedMetrics] = None
) -> str: ...
def to_query_request(self) -> PyQueryRequest: ...
class HybridQuery:
@@ -553,10 +493,6 @@ class MergeResult:
num_attempts: int
num_rows: int
class FtsToken:
text: str
position: int
class LsmWriteSpec:
"""Specification selecting Lance's MemWAL LSM-style write path for
`merge_insert`."""
+62 -38
View File
@@ -41,7 +41,6 @@ from lance_namespace import (
ListTablesResponse,
connect as namespace_connect,
)
from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
from . import __version__
from ._lancedb import connect as lancedb_connect # type: ignore
@@ -747,12 +746,10 @@ class LanceDBConnection(DBConnection):
"""
if namespace_path is None:
namespace_path = []
return LOOP.run(
self._conn.list_namespaces(
namespace_path=namespace_path,
page_token=page_token,
limit=limit,
)
return self._namespace_conn().list_namespaces(
namespace_path=namespace_path,
page_token=page_token,
limit=limit,
)
@override
@@ -762,12 +759,10 @@ class LanceDBConnection(DBConnection):
mode: Optional[str] = None,
properties: Optional[Dict[str, str]] = None,
) -> CreateNamespaceResponse:
return LOOP.run(
self._conn.create_namespace(
namespace_path=namespace_path,
mode=mode,
properties=properties,
)
return self._namespace_conn().create_namespace(
namespace_path=namespace_path,
mode=mode,
properties=properties,
)
@override
@@ -777,24 +772,19 @@ class LanceDBConnection(DBConnection):
mode: Optional[str] = None,
behavior: Optional[str] = None,
) -> DropNamespaceResponse:
try:
return LOOP.run(
self._conn.drop_namespace(
namespace_path=namespace_path,
mode=mode,
behavior=behavior,
)
)
except RuntimeError as e:
if "Namespace not empty" in str(e):
raise NamespaceNotEmptyError(str(e)) from e
raise
return self._namespace_conn().drop_namespace(
namespace_path=namespace_path,
mode=mode,
behavior=behavior,
)
@override
def describe_namespace(
self, namespace_path: List[str]
) -> DescribeNamespaceResponse:
return LOOP.run(self._conn.describe_namespace(namespace_path=namespace_path))
return self._namespace_conn().describe_namespace(
namespace_path=namespace_path,
)
@override
def list_tables(
@@ -823,6 +813,12 @@ class LanceDBConnection(DBConnection):
"""
if namespace_path is None:
namespace_path = []
if namespace_path:
return self._namespace_conn().list_tables(
namespace_path=namespace_path,
page_token=page_token,
limit=limit,
)
return LOOP.run(
self._conn.list_tables(
namespace_path=namespace_path, page_token=page_token, limit=limit
@@ -920,6 +916,22 @@ class LanceDBConnection(DBConnection):
raise ValueError("mode must be either 'create' or 'overwrite'")
validate_table_name(name)
if namespace_path:
return self._namespace_conn().create_table(
name,
data=data,
schema=schema,
mode=mode,
exist_ok=exist_ok,
on_bad_vectors=on_bad_vectors,
fill_value=fill_value,
embedding_functions=embedding_functions,
namespace_path=namespace_path,
storage_options=storage_options,
data_storage_version=data_storage_version,
enable_v2_manifest_paths=enable_v2_manifest_paths,
)
tbl = LanceTable.create(
self,
name,
@@ -932,11 +944,22 @@ class LanceDBConnection(DBConnection):
embedding_functions=embedding_functions,
namespace_path=namespace_path,
storage_options=storage_options,
data_storage_version=data_storage_version,
enable_v2_manifest_paths=enable_v2_manifest_paths,
)
return tbl
def _namespace_conn(self) -> DBConnection:
"""Return a LanceNamespaceDBConnection backed by this connection's
directory namespace. Used to delegate child-namespace operations."""
from lancedb.namespace import LanceNamespaceDBConnection
return LanceNamespaceDBConnection(
self.namespace_client(),
read_consistency_interval=self.read_consistency_interval,
storage_options=self.storage_options,
namespace_client_impl=None,
namespace_client_properties=None,
)
@override
def open_table(
self,
@@ -983,7 +1006,14 @@ class LanceDBConnection(DBConnection):
stacklevel=2,
)
try:
if namespace_path:
tbl = self._namespace_conn().open_table(
name,
namespace_path=namespace_path,
storage_options=storage_options,
index_cache_size=index_cache_size,
)
else:
tbl = LanceTable.open(
self,
name,
@@ -991,15 +1021,6 @@ class LanceDBConnection(DBConnection):
storage_options=storage_options,
index_cache_size=index_cache_size,
)
except (RuntimeError, ValueError) as e:
if namespace_path and (
"Table not found" in str(e) or "was not found" in str(e)
):
table_id = namespace_path + [name]
raise TableNotFoundError(
f"Table not found: {'$'.join(table_id)}"
) from e
raise
if branch is not None:
tbl = tbl.branches.checkout(branch, version)
@@ -1083,6 +1104,9 @@ class LanceDBConnection(DBConnection):
"""
if namespace_path is None:
namespace_path = []
if namespace_path:
self._namespace_conn().drop_table(name, namespace_path=namespace_path)
return
LOOP.run(
self._conn.drop_table(
name, namespace_path=namespace_path, ignore_missing=ignore_missing
+10 -23
View File
@@ -4,7 +4,7 @@
import os
from functools import cached_property
from typing import List, Optional, Union
from typing import List, Union
import numpy as np
@@ -15,8 +15,6 @@ from .base import TextEmbeddingFunction
from .registry import register
from .utils import TEXT, api_key_not_found_help
EMBEDDING_BATCH_SIZE = 100
@register("gemini-text")
class GeminiText(TextEmbeddingFunction):
@@ -83,7 +81,6 @@ class GeminiText(TextEmbeddingFunction):
"""
name: str = "gemini-embedding-001"
dim: Optional[int] = None
query_task_type: str = "retrieval_query"
source_task_type: str = "retrieval_document"
@@ -96,8 +93,6 @@ class GeminiText(TextEmbeddingFunction):
model_config["ignored_types"] = (cached_property,)
def ndims(self):
if self.dim:
return self.dim
# TODO: fix hardcoding
return 768
@@ -138,22 +133,22 @@ class GeminiText(TextEmbeddingFunction):
contents.append({"parts": [{"text": text}]})
# Build config
config_kwargs = {"output_dimensionality": self.ndims()}
config_kwargs = {}
if task_type:
config_kwargs["task_type"] = task_type.upper() # API expects uppercase
config = types.EmbedContentConfig(**config_kwargs) if config_kwargs else None
# Call embed_content in groups of at most EMBEDDING_BATCH_SIZE docs at a time
# Call embed_content for each content
embeddings = []
for i in range(0, len(contents), EMBEDDING_BATCH_SIZE):
chunk = contents[i : i + EMBEDDING_BATCH_SIZE]
for content in contents:
config = (
types.EmbedContentConfig(**config_kwargs) if config_kwargs else None
)
response = self.client.models.embed_content(
model=self.name,
contents=chunk,
contents=content,
config=config,
)
embeddings.extend([np.array(e.values) for e in response.embeddings])
embeddings.append(response.embeddings[0].values)
return embeddings
@@ -165,13 +160,5 @@ class GeminiText(TextEmbeddingFunction):
api_key_not_found_help("google")
from google import genai as genai_module
from lancedb import __version__
return genai_module.Client(
api_key=os.environ.get("GOOGLE_API_KEY"),
http_options={
"headers": {
"x-goog-api-client": f"lancedb/{__version__}",
}
},
)
return genai_module.Client(api_key=os.environ.get("GOOGLE_API_KEY"))
+34 -103
View File
@@ -14,76 +14,29 @@ import numpy as np
DEFAULT_WATSONX_URL = "https://us-south.ml.cloud.ibm.com"
# Models currently available on the watsonx.ai SaaS platform.
# These are the IDs advertised to new users via model_names() and shown in
# validation error messages. Regional availability and withdrawal dates are
# documented at:
# https://www.ibm.com/docs/en/watsonx/saas?topic=models-supported-encoder
CURRENT_MODELS: dict[str, int] = {
"ibm/granite-embedding-278m-multilingual": 768,
"ibm/slate-125m-english-rtrvr-v2": 768,
"ibm/slate-30m-english-rtrvr-v2": 384,
"intfloat/multilingual-e5-large": 1024,
}
# Full dimension map including legacy model IDs from earlier releases.
# Kept so that existing tables whose stored metadata uses these names can still
# resolve dimensions on load without raising an error. These IDs are NOT
# advertised to new users.
MODELS_DIMS: dict[str, int] = {
**CURRENT_MODELS,
# Deprecated — withdrawal announced but still functional until the dates above.
"sentence-transformers/all-minilm-l6-v2": 384,
# Pre-v2 legacy names retained for metadata compatibility only.
MODELS_DIMS = {
"ibm/slate-125m-english-rtrvr": 768,
"ibm/slate-30m-english-rtrvr": 384,
"sentence-transformers/all-minilm-l12-v2": 384,
"intfloat/multilingual-e5-large": 1024,
}
@register("watsonx")
class WatsonxEmbeddings(TextEmbeddingFunction):
"""
An embedding function that uses the IBM watsonx.ai Embeddings API.
API Docs:
https://cloud.ibm.com/apidocs/watsonx-ai#text-embeddings
---------
https://cloud.ibm.com/apidocs/watsonx-ai#text-embeddings
Supported embedding models:
https://dataplatform.cloud.ibm.com/docs/content/wsj/analyze-data/fm-models-embed.html?context=wx
Parameters
----------
name : str, default "ibm/slate-125m-english-rtrvr"
The ID of the embedding model to use. For new tables,
``"ibm/granite-embedding-278m-multilingual"`` is recommended.
api_key : str, optional
IBM Cloud API key. Falls back to the ``WATSONX_API_KEY`` environment
variable when not provided.
project_id : str, optional
watsonx.ai project ID. Explicit value takes precedence over the
``WATSONX_PROJECT_ID`` environment variable. Mutually exclusive with
``space_id`` exactly one must be supplied.
space_id : str, optional
watsonx.ai deployment space ID. Explicit value takes precedence over
the ``WATSONX_SPACE_ID`` environment variable. Mutually exclusive with
``project_id`` exactly one must be supplied.
url : str, optional
watsonx.ai service URL. Defaults to
``"https://us-south.ml.cloud.ibm.com"``.
params : dict, optional
Extra parameters forwarded verbatim to ``Embeddings`` (e.g.
``{"truncate_input_tokens": 512}``).
---------------------------
https://dataplatform.cloud.ibm.com/docs/content/wsj/analyze-data/fm-models-embed.html?context=wx
"""
# Intentionally kept at the original pre-PR default so that existing tables
# whose stored metadata contains model:{} reload with the same model they
# were created with. New users should pass name= explicitly, e.g.
# name="ibm/granite-embedding-278m-multilingual".
name: str = "ibm/slate-125m-english-rtrvr"
api_key: Optional[str] = None
project_id: Optional[str] = None
space_id: Optional[str] = None
url: Optional[str] = None
params: Optional[Dict] = None
@@ -93,13 +46,12 @@ class WatsonxEmbeddings(TextEmbeddingFunction):
@staticmethod
def model_names():
"""Return the IDs of models currently available for new tables.
Legacy / deprecated IDs are intentionally excluded. They remain
resolvable for dimension lookups on existing tables via ``MODELS_DIMS``,
but should not be used when creating new tables.
"""
return list(CURRENT_MODELS.keys())
return [
"ibm/slate-125m-english-rtrvr",
"ibm/slate-30m-english-rtrvr",
"sentence-transformers/all-minilm-l12-v2",
"intfloat/multilingual-e5-large",
]
def ndims(self):
return self._ndims
@@ -107,10 +59,7 @@ class WatsonxEmbeddings(TextEmbeddingFunction):
@cached_property
def _ndims(self):
if self.name not in MODELS_DIMS:
raise ValueError(
f"Unknown model '{self.name}'. "
f"Available models: {list(CURRENT_MODELS.keys())}"
)
raise ValueError(f"Unknown model name {self.name}")
return MODELS_DIMS[self.name]
def generate_embeddings(
@@ -132,45 +81,27 @@ class WatsonxEmbeddings(TextEmbeddingFunction):
"ibm_watsonx_ai.foundation_models"
)
# --- credentials ---
# Explicit field takes priority; env var is the fallback.
api_key = self.api_key or os.environ.get("WATSONX_API_KEY")
if not api_key:
raise ValueError(
"WATSONX_API_KEY not set. Either set it in your environment or "
"pass it as `api_key` argument to WatsonxEmbeddings."
)
credentials = ibm_watsonx_ai.Credentials(
api_key=api_key,
url=self.url or DEFAULT_WATSONX_URL,
)
# --- project_id / space_id (exactly one required) ---
# Explicit field always wins; env var is consulted only when the
# corresponding field was not set, so passing project_id= never
# conflicts with a stray WATSONX_SPACE_ID env var and vice-versa.
space_id, project_id = self.space_id, self.project_id
if project_id is None and space_id is None:
# Neither was passed explicitly — fall back to env vars.
project_id = os.environ.get("WATSONX_PROJECT_ID")
space_id = os.environ.get("WATSONX_SPACE_ID")
if project_id and space_id:
raise ValueError("Provide either `project_id` or `space_id`, not both.")
if not project_id and not space_id:
raise ValueError(
"Either WATSONX_PROJECT_ID or WATSONX_SPACE_ID must be set. "
"Pass one as an argument to WatsonxEmbeddings or set the "
"corresponding environment variable."
)
client_kwargs: Dict = dict(model_id=self.name, credentials=credentials)
kwargs = {"model_id": self.name}
if self.params:
client_kwargs["params"] = self.params
if project_id:
client_kwargs["project_id"] = project_id
kwargs["params"] = self.params
if self.project_id:
kwargs["project_id"] = self.project_id
elif "WATSONX_PROJECT_ID" in os.environ:
kwargs["project_id"] = os.environ["WATSONX_PROJECT_ID"]
else:
client_kwargs["space_id"] = space_id
raise ValueError("WATSONX_PROJECT_ID must be set or passed")
return ibm_watsonx_ai_foundation_models.Embeddings(**client_kwargs)
creds_kwargs = {}
if self.api_key:
creds_kwargs["api_key"] = self.api_key
elif "WATSONX_API_KEY" in os.environ:
creds_kwargs["api_key"] = os.environ["WATSONX_API_KEY"]
else:
raise ValueError("WATSONX_API_KEY must be set or passed")
if self.url:
creds_kwargs["url"] = self.url
else:
creds_kwargs["url"] = DEFAULT_WATSONX_URL
kwargs["credentials"] = ibm_watsonx_ai.Credentials(**creds_kwargs)
return ibm_watsonx_ai_foundation_models.Embeddings(**kwargs)
-2
View File
@@ -127,8 +127,6 @@ class FTS:
- "whitespace": Split text by whitespace, but not punctuation.
- "raw": No tokenization. The entire text is treated as a single token.
- "ngram": N-gram tokenizer for substring-style matching.
- "icu": ICU dictionary-based word segmentation.
- "icu/split": ICU segmentation with simple-style delimiter splitting.
- "jieba/*": Jieba tokenizer loaded from Lance's language model home.
- "lindera/*": Lindera tokenizer loaded from Lance's language model home.
language : str, default "English"
+4 -9
View File
@@ -885,7 +885,7 @@ class Permutation:
This method refines the current selection, potentially removing columns. It
will not add back columns that were previously removed.
If any of the columns do not exist then an error will be raised.
If any of the columns do not exist then an error will be raised
This does not introduce a post-processing step. It simply reduces the amount
of data we read.
@@ -898,14 +898,9 @@ class Permutation:
for name in columns:
value = self.selection.get(name, None)
if value is None:
if name == "_rowid":
# _rowid is a system column not in the default schema
# but can be explicitly selected
value = "_rowid"
else:
raise ValueError(
f"Cannot select column `{name}` because it does not exist"
)
raise ValueError(
f"Cannot select column `{name}` because it does not exist"
)
new_selection[name] = value
return self._with_selection(new_selection)
+101 -379
View File
@@ -15,12 +15,10 @@ from typing import (
List,
Literal,
Optional,
Protocol,
Tuple,
Type,
TypeVar,
Union,
runtime_checkable,
)
import deprecation
@@ -41,21 +39,15 @@ from .expr import Expr
from .rerankers.base import Reranker
from .rerankers.rrf import RRFReranker
from .rerankers.util import check_reranker_result
from .schema import is_blob_like_field, schema_has_blob_field
from .util import flatten_columns
from ._blob import (
BLOB_MODE_TO_HANDLING,
FetchBlobsAsync,
FetchBlobsSync,
blob_auto_row_id_for_scan,
blob_v2_projection_sources,
finalize_blob_query_table,
replace_v2_blob_columns_with_bytes,
replace_v2_blob_columns_with_bytes_sync,
supports_blob_auto_row_id,
validate_blob_mode,
)
from .types import BlobMode, QueryProjection
BlobMode = Literal["lazy", "bytes", "descriptions"]
_BLOB_MODE_TO_HANDLING = {
"lazy": "blobs_descriptions",
"bytes": "all_binary",
"descriptions": "blobs_descriptions",
}
if TYPE_CHECKING:
import sys
@@ -79,25 +71,27 @@ if TYPE_CHECKING:
from typing_extensions import Self
T = TypeVar("T", bound="LanceModel")
AnalyzePlanDistributedMetrics = Literal["aggregate", "per_worker", "full"]
@runtime_checkable
class _LanceScanner(Protocol):
projected_schema: pa.Schema | None
schema: pa.Schema | None
def _validate_blob_mode(blob_mode: BlobMode) -> None:
if blob_mode not in _BLOB_MODE_TO_HANDLING:
modes = ", ".join(repr(mode) for mode in _BLOB_MODE_TO_HANDLING)
raise ValueError(f"blob_mode must be one of {modes}, got {blob_mode!r}")
def to_pandas(self, blob_mode: BlobMode | None = ..., **kwargs) -> pd.DataFrame: ...
def to_pyarrow(self): ...
def _field_is_blob(field: pa.Field) -> bool:
metadata = field.metadata or {}
return metadata.get(b"lance-encoding:blob") == b"true" or (
metadata.get("lance-encoding:blob") == "true"
)
def to_table(self) -> pa.Table: ...
def to_reader(self): ...
def _schema_has_blob_field(schema: pa.Schema) -> bool:
return any(_field_is_blob(field) for field in schema)
def _blob_mode_requires_native_pandas(blob_mode: BlobMode, schema: pa.Schema) -> bool:
return blob_mode in BLOB_MODE_TO_HANDLING and schema_has_blob_field(schema)
return blob_mode in _BLOB_MODE_TO_HANDLING and _schema_has_blob_field(schema)
def _unsupported_blob_pandas_error(reason: str) -> RuntimeError:
@@ -146,7 +140,13 @@ def _combine_where(
return f"({existing_sql}) AND ({new_sql})"
def _projection_to_scanner_kwargs(columns: QueryProjection) -> Dict[str, Any]:
def _projection_to_scanner_kwargs(
columns: Optional[
Union[
List[str], List[Tuple[str, Union[str, Expr]]], Dict[str, Union[str, Expr]]
]
],
) -> Dict[str, Any]:
if columns is None:
return {}
if isinstance(columns, list):
@@ -171,11 +171,7 @@ def _projection_to_scanner_kwargs(columns: QueryProjection) -> Dict[str, Any]:
def _scanner_kwargs_for_query(
query: Query,
blob_mode: BlobMode,
dataset: Optional[Any] = None,
*,
with_row_id: Optional[bool] = None,
query: Query, blob_mode: BlobMode, dataset: Optional[Any] = None
) -> Dict[str, Any]:
fragments = _scanner_fragments_for_query(query, dataset)
kwargs = {
@@ -183,10 +179,10 @@ def _scanner_kwargs_for_query(
"filter": _filter_to_sql(query.filter),
"limit": query.limit,
"offset": query.offset,
"with_row_id": with_row_id if with_row_id is not None else query.with_row_id,
"with_row_id": query.with_row_id,
"with_row_address": query.with_row_address,
"fast_search": query.fast_search,
"blob_handling": BLOB_MODE_TO_HANDLING[blob_mode],
"blob_handling": _BLOB_MODE_TO_HANDLING[blob_mode],
"fragments": fragments,
}
return {key: value for key, value in kwargs.items() if value is not None}
@@ -219,11 +215,11 @@ def _scanner_fragments_for_query(query: Query, dataset: Optional[Any]) -> Option
def _ensure_lazy_blob_frame(
df: "pd.DataFrame", schema: pa.Schema, blob_mode: BlobMode
) -> "pd.DataFrame":
if blob_mode != "lazy" or not schema_has_blob_field(schema) or len(df) == 0:
if blob_mode != "lazy" or not _schema_has_blob_field(schema) or len(df) == 0:
return df
for field in schema:
if not is_blob_like_field(field) or field.name not in df.columns:
if not _field_is_blob(field) or field.name not in df.columns:
continue
value = df[field.name].iloc[0]
if value is not None and not hasattr(value, "readall"):
@@ -233,7 +229,7 @@ def _ensure_lazy_blob_frame(
return df
def _scanner_to_table(scanner: _LanceScanner) -> pa.Table:
def _scanner_to_table(scanner: Any) -> pa.Table:
if hasattr(scanner, "to_pyarrow"):
reader = scanner.to_pyarrow()
return reader.read_all()
@@ -243,9 +239,7 @@ def _scanner_to_table(scanner: _LanceScanner) -> pa.Table:
return reader.read_all()
def _scanner_to_pandas(
scanner: _LanceScanner, blob_mode: BlobMode, **kwargs
) -> pd.DataFrame:
def _scanner_to_pandas(scanner: Any, blob_mode: BlobMode, **kwargs) -> "pd.DataFrame":
schema = getattr(scanner, "projected_schema", None)
if schema is None:
schema = getattr(scanner, "schema", None)
@@ -266,71 +260,13 @@ def _scanner_to_pandas(
return df
tbl = _scanner_to_table(scanner)
if blob_mode == "lazy" and schema_has_blob_field(tbl.schema):
if blob_mode == "lazy" and _schema_has_blob_field(tbl.schema):
raise _unsupported_blob_pandas_error(
"the Lance scanner does not expose to_pandas"
)
return tbl.to_pandas(**kwargs)
def _finish_plain_scan_pandas(
scanner: _LanceScanner,
*,
blob_mode: BlobMode,
blob_sources: dict[str, str],
fetch_blobs: FetchBlobsSync,
strip_auto_row_id: bool,
flatten: Optional[Union[int, bool]],
**kwargs,
) -> pd.DataFrame:
if blob_sources:
tbl = _scanner_to_table(scanner)
tbl = replace_v2_blob_columns_with_bytes_sync(tbl, blob_sources, fetch_blobs)
if strip_auto_row_id and "_rowid" in tbl.column_names:
tbl = tbl.drop_columns(["_rowid"])
if flatten is not None:
tbl = flatten_columns(tbl, flatten)
return tbl.to_pandas(**kwargs)
if flatten is not None:
tbl = flatten_columns(_scanner_to_table(scanner), flatten)
if strip_auto_row_id and "_rowid" in tbl.column_names:
tbl = tbl.drop_columns(["_rowid"])
return tbl.to_pandas(**kwargs)
df = _scanner_to_pandas(scanner, blob_mode, **kwargs)
if strip_auto_row_id and "_rowid" in df.columns:
return df.drop(columns=["_rowid"])
return df
async def _finish_plain_scan_pandas_async(
scanner: _LanceScanner,
*,
blob_mode: BlobMode,
blob_sources: dict[str, str],
fetch_blobs: FetchBlobsAsync,
strip_auto_row_id: bool,
flatten: Optional[Union[int, bool]],
**kwargs,
) -> pd.DataFrame:
if blob_sources:
tbl = _scanner_to_table(scanner)
tbl = await replace_v2_blob_columns_with_bytes(tbl, blob_sources, fetch_blobs)
if strip_auto_row_id and "_rowid" in tbl.column_names:
tbl = tbl.drop_columns(["_rowid"])
if flatten is not None:
tbl = flatten_columns(tbl, flatten)
return tbl.to_pandas(**kwargs)
if flatten is not None:
tbl = flatten_columns(_scanner_to_table(scanner), flatten)
if strip_auto_row_id and "_rowid" in tbl.column_names:
tbl = tbl.drop_columns(["_rowid"])
return tbl.to_pandas(**kwargs)
df = _scanner_to_pandas(scanner, blob_mode, **kwargs)
if strip_auto_row_id and "_rowid" in df.columns:
return df.drop(columns=["_rowid"])
return df
# Pydantic validation function for vector queries
def ensure_vector_query(
val: Any,
@@ -738,7 +674,7 @@ class Query(pydantic.BaseModel):
distance_type: Optional[str] = None
# which columns to return in the results (dict values may be str or Expr)
columns: QueryProjection = None
columns: Optional[Union[List[str], Dict[str, Union[str, Expr]]]] = None
# minimum number of IVF partitions to search
#
@@ -1022,7 +958,7 @@ class LanceQueryBuilder(ABC):
Forwarded to pyarrow.Table.to_pandas after query execution and
optional flattening.
"""
validate_blob_mode(blob_mode)
_validate_blob_mode(blob_mode)
output_schema = getattr(self, "output_schema", None)
if output_schema is not None:
schema = output_schema()
@@ -1081,11 +1017,6 @@ class LanceQueryBuilder(ABC):
Execute the query and return the results as a pyarrow
[RecordBatchReader](https://arrow.apache.org/docs/python/generated/pyarrow.RecordBatchReader.html)
For v2 blob projections, ``to_batches`` keeps the auto ``_rowid``
column visible so batch consumers can call ``fetch_blobs``. Use
``to_arrow``, ``to_list``, or ``to_pandas`` if you want LanceDB to hide
auto row ids in the final collected result.
Parameters
----------
batch_size: int
@@ -1264,42 +1195,6 @@ class LanceQueryBuilder(ABC):
self._with_row_id = with_row_id
return self
def _user_requested_row_id(self) -> bool:
return self._with_row_id is True
def _blob_auto_row_id_enabled(self) -> bool:
if not supports_blob_auto_row_id(self._table):
return False
return blob_auto_row_id_for_scan(
self._table,
self._table.schema,
self._columns,
with_row_id=self._with_row_id,
)
def _scan_needs_row_id(self) -> bool:
return self._user_requested_row_id() or self._blob_auto_row_id_enabled()
def _query_for_scan(self) -> Query:
query = self.to_query_object()
if self._scan_needs_row_id():
query.with_row_id = True
return query
def _finalize_blob_query_table(self, tbl: pa.Table) -> pa.Table:
blob_auto_row_id = self._blob_auto_row_id_enabled()
blob_paths = (
blob_v2_projection_sources(self._table.schema, self._columns).keys()
if blob_auto_row_id
else ()
)
return finalize_blob_query_table(
tbl,
user_requested_row_id=self._user_requested_row_id(),
blob_auto_row_id=blob_auto_row_id,
blob_paths=blob_paths,
)
def with_row_address(self, with_row_address: bool = True) -> Self:
"""Set whether to return row addresses.
@@ -1373,9 +1268,7 @@ class LanceQueryBuilder(ABC):
self._order_by = ordering
return self
def analyze_plan(
self, distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate"
) -> str:
def analyze_plan(self) -> str:
"""
Run the query and return its execution plan with runtime metrics.
@@ -1413,22 +1306,12 @@ class LanceQueryBuilder(ABC):
fragments_scanned=..., ranges_scanned=1, rows_scanned=1,
bytes_read=..., iops=..., requests=..., task_wait_time=...]
Parameters
----------
distributed_metrics : Literal["aggregate", "per_worker", "full"]
Defaults to "aggregate".
How distributed worker metrics are displayed for remote query plans.
"aggregate" preserves the legacy summary, "per_worker" shows each
worker separately, and "full" includes both.
Returns
-------
plan : str
The physical query execution plan with runtime metrics.
"""
return self._table._analyze_plan(
self.to_query_object(), distributed_metrics=distributed_metrics
)
return self._table._analyze_plan(self.to_query_object())
def vector(self, vector: Union[np.ndarray, list]) -> Self:
"""Set the vector to search for.
@@ -1488,29 +1371,13 @@ class LanceQueryBuilder(ABC):
return None
dataset = self._table.to_lance()
blob_auto_row_id = self._blob_auto_row_id_enabled()
blob_sources = (
blob_v2_projection_sources(self._table.schema, query.columns)
if blob_mode == "bytes"
else {}
)
scanner = dataset.scanner(
**_scanner_kwargs_for_query(
query,
"descriptions" if blob_sources else blob_mode,
dataset,
with_row_id=query.with_row_id or blob_auto_row_id or bool(blob_sources),
)
)
return _finish_plain_scan_pandas(
scanner,
blob_mode=blob_mode,
blob_sources=blob_sources,
fetch_blobs=self._table.fetch_blobs,
strip_auto_row_id=blob_auto_row_id,
flatten=flatten,
**kwargs,
**_scanner_kwargs_for_query(query, blob_mode, dataset)
)
if flatten is not None:
tbl = flatten_columns(_scanner_to_table(scanner), flatten)
return tbl.to_pandas(**kwargs)
return _scanner_to_pandas(scanner, blob_mode, **kwargs)
@abstractmethod
def to_query_object(self) -> Query:
@@ -1758,9 +1625,7 @@ class LanceVectorQueryBuilder(LanceQueryBuilder):
The maximum time to wait for the query to complete.
If None, wait indefinitely.
"""
return self._finalize_blob_query_table(
self.to_batches(timeout=timeout).read_all()
)
return self.to_batches(timeout=timeout).read_all()
def to_query_object(self) -> Query:
"""
@@ -1820,7 +1685,7 @@ class LanceVectorQueryBuilder(LanceQueryBuilder):
vector = self._query if isinstance(self._query, list) else self._query.tolist()
if isinstance(vector[0], np.ndarray):
vector = [v.tolist() for v in vector]
query = self._query_for_scan()
query = self.to_query_object()
result_set = self._table._execute_query(
query, batch_size=batch_size, timeout=timeout
)
@@ -1964,7 +1829,8 @@ class LanceFtsQueryBuilder(LanceQueryBuilder):
Parameters
----------
phrase_query: bool, default True
If True, then an unquoted string query will be wrapped in quotes.
If True, then the query will be wrapped in quotes and
double quotes replaced by single quotes.
Returns
-------
@@ -1974,21 +1840,6 @@ class LanceFtsQueryBuilder(LanceQueryBuilder):
self._phrase_query = phrase_query
return self
def _query_with_phrase_semantics(self) -> str | FullTextQuery:
query = self._query
if not self._phrase_query:
return query
if isinstance(query, str):
if not query.startswith('"') or not query.endswith('"'):
return f'"{query}"'
return query
if isinstance(query, PhraseQuery):
return query
raise TypeError(
"phrase_query() requires a string or PhraseQuery, "
f"got {type(query).__name__}"
)
def fast_search(self) -> LanceFtsQueryBuilder:
"""
Skip a flat search of unindexed data. This will improve
@@ -2013,7 +1864,7 @@ class LanceFtsQueryBuilder(LanceQueryBuilder):
fragments=self._fragments,
fragment_ids=self._fragment_ids,
full_text_query=FullTextSearchQuery(
query=self._query_with_phrase_semantics(), columns=self._fts_columns
query=self._query, columns=self._fts_columns
),
offset=self._offset,
fast_search=self._fast_search,
@@ -2031,13 +1882,22 @@ class LanceFtsQueryBuilder(LanceQueryBuilder):
def to_arrow(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
self._table._ensure_no_legacy_fts_index()
query = self._query_for_scan()
query = self._query
if self._phrase_query:
if isinstance(query, str):
if not query.startswith('"') or not query.endswith('"'):
self._query = f'"{query}"'
elif isinstance(query, FullTextQuery) and not isinstance(
query, PhraseQuery
):
raise TypeError("Please use PhraseQuery for phrase queries.")
query = self.to_query_object()
results = self._table._execute_query(query, timeout=timeout)
results = results.read_all()
if self._reranker is not None:
results = self._reranker.rerank_fts(self._query, results)
check_reranker_result(results)
return self._finalize_blob_query_table(results)
return results
def to_batches(
self, /, batch_size: Optional[int] = None, timeout: Optional[timedelta] = None
@@ -2065,9 +1925,7 @@ class LanceFtsQueryBuilder(LanceQueryBuilder):
class LanceEmptyQueryBuilder(LanceQueryBuilder):
def to_arrow(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
return self._finalize_blob_query_table(
self.to_batches(timeout=timeout).read_all()
)
return self.to_batches(timeout=timeout).read_all()
def to_query_object(self) -> Query:
return Query(
@@ -2089,7 +1947,7 @@ class LanceEmptyQueryBuilder(LanceQueryBuilder):
def to_batches(
self, /, batch_size: Optional[int] = None, timeout: Optional[timedelta] = None
) -> pa.RecordBatchReader:
query = self._query_for_scan()
query = self.to_query_object()
return self._table._execute_query(query, batch_size=batch_size, timeout=timeout)
def rerank(self, reranker: Reranker) -> LanceEmptyQueryBuilder:
@@ -2161,13 +2019,14 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
return vector_query, text_query
def phrase_query(self, phrase_query: bool = True) -> LanceHybridQueryBuilder:
def phrase_query(self, phrase_query: bool = None) -> LanceHybridQueryBuilder:
"""Set whether to use phrase query.
Parameters
----------
phrase_query: bool, default True
If True, then an unquoted string query will be wrapped in quotes.
If True, then the query will be wrapped in quotes and
double quotes replaced by single quotes.
Returns
-------
@@ -2192,25 +2051,15 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
fts_results = fts_future.result()
vector_results = vector_future.result()
results = self._combine_hybrid_results(
return self._combine_hybrid_results(
fts_results=fts_results,
vector_results=vector_results,
norm=self._norm,
fts_query=self._fts_query._query,
reranker=self._reranker,
limit=self._limit,
with_row_ids=True,
with_row_ids=self._with_row_id,
)
return self._finish_hybrid_results(results)
def _finish_hybrid_results(self, results: pa.Table) -> pa.Table:
if self._user_requested_row_id():
return results
if self._blob_auto_row_id_enabled():
return self._finalize_blob_query_table(results)
if "_rowid" in results.column_names:
return results.drop(["_rowid"])
return results
@staticmethod
def _combine_hybrid_results(
@@ -2594,17 +2443,9 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
indented_fts = "\n".join(" " + line for line in fts_plan.splitlines())
return f"{reranker_label}\n {indented_vector}\n {indented_fts}"
def analyze_plan(
self, distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate"
) -> str:
def analyze_plan(self):
"""Execute the query and display with runtime metrics.
Parameters
----------
distributed_metrics : Literal["aggregate", "per_worker", "full"]
Defaults to "aggregate".
How distributed worker metrics are displayed for remote query plans.
Returns
-------
plan : str
@@ -2612,19 +2453,9 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
self._create_query_builders()
results = ["Vector Search Plan:"]
results.append(
self._table._analyze_plan(
self._vector_query.to_query_object(),
distributed_metrics=distributed_metrics,
)
)
results.append(self._table._analyze_plan(self._vector_query.to_query_object()))
results.append("FTS Search Plan:")
results.append(
self._table._analyze_plan(
self._fts_query.to_query_object(),
distributed_metrics=distributed_metrics,
)
)
results.append(self._table._analyze_plan(self._fts_query.to_query_object()))
return "\n".join(results)
def _create_query_builders(self):
@@ -2669,7 +2500,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
self._vector_query.ef(self._ef)
if self._bypass_vector_index:
self._vector_query.bypass_vector_index()
if self._lower_bound is not None or self._upper_bound is not None:
if self._lower_bound or self._upper_bound:
self._vector_query.distance_range(
lower_bound=self._lower_bound, upper_bound=self._upper_bound
)
@@ -2699,9 +2530,6 @@ class AsyncQueryBase(object):
self._with_row_address = None
self._fragments = None
self._fragment_ids = None
self._with_row_id = None
self._blob_auto_row_id = False
self._blob_paths: tuple[str, ...] = ()
def to_query_object(self) -> Query:
"""
@@ -2711,46 +2539,11 @@ class AsyncQueryBase(object):
python and more easily serializable.
"""
query = Query.from_inner(self._inner.to_query_request())
query.with_row_id = self._user_requested_row_id()
query.with_row_address = self._with_row_address
query.fragments = self._fragments
query.fragment_ids = self._fragment_ids
return query
def _user_requested_row_id(self) -> bool:
return self._with_row_id is True
def _blob_auto_row_id_enabled(self) -> bool:
return self._blob_auto_row_id
def _finalize_blob_query_table(self, tbl: pa.Table) -> pa.Table:
return finalize_blob_query_table(
tbl,
user_requested_row_id=self._user_requested_row_id(),
blob_auto_row_id=self._blob_auto_row_id_enabled(),
blob_paths=self._blob_paths,
)
async def _maybe_add_blob_row_id(self) -> None:
if self._table is None or not supports_blob_auto_row_id(self._table):
self._blob_auto_row_id = False
self._blob_paths = ()
return
req = self._inner.to_query_request()
schema = await self._table.schema()
self._blob_auto_row_id = blob_auto_row_id_for_scan(
self._table,
schema,
req.select,
with_row_id=self._with_row_id,
)
if not self._blob_auto_row_id:
self._blob_paths = ()
return
self._blob_paths = tuple(blob_v2_projection_sources(schema, req.select).keys())
self._inner.with_row_id()
def select(self, columns: Union[List[str], dict[str, str]]) -> Self:
"""
Return only the specified columns.
@@ -2803,7 +2596,6 @@ class AsyncQueryBase(object):
"""
Include the _rowid column in the results.
"""
self._with_row_id = True
self._inner.with_row_id()
return self
@@ -2850,7 +2642,6 @@ class AsyncQueryBase(object):
If not specified, no timeout is applied. If the query does not
complete within the specified time, an error will be raised.
"""
await self._maybe_add_blob_row_id()
return AsyncRecordBatchReader(
await self._inner.execute(
max_batch_length=max_batch_length, timeout=timeout
@@ -2881,8 +2672,8 @@ class AsyncQueryBase(object):
complete within the specified time, an error will be raised.
"""
batch_iter = await self.to_batches(timeout=timeout)
return self._finalize_blob_query_table(
pa.Table.from_batches(await batch_iter.read_all(), schema=batch_iter.schema)
return pa.Table.from_batches(
await batch_iter.read_all(), schema=batch_iter.schema
)
async def to_list(self, timeout: Optional[timedelta] = None) -> List[dict]:
@@ -2949,7 +2740,7 @@ class AsyncQueryBase(object):
Forwarded to pyarrow.Table.to_pandas after query execution and
optional flattening.
"""
validate_blob_mode(blob_mode)
_validate_blob_mode(blob_mode)
if hasattr(self._inner, "output_schema"):
schema = await self.output_schema()
if _blob_mode_requires_native_pandas(blob_mode, schema):
@@ -2990,36 +2781,14 @@ class AsyncQueryBase(object):
if not _query_is_plain_scan(query):
return None
schema = await self._table.schema()
blob_auto_row_id = blob_auto_row_id_for_scan(
self._table,
schema,
query.columns,
with_row_id=self._with_row_id,
)
blob_sources = (
blob_v2_projection_sources(schema, query.columns)
if blob_mode == "bytes"
else {}
)
dataset = await self._table._to_lance()
scanner = dataset.scanner(
**_scanner_kwargs_for_query(
query,
"descriptions" if blob_sources else blob_mode,
dataset,
with_row_id=query.with_row_id or blob_auto_row_id or bool(blob_sources),
)
)
return await _finish_plain_scan_pandas_async(
scanner,
blob_mode=blob_mode,
blob_sources=blob_sources,
fetch_blobs=self._table.fetch_blobs,
strip_auto_row_id=blob_auto_row_id,
flatten=flatten,
**kwargs,
**_scanner_kwargs_for_query(query, blob_mode, dataset)
)
if flatten is not None:
tbl = flatten_columns(_scanner_to_table(scanner), flatten)
return tbl.to_pandas(**kwargs)
return _scanner_to_pandas(scanner, blob_mode, **kwargs)
async def to_polars(
self,
@@ -3111,22 +2880,14 @@ class AsyncQueryBase(object):
""" # noqa: E501
return await self._inner.explain_plan(verbose)
async def analyze_plan(
self, distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate"
) -> str:
async def analyze_plan(self):
"""Execute the query and display with runtime metrics.
Parameters
----------
distributed_metrics : Literal["aggregate", "per_worker", "full"]
Defaults to "aggregate".
How distributed worker metrics are displayed for remote query plans.
Returns
-------
plan : str
"""
return await self._inner.analyze_plan(distributed_metrics)
return await self._inner.analyze_plan()
class AsyncStandardQuery(AsyncQueryBase):
@@ -3812,24 +3573,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)
req = fts_query._inner.to_query_request()
blob_auto_row_id = False
blob_paths: tuple[str, ...] = ()
if self._table is not None and supports_blob_auto_row_id(self._table):
schema = await self._table.schema()
blob_auto_row_id = blob_auto_row_id_for_scan(
self._table,
schema,
req.select,
with_row_id=self._with_row_id,
)
if blob_auto_row_id:
blob_paths = tuple(
blob_v2_projection_sources(schema, req.select).keys()
)
self._blob_auto_row_id = blob_auto_row_id
self._blob_paths = blob_paths
# save the row ID choice that was made on the query builder and force it
# to actually fetch the row ids because we need this for reranking
with_row_ids = self._inner.get_with_row_id()
fts_query.with_row_id()
vec_query.with_row_id()
@@ -3845,14 +3591,8 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
fts_query=fts_query.get_query(),
reranker=self._reranker,
limit=self._inner.get_limit(),
with_row_ids=True,
with_row_ids=with_row_ids,
)
if (
not self._user_requested_row_id()
and not blob_auto_row_id
and "_rowid" in result.column_names
):
result = result.drop(["_rowid"])
return AsyncRecordBatchReader(result, max_batch_length=max_batch_length)
@@ -3875,16 +3615,18 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
>>> asyncio.run(doctest_example()) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
RRFReranker(K=60)
ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance]
LanceRead: uri=..., projection=[text], source=stream(_rowid)
GlobalLimitExec: skip=0, fetch=10
FilterExec: _distance@2 IS NOT NULL
SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false]
KNNVectorDistance: metric=l2
LanceRead: uri=..., projection=[vector], ...
Take: columns="vector, _rowid, _distance, (text)"
CoalesceBatchesExec: target_batch_size=1024
GlobalLimitExec: skip=0, fetch=10
FilterExec: _distance@2 IS NOT NULL
SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false]
KNNVectorDistance: metric=l2
LanceRead: uri=..., projection=[vector], ...
ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score]
LanceRead: uri=..., projection=[vector, text], source=stream(_rowid)
GlobalLimitExec: skip=0, fetch=10
MatchQuery: column=text, query=[hello]
Take: columns="_rowid, _score, (vector), (text)"
CoalesceBatchesExec: target_batch_size=1024
GlobalLimitExec: skip=0, fetch=10
MatchQuery: column=text, query=hello
Parameters
----------
@@ -3903,9 +3645,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
indented_fts = "\n".join(" " + line for line in fts_plan.splitlines())
return f"{self._reranker}\n {indented_vector}\n {indented_fts}"
async def analyze_plan(
self, distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate"
) -> str:
async def analyze_plan(self):
"""
Execute the query and return the physical execution plan with runtime metrics.
@@ -3914,24 +3654,14 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
elapsed time, I/O stats, and more. Its useful for debugging and
performance analysis.
Parameters
----------
distributed_metrics : Literal["aggregate", "per_worker", "full"]
Defaults to "aggregate".
How distributed worker metrics are displayed for remote query plans.
Returns
-------
plan : str
"""
results = ["Vector Search Query:"]
results.append(
await self._inner.to_vector_query().analyze_plan(distributed_metrics)
)
results.append(await self._inner.to_vector_query().analyze_plan())
results.append("FTS Search Query:")
results.append(
await self._inner.to_fts_query().analyze_plan(distributed_metrics)
)
results.append(await self._inner.to_fts_query().analyze_plan())
return "\n".join(results)
@@ -4215,22 +3945,14 @@ class BaseQueryBuilder(object):
""" # noqa: E501
return LOOP.run(self._inner.explain_plan(verbose))
def analyze_plan(
self, distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate"
) -> str:
def analyze_plan(self):
"""Execute the query and display with runtime metrics.
Parameters
----------
distributed_metrics : Literal["aggregate", "per_worker", "full"]
Defaults to "aggregate".
How distributed worker metrics are displayed for remote query plans.
Returns
-------
plan : str
"""
return LOOP.run(self._inner.analyze_plan(distributed_metrics))
return LOOP.run(self._inner.analyze_plan())
class LanceTakeQueryBuilder(BaseQueryBuilder):
+3 -54
View File
@@ -28,7 +28,6 @@ from lancedb._lancedb import (
UpdateFieldMetadataResult,
DeleteResult,
DropColumnsResult,
FtsToken,
IndexConfig,
LsmWriteSpec,
MergeResult,
@@ -56,12 +55,7 @@ from lancedb.merge import LanceMergeInsertBuilder
from lancedb.embeddings import EmbeddingFunctionRegistry
from lancedb.table import _normalize_progress
from ..query import (
AnalyzePlanDistributedMetrics,
LanceQueryBuilder,
LanceTakeQueryBuilder,
LanceVectorQueryBuilder,
)
from ..query import LanceVectorQueryBuilder, LanceQueryBuilder, LanceTakeQueryBuilder
from ..table import AsyncTable, BlobMode, Branches, IndexStatistics, Query, Table, Tags
from ..types import BaseTokenizerType
@@ -250,23 +244,6 @@ class RemoteTable(Table):
"""List all the indices on the table"""
return LOOP.run(self._table.list_indices())
def tokenize(
self,
query: str,
*,
column: Optional[str] = None,
index_name: Optional[str] = None,
) -> Iterable[FtsToken]:
"""Tokenize a query using the tokenizer configured on an FTS index.
Model-backed tokenizers such as ``jieba/*`` and ``lindera/*`` are
rebuilt in the client process from index metadata, so the same tokenizer
model files must exist locally.
"""
return LOOP.run(
self._table.tokenize(query, column=column, index_name=index_name)
)
def index_stats(self, index_uuid: str) -> Optional[IndexStatistics]:
"""List all the stats of a specified index"""
return LOOP.run(self._table.index_stats(index_uuid))
@@ -574,7 +551,6 @@ class RemoteTable(Table):
on_bad_vectors: str = "error",
fill_value: float = 0.0,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
) -> AddResult:
"""Add more data to the [Table](Table). It has the same API signature as
the OSS version.
@@ -600,12 +576,6 @@ class RemoteTable(Table):
progress: bool, callable, or tqdm-like, optional
A callback or tqdm-compatible progress bar. See
:meth:`Table.add` for details.
write_parallelism: int, optional
Number of partitions to write in parallel. Higher values increase
throughput but also peak memory use, since each partition buffers
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
Returns
-------
@@ -621,7 +591,6 @@ class RemoteTable(Table):
on_bad_vectors=on_bad_vectors,
fill_value=fill_value,
progress=progress,
write_parallelism=write_parallelism,
)
)
finally:
@@ -731,15 +700,8 @@ class RemoteTable(Table):
def _explain_plan(self, query: Query, verbose: Optional[bool] = False) -> str:
return LOOP.run(self._table._explain_plan(query, verbose))
def _analyze_plan(
self,
query: Query,
*,
distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate",
) -> str:
return LOOP.run(
self._table._analyze_plan(query, distributed_metrics=distributed_metrics)
)
def _analyze_plan(self, query: Query) -> str:
return LOOP.run(self._table._analyze_plan(query))
def _output_schema(self, query: Query) -> pa.Schema:
return LOOP.run(self._table._output_schema(query))
@@ -1032,19 +994,6 @@ class RemoteTable(Table):
"migrate_v2_manifest_paths() is not supported on the LanceDB Cloud"
)
def blob_columns(self) -> list[str]:
raise NotImplementedError(
"blob_columns() is not yet supported on the LanceDB Cloud"
)
def fetch_blobs(self, column: str, row_ids) -> pa.LargeBinaryArray:
raise NotImplementedError("fetch_blobs() is not supported on LanceDB Cloud")
def fetch_blob_files(self, column: str, row_ids):
raise NotImplementedError(
"fetch_blob_files() is not supported on LanceDB Cloud"
)
def head(self, n=5) -> pa.Table:
"""
Return the first `n` rows of the table.
@@ -12,7 +12,6 @@ from .rrf import RRFReranker
from .mrr import MRRReranker
from .answerdotai import AnswerdotaiRerankers
from .voyageai import VoyageAIReranker
from .watsonx import WatsonxReranker
__all__ = [
"Reranker",
@@ -26,5 +25,4 @@ __all__ = [
"AnswerdotaiRerankers",
"VoyageAIReranker",
"MRRReranker",
"WatsonxReranker",
]
-188
View File
@@ -1,188 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
import os
from functools import cached_property
from typing import Dict, Optional
import pyarrow as pa
from ..util import attempt_import_or_raise
from .base import Reranker
DEFAULT_WATSONX_URL = "https://us-south.ml.cloud.ibm.com"
class WatsonxReranker(Reranker):
"""
Reranks the results using the IBM watsonx.ai Rerank API.
Uses the ``ibm_watsonx_ai`` SDK (``Rerank.generate``) under the hood.
API Docs:
https://cloud.ibm.com/docs/apis/watsonx-ai#text-rerank
Supported rerank models:
https://dataplatform.cloud.ibm.com/docs/content/wsj/analyze-data/fm-models-embed.html?context=wx#rerank
Parameters
----------
model_name : str, default "cross-encoder/ms-marco-minilm-l-12-v2"
The ID of the rerank model to use.
column : str, default "text"
The name of the column to use as input to the reranker.
top_n : int, optional
Return only the top-n results. If ``None``, all results are returned.
return_score : str, default "relevance"
Options are ``"relevance"`` or ``"all"``.
api_key : str, optional
IBM Cloud API key. Falls back to the ``WATSONX_API_KEY`` environment
variable when not provided.
project_id : str, optional
watsonx.ai project ID. Explicit value takes precedence over the
``WATSONX_PROJECT_ID`` environment variable. Mutually exclusive with
``space_id`` exactly one must be supplied.
space_id : str, optional
watsonx.ai deployment space ID. Explicit value takes precedence over
the ``WATSONX_SPACE_ID`` environment variable. Mutually exclusive with
``project_id`` exactly one must be supplied.
url : str, optional
watsonx.ai service URL. Defaults to
``"https://us-south.ml.cloud.ibm.com"``.
truncate_input_tokens : int, optional
Truncate each input to this many tokens before scoring. Passed
directly to the ``parameters`` dict of ``Rerank.generate``.
"""
def __init__(
self,
model_name: str = "cross-encoder/ms-marco-minilm-l-12-v2",
column: str = "text",
top_n: Optional[int] = None,
return_score: str = "relevance",
api_key: Optional[str] = None,
project_id: Optional[str] = None,
space_id: Optional[str] = None,
url: Optional[str] = None,
truncate_input_tokens: Optional[int] = None,
):
super().__init__(return_score)
self.model_name = model_name
self.column = column
self.top_n = top_n
self.api_key = api_key
self.project_id = project_id
self.space_id = space_id
self.url = url
self.truncate_input_tokens = truncate_input_tokens
def __str__(self) -> str:
return f"WatsonxReranker(model_name={self.model_name})"
@cached_property
def _client(self):
ibm_watsonx_ai = attempt_import_or_raise("ibm_watsonx_ai")
ibm_watsonx_ai_foundation_models = attempt_import_or_raise(
"ibm_watsonx_ai.foundation_models"
)
# --- credentials ---
api_key = self.api_key or os.environ.get("WATSONX_API_KEY")
if not api_key:
raise ValueError(
"WATSONX_API_KEY not set. Either set it in your environment or "
"pass it as `api_key` argument to WatsonxReranker."
)
credentials = ibm_watsonx_ai.Credentials(
api_key=api_key,
url=self.url or DEFAULT_WATSONX_URL,
)
# --- project_id / space_id (exactly one required) ---
# Explicit field always wins; env vars are consulted only when neither
# was passed explicitly, so a stray WATSONX_SPACE_ID never overrides an
# explicit project_id and vice-versa.
project_id = self.project_id
space_id = self.space_id
if project_id is None and space_id is None:
# Neither was passed explicitly — fall back to env vars.
project_id = os.environ.get("WATSONX_PROJECT_ID")
space_id = os.environ.get("WATSONX_SPACE_ID")
if project_id and space_id:
raise ValueError("Provide either `project_id` or `space_id`, not both.")
if not project_id and not space_id:
raise ValueError(
"Either WATSONX_PROJECT_ID or WATSONX_SPACE_ID must be set. "
"Pass one as an argument to WatsonxReranker or set the corresponding "
"environment variable."
)
kwargs: Dict = dict(model_id=self.model_name, credentials=credentials)
if project_id:
kwargs["project_id"] = project_id
else:
kwargs["space_id"] = space_id
return ibm_watsonx_ai_foundation_models.Rerank(**kwargs)
def _build_params(self) -> Dict:
"""Build the ``parameters`` dict forwarded to ``Rerank.generate``."""
return_options: Dict = {"inputs": True}
if self.top_n is not None:
return_options["top_n"] = self.top_n
params: Dict = {"return_options": return_options}
if self.truncate_input_tokens is not None:
params["truncate_input_tokens"] = self.truncate_input_tokens
return params
def _rerank(self, result_set: pa.Table, query: str) -> pa.Table:
result_set = self._handle_empty_results(result_set)
if len(result_set) == 0:
return result_set
docs = result_set[self.column].to_pylist()
response = self._client.generate(
query=query,
inputs=docs,
params=self._build_params(),
)
results = response["results"]
indices, scores = zip(
*[(result["index"], result["score"]) for result in results]
)
result_set = result_set.take(list(indices))
result_set = result_set.append_column(
"_relevance_score", pa.array(scores, type=pa.float32())
)
return result_set
def rerank_hybrid(
self,
query: str,
vector_results: pa.Table,
fts_results: pa.Table,
) -> pa.Table:
if self.score == "all":
combined_results = self._merge_and_keep_scores(vector_results, fts_results)
else:
combined_results = self.merge_results(vector_results, fts_results)
combined_results = self._rerank(combined_results, query)
if self.score == "relevance":
combined_results = self._keep_relevance_score(combined_results)
return combined_results
def rerank_vector(self, query: str, vector_results: pa.Table) -> pa.Table:
vector_results = self._rerank(vector_results, query)
if self.score == "relevance":
vector_results = vector_results.drop_columns(["_distance"])
return vector_results
def rerank_fts(self, query: str, fts_results: pa.Table) -> pa.Table:
fts_results = self._rerank(fts_results, query)
if self.score == "relevance":
fts_results = fts_results.drop_columns(["_score"])
return fts_results
+2 -154
View File
@@ -7,158 +7,10 @@ import sys
from typing import Callable, Iterator, Optional
from lancedb.arrow import to_arrow
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.dataset as ds
from .pydantic import LanceModel
# pyarrow's default scanner settings are tuned for narrow rows. For wide rows
# (e.g. embedding columns) they buffer a huge read-ahead window in host memory
# and can OOM the client during bulk ingestion. We size the scanner so the
# estimated in-flight memory stays within a budget, while leaving narrow
# datasets on pyarrow's defaults (no throughput regression).
_SCAN_MEMORY_BUDGET_BYTES = 1024 * 1024 * 1024 # ~1 GiB in-flight target
_TARGET_BATCH_BYTES = 16 * 1024 * 1024 # ~16 MiB per batch
_MIN_BATCH_ROWS = 512
# pyarrow defaults (see arrow/dataset ScanOptions); we never exceed these.
_PA_DEFAULT_BATCH_ROWS = 131_072
_PA_DEFAULT_BATCH_READAHEAD = 16
_PA_DEFAULT_FRAGMENT_READAHEAD = 4
# Read-ahead used for wide rows. pyarrow reads a whole parquet row group at a
# time and keeps `batch_readahead` of them resident, so read-ahead depth (not
# batch size) dominates peak memory for wide data; keep both small but leave a
# little prefetch for throughput. Tuned empirically against embedding datasets.
_WIDE_BATCH_READAHEAD = 2
_WIDE_FRAGMENT_READAHEAD = 1
# Estimate for variable-width columns (string/binary/list) whose true width is
# unknown from the schema alone. Only needs to be large enough to flag "wide".
_VARIABLE_WIDTH_ESTIMATE = 128
# Rows peeked from a rescannable source to refine the list-length guess for
# variable-length list columns (e.g. embeddings stored as `list<float32>`
# instead of `list<float32, N>`), whose per-row width the schema can't tell us.
_SAMPLE_ROWS = 10
def _observed_list_length(sample: pa.ChunkedArray) -> Optional[int]:
"""Average element count per row observed in a list/large_list sample."""
if len(sample) == 0:
return None
mean = pc.mean(pc.list_value_length(sample)).as_py()
return None if mean is None else max(1, round(mean))
def _estimate_field_width(
dtype: pa.DataType, sample: Optional[pa.ChunkedArray] = None
) -> int:
if pa.types.is_fixed_size_list(dtype):
return dtype.list_size * _estimate_field_width(dtype.value_type)
if pa.types.is_struct(dtype):
return sum(
_estimate_field_width(
dtype.field(i).type,
pc.struct_field(sample, [i]) if sample is not None else None,
)
for i in range(dtype.num_fields)
)
if pa.types.is_dictionary(dtype):
return _estimate_field_width(dtype.value_type)
if pa.types.is_fixed_size_binary(dtype):
return dtype.byte_width
if pa.types.is_boolean(dtype):
return 1
if (pa.types.is_list(dtype) or pa.types.is_large_list(dtype)) and (
sample is not None
):
observed_length = _observed_list_length(sample)
if observed_length is not None:
return observed_length * _estimate_field_width(dtype.value_type)
# Fixed-width scalars (ints, floats, temporal, decimal) expose bit_width;
# variable-width types (string, binary, list, map, ...) raise ValueError.
try:
return max(1, dtype.bit_width // 8)
except (ValueError, AttributeError):
return _VARIABLE_WIDTH_ESTIMATE
def _estimate_bytes_per_row(
schema: pa.Schema, sample: Optional[pa.Table] = None
) -> int:
return max(
1,
sum(
_estimate_field_width(
field.type, sample.column(field.name) if sample is not None else None
)
for field in schema
),
)
def _sample_head(head: Callable[..., pa.Table]) -> Optional[pa.Table]:
"""Best-effort peek at a few rows to refine the bytes-per-row estimate.
Uses a tight batch size and no read-ahead so the peek itself can't trigger
the wide-row memory blowup this module exists to avoid. Returns None (fall
back to the schema-only estimate) if sampling isn't possible for any
reason, e.g. an empty dataset.
"""
try:
sample = head(
_SAMPLE_ROWS,
batch_size=_SAMPLE_ROWS,
batch_readahead=1,
fragment_readahead=1,
)
except Exception:
return None
return sample if sample.num_rows > 0 else None
def _bounded_scanner_kwargs(
schema: pa.Schema, sample: Optional[pa.Table] = None
) -> dict:
"""Scanner kwargs that cap in-flight memory for wide rows.
Narrow datasets keep pyarrow's defaults unchanged (no throughput
regression). For wide rows (e.g. embedding columns) pyarrow's default
read-ahead buffers many large batches/row-groups at once, which can OOM the
client during bulk ingestion, so we shrink the batch size and read-ahead to
keep the estimated in-flight memory near the budget.
Read-ahead (not just batch size) has to drop: pyarrow reads a whole parquet
row group at a time and keeps `batch_readahead`/`fragment_readahead` of them
resident, so a small batch size alone still pins large row-group buffers.
`sample`, if given, is a small (see `_SAMPLE_ROWS`) table of rows from the
source used to refine the estimate for variable-length list columns (e.g.
embeddings stored without a fixed size), whose width the schema alone
can't tell us.
"""
bytes_per_row = _estimate_bytes_per_row(schema, sample)
# If pyarrow's defaults already stay within budget, leave them alone so
# narrow datasets keep their throughput. A "unit" of in-flight memory is one
# default-sized batch, held `batch_readahead + fragment_readahead` deep.
default_in_flight = (
_PA_DEFAULT_BATCH_ROWS
* bytes_per_row
* (_PA_DEFAULT_BATCH_READAHEAD + _PA_DEFAULT_FRAGMENT_READAHEAD)
)
if default_in_flight <= _SCAN_MEMORY_BUDGET_BYTES:
return {}
# Wide rows: cap batch bytes and pull read-ahead down so only a couple of
# large row-group buffers are resident at once.
batch_size = min(
_PA_DEFAULT_BATCH_ROWS,
max(_MIN_BATCH_ROWS, _TARGET_BATCH_BYTES // bytes_per_row),
)
return {
"batch_size": batch_size,
"batch_readahead": _WIDE_BATCH_READAHEAD,
"fragment_readahead": _WIDE_FRAGMENT_READAHEAD,
}
@dataclass
class Scannable:
@@ -204,12 +56,10 @@ def _from_table(data: pa.Table) -> Scannable:
@to_scannable.register(ds.Dataset)
def _from_dataset(data: ds.Dataset) -> Scannable:
sample = _sample_head(data.head)
scanner_kwargs = _bounded_scanner_kwargs(data.schema, sample)
return Scannable(
schema=data.schema,
num_rows=data.count_rows(),
reader=lambda: data.scanner(**scanner_kwargs).to_reader(),
reader=lambda: data.scanner().to_reader(),
)
@@ -356,12 +206,10 @@ def _register_optional_converters():
@to_scannable.register(lance.LanceDataset)
def _from_lance(data: lance.LanceDataset) -> Scannable:
sample = _sample_head(data.head)
scanner_kwargs = _bounded_scanner_kwargs(data.schema, sample)
return Scannable(
schema=data.schema,
num_rows=data.count_rows(),
reader=lambda: data.scanner(**scanner_kwargs).to_reader(),
reader=lambda: data.scanner().to_reader(),
)
+1 -125
View File
@@ -2,134 +2,10 @@
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Schema helpers for Lance blob columns."""
"""Schema related utilities."""
import pyarrow as pa
_BLOB_EXTENSION_NAME = "lance.blob.v2"
_BLOB_V1_KEY = "lance-encoding:blob"
_ARROW_EXT_NAME_KEY = "ARROW:extension:name"
class BlobType(pa.ExtensionType):
"""PyArrow extension type for a Lance blob v2 column.
Queries return descriptors; call :meth:`~lancedb.table.Table.fetch_blob_files`
for lazy reads or :meth:`~lancedb.table.Table.fetch_blobs` for eager bytes.
"""
def __init__(self) -> None:
storage_type = pa.struct(
[
pa.field("data", pa.large_binary(), nullable=True),
pa.field("uri", pa.utf8(), nullable=True),
pa.field("position", pa.uint64(), nullable=True),
pa.field("size", pa.uint64(), nullable=True),
]
)
super().__init__(storage_type, _BLOB_EXTENSION_NAME)
def __arrow_ext_serialize__(self) -> bytes:
return b""
@classmethod
def __arrow_ext_deserialize__(
cls, storage_type: pa.DataType, serialized: bytes
) -> "BlobType":
return cls()
def __reduce__(self):
# Ensure pickle round-trips on older pyarrow (apache/arrow#35599).
return type(self).__arrow_ext_deserialize__, (
self.storage_type,
self.__arrow_ext_serialize__(),
)
try:
pa.register_extension_type(BlobType()) # type: ignore[arg-type]
except pa.ArrowKeyError:
pass
def _metadata_value(metadata: dict, key: str):
return metadata.get(key.encode()) or metadata.get(key)
def _metadata_marks_blob_v2(metadata: dict) -> bool:
if not metadata:
return False
extension_name = _metadata_value(metadata, _ARROW_EXT_NAME_KEY)
return extension_name in (_BLOB_EXTENSION_NAME, _BLOB_EXTENSION_NAME.encode())
def _metadata_marks_legacy_blob(metadata: dict) -> bool:
if not metadata:
return False
return _metadata_value(metadata, _BLOB_V1_KEY) in ("true", b"true")
def is_blob_v2_field(field: pa.Field) -> bool:
"""Return True if `field` declares a blob v2 extension column."""
field_type = field.type
if (
isinstance(field_type, pa.ExtensionType)
and field_type.extension_name == _BLOB_EXTENSION_NAME
):
return True
return _metadata_marks_blob_v2(field.metadata or {})
def is_blob_like_field(field: pa.Field) -> bool:
"""Blob detection for ``to_pandas(blob_mode=...)`` and scanner paths only.
Matches v2 extension fields on table schema, legacy ``lance-encoding:blob``
storage columns, and v2 query descriptor fields (the engine tags those with
the same metadata). Not used for fetch or auto ``_rowid``.
"""
return is_blob_v2_field(field) or _metadata_marks_legacy_blob(field.metadata or {})
def _collect_blob_paths(schema: pa.Schema, is_blob) -> list[str]:
paths: list[str] = []
def walk(fields, prefix: str) -> None:
for field in fields:
path = f"{prefix}.{field.name}" if prefix else field.name
if is_blob(field):
paths.append(path)
elif pa.types.is_struct(field.type):
walk(field.type, path)
elif (
pa.types.is_list(field.type)
or pa.types.is_large_list(field.type)
or pa.types.is_fixed_size_list(field.type)
):
walk([field.type.value_field], path)
walk(schema, "")
return paths
def blob_column_paths(schema: pa.Schema) -> list[str]:
"""Dotted paths of blob-like columns (v2 extension or legacy metadata)."""
return _collect_blob_paths(schema, is_blob_like_field)
def blob_v2_column_paths(schema: pa.Schema) -> list[str]:
return _collect_blob_paths(schema, is_blob_v2_field)
def schema_has_blob_field(schema: pa.Schema) -> bool:
return bool(blob_column_paths(schema))
def blob(name: str, nullable: bool = True) -> pa.Field:
"""Create a Lance blob v2 column field."""
return pa.field(name, BlobType(), nullable=nullable)
def vector(dimension: int, value_type: pa.DataType = pa.float32()) -> pa.DataType:
"""A help function to create a vector type.
+41 -272
View File
@@ -29,14 +29,6 @@ from urllib.parse import urlparse
from lancedb.scannable import _register_optional_converters, to_scannable
from . import __version__
from ._blob import (
BlobFile,
_normalize_blob_row_ids,
_wrap_blob_files,
strip_auto_row_ids,
validate_blob_mode,
)
from .types import BlobMode
from lancedb.arrow import peek_reader
from lancedb.background_loop import LOOP, embedding_executor
from .dependencies import (
@@ -73,7 +65,6 @@ from .expr import Expr
from .merge import LanceMergeInsertBuilder
from .pydantic import LanceModel, model_to_dict
from .query import (
AnalyzePlanDistributedMetrics,
AsyncFTSQuery,
AsyncHybridQuery,
AsyncQuery,
@@ -97,7 +88,10 @@ from .util import (
value_to_sql,
)
from .index import lang_mapping
from .schema import blob_v2_column_paths, schema_has_blob_field
BlobMode = Literal["lazy", "bytes", "descriptions"]
_VALID_BLOB_MODES = ("lazy", "bytes", "descriptions")
def _should_push_down_query_table(
@@ -106,6 +100,23 @@ def _should_push_down_query_table(
return namespace_client is not None and "QueryTable" in pushdown_operations
def _validate_blob_mode(blob_mode: BlobMode) -> None:
if blob_mode not in _VALID_BLOB_MODES:
modes = ", ".join(repr(mode) for mode in _VALID_BLOB_MODES)
raise ValueError(f"blob_mode must be one of {modes}, got {blob_mode!r}")
def _field_is_blob(field: pa.Field) -> bool:
metadata = field.metadata or {}
return metadata.get(b"lance-encoding:blob") == b"true" or (
metadata.get("lance-encoding:blob") == "true"
)
def _schema_has_blob_field(schema: pa.Schema) -> bool:
return any(_field_is_blob(field) for field in schema)
_MODEL_BACKED_TOKENIZER_PREFIXES = ("jieba", "lindera")
_MODEL_BACKED_TOKENIZER_ERRORS = (
"unknown base tokenizer",
@@ -174,7 +185,6 @@ if TYPE_CHECKING:
UpdateFieldMetadataResult,
DeleteResult,
DropColumnsResult,
FtsToken,
LsmWriteSpec,
MergeResult,
UpdateResult,
@@ -1149,8 +1159,6 @@ class Table(ABC):
- "whitespace": Split text by whitespace, but not punctuation.
- "raw": No tokenization. The entire text is treated as a single token.
- "ngram": N-Gram tokenizer.
- "icu": ICU dictionary-based word segmentation.
- "icu/split": ICU segmentation with simple-style delimiter splitting.
- "jieba/*": Jieba tokenizer loaded from Lance's language model home.
- "lindera/*": Lindera tokenizer loaded from Lance's language model home.
language : str, default "English"
@@ -1199,7 +1207,6 @@ class Table(ABC):
on_bad_vectors: OnBadVectorsType = "error",
fill_value: float = 0.0,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
) -> AddResult:
"""Add more data to the [Table](Table).
@@ -1245,13 +1252,6 @@ class Table(ABC):
with tqdm() as pbar:
table.add(data, progress=pbar)
write_parallelism: int, optional
Number of partitions to write in parallel. Higher values increase
throughput but also peak memory use, since each partition buffers
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
Returns
-------
AddResult
@@ -1523,31 +1523,6 @@ class Table(ABC):
A query object that can be executed to get the rows.
"""
@abstractmethod
def blob_columns(self) -> list[str]:
"""Names of the blob v2 columns declared on this table."""
@abstractmethod
def fetch_blobs(
self, column: str, row_ids: Union[list[int], pa.Table]
) -> pa.LargeBinaryArray:
"""Materialize full blob bytes for ``column`` at the given rows.
Convenience for small payloads. For large values use
:meth:`fetch_blob_files`.
"""
@abstractmethod
def fetch_blob_files(
self, column: str, row_ids: Union[list[int], pa.Table]
) -> "list[Optional[BlobFile]]":
"""Open lazy, seekable :class:`~lancedb._blob.BlobFile` handles.
Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
a ``list[int]`` or query ``pyarrow.Table`` with ``_rowid`` (or stashed
row-id metadata). Null rows are ``None``. Local tables only.
"""
@abstractmethod
def _execute_query(
self,
@@ -1561,12 +1536,7 @@ class Table(ABC):
def _explain_plan(self, query: Query, verbose: Optional[bool] = False) -> str: ...
@abstractmethod
def _analyze_plan(
self,
query: Query,
*,
distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate",
) -> str: ...
def _analyze_plan(self, query: Query) -> str: ...
@abstractmethod
def _output_schema(self, query: Query) -> pa.Schema: ...
@@ -1816,24 +1786,6 @@ class Table(ABC):
[Table.create_index][lancedb.table.Table.create_index]
"""
@abstractmethod
def tokenize(
self,
query: str,
*,
column: Optional[str] = None,
index_name: Optional[str] = None,
) -> Iterable[FtsToken]:
"""
Tokenize a query using the tokenizer configured on an FTS index.
Specify exactly one of ``column`` or ``index_name``.
Model-backed tokenizers such as ``jieba/*`` and ``lindera/*`` are
rebuilt in the client process from index metadata. For remote tables,
this means the same tokenizer model files must also exist locally.
"""
@abstractmethod
def index_stats(self, index_name: str) -> Optional[IndexStatistics]:
"""
@@ -2203,7 +2155,7 @@ class LanceTable(Table):
namespace_client = self._namespace_client
if namespace_client is None:
conn_uri = getattr(self._conn, "uri", "")
if get_uri_scheme(conn_uri) == "namespace" or self._namespace_path:
if get_uri_scheme(conn_uri) == "namespace":
namespace_client = self._conn.namespace_client()
self._namespace_client = namespace_client
@@ -2252,19 +2204,6 @@ class LanceTable(Table):
def take_row_ids(self, row_ids: list[int]) -> LanceTakeQueryBuilder:
return LanceTakeQueryBuilder(self._table.take_row_ids(row_ids))
def blob_columns(self) -> list[str]:
return LOOP.run(self._table.blob_columns())
def fetch_blobs(
self, column: str, row_ids: Union[list[int], pa.Table]
) -> pa.LargeBinaryArray:
return LOOP.run(self._table.fetch_blobs(column, row_ids))
def fetch_blob_files(
self, column: str, row_ids: Union[list[int], pa.Table]
) -> "list[Optional[BlobFile]]":
return LOOP.run(self._table.fetch_blob_files(column, row_ids))
@property
def tags(self) -> Tags:
"""Tag management for the table.
@@ -2460,14 +2399,9 @@ class LanceTable(Table):
-------
pd.DataFrame
"""
validate_blob_mode(blob_mode)
if blob_mode == "descriptions" or not schema_has_blob_field(self.schema):
arrow_tbl = self.to_arrow()
if blob_mode == "descriptions":
arrow_tbl = strip_auto_row_ids(
arrow_tbl, blob_v2_column_paths(self.schema)
)
return arrow_tbl.to_pandas(**kwargs)
_validate_blob_mode(blob_mode)
if blob_mode == "descriptions" or not _schema_has_blob_field(self.schema):
return self.to_arrow().to_pandas(**kwargs)
if (
blob_mode == "lazy"
@@ -2476,9 +2410,6 @@ class LanceTable(Table):
):
return self.to_arrow().to_pandas(**kwargs)
if blob_mode == "bytes" and blob_v2_column_paths(self.schema):
return self.search().to_pandas(blob_mode=blob_mode, **kwargs)
return self.to_lance().to_pandas(blob_mode=blob_mode, **kwargs)
def to_arrow(self) -> pa.Table:
@@ -3166,7 +3097,6 @@ class LanceTable(Table):
on_bad_vectors: OnBadVectorsType = "error",
fill_value: float = 0.0,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
) -> AddResult:
"""Add data to the table.
If vector columns are missing and the table
@@ -3188,12 +3118,6 @@ class LanceTable(Table):
progress: bool, callable, or tqdm-like, optional
A callback or tqdm-compatible progress bar. See
:meth:`Table.add` for details.
write_parallelism: int, optional
Number of partitions to write in parallel. Higher values increase
throughput but also peak memory use, since each partition buffers
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
Returns
-------
@@ -3209,7 +3133,6 @@ class LanceTable(Table):
on_bad_vectors=on_bad_vectors,
fill_value=fill_value,
progress=progress,
write_parallelism=write_parallelism,
)
)
finally:
@@ -3652,15 +3575,8 @@ class LanceTable(Table):
def _explain_plan(self, query: Query, verbose: Optional[bool] = False) -> str:
return LOOP.run(self._table._explain_plan(query, verbose))
def _analyze_plan(
self,
query: Query,
*,
distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate",
) -> str:
return LOOP.run(
self._table._analyze_plan(query, distributed_metrics=distributed_metrics)
)
def _analyze_plan(self, query: Query) -> str:
return LOOP.run(self._table._analyze_plan(query))
def _output_schema(self, query: Query) -> pa.Schema:
return LOOP.run(self._table._output_schema(query))
@@ -3794,26 +3710,6 @@ class LanceTable(Table):
"""
return LOOP.run(self._table.list_indices())
def tokenize(
self,
query: str,
*,
column: Optional[str] = None,
index_name: Optional[str] = None,
) -> Iterable[FtsToken]:
"""
Tokenize a query using the tokenizer configured on an FTS index.
Specify exactly one of ``column`` or ``index_name``.
Model-backed tokenizers such as ``jieba/*`` and ``lindera/*`` are
rebuilt in the client process from index metadata. For remote tables,
this means the same tokenizer model files must also exist locally.
"""
return LOOP.run(
self._table.tokenize(query, column=column, index_name=index_name)
)
def index_stats(self, index_name: str) -> Optional[IndexStatistics]:
"""
Retrieve statistics about an index
@@ -4183,58 +4079,17 @@ def _handle_bad_vector_column(
raise ValueError(
"`fill_value` must not be None if `on_bad_vectors` is 'fill'"
)
vec_arr = _fill_bad_vector_values(vec_arr, dim, fill_value)
vec_arr = pc.if_else(
is_bad,
pa.scalar([fill_value] * dim, type=vec_arr.type),
vec_arr,
)
else:
raise ValueError(f"Invalid value for on_bad_vectors: {on_bad_vectors}")
return data.set_column(position, vector_column_name, vec_arr)
def _fill_bad_vector_values(
arr: Union[pa.Array, pa.ChunkedArray],
dim: int,
fill_value: float,
) -> pa.Array:
if not isinstance(arr, pa.ChunkedArray):
arr = pa.chunked_array([arr])
arr = arr.combine_chunks()
# A fixed-size slice truncates long vectors and pads short vectors with nulls.
# Slice an array marking the original child nulls in parallel so padding nulls
# can be distinguished from null values that were already present.
sliced = pc.list_slice(arr, 0, dim, return_fixed_size_list=True)
child_nulls = pc.is_null(arr.values)
parent_nulls = pc.is_null(arr)
if pa.types.is_list(arr.type):
original_child_nulls = pa.ListArray.from_arrays(
arr.offsets, child_nulls, mask=parent_nulls
)
elif pa.types.is_large_list(arr.type):
original_child_nulls = pa.LargeListArray.from_arrays(
arr.offsets, child_nulls, mask=parent_nulls
)
else:
original_child_nulls = pa.FixedSizeListArray.from_arrays(
child_nulls, arr.type.list_size, mask=parent_nulls
)
sliced_child_nulls = pc.list_slice(
original_child_nulls, 0, dim, return_fixed_size_list=True
)
needs_fill = pc.is_null(sliced_child_nulls.values)
values = sliced.values
if pa.types.is_floating(values.type):
values_for_nan_check = (
values.cast(pa.float32()) if pa.types.is_float16(values.type) else values
)
needs_fill = pc.or_kleene(needs_fill, pc.is_nan(values_for_nan_check))
fill_scalar = pa.scalar(fill_value).cast(values.type)
filled_values = pc.if_else(needs_fill, fill_scalar, values)
filled = pa.FixedSizeListArray.from_arrays(filled_values, dim)
return filled.cast(arr.type)
def has_nan_values(arr: Union[pa.ListArray, pa.ChunkedArray]) -> pa.BooleanArray:
if isinstance(arr, pa.ChunkedArray):
values = pa.chunked_array([chunk.flatten() for chunk in arr.chunks])
@@ -4684,18 +4539,14 @@ class AsyncTable:
-------
pd.DataFrame
"""
validate_blob_mode(blob_mode)
schema = await self.schema()
if blob_mode == "descriptions" or not schema_has_blob_field(schema):
arrow_tbl = await self.to_arrow()
if blob_mode == "descriptions":
arrow_tbl = strip_auto_row_ids(arrow_tbl, blob_v2_column_paths(schema))
return arrow_tbl.to_pandas(**kwargs)
_validate_blob_mode(blob_mode)
if blob_mode == "descriptions" or not _schema_has_blob_field(
await self.schema()
):
return (await self.to_arrow()).to_pandas(**kwargs)
if blob_mode == "lazy" and get_uri_scheme(await self.uri()) == "memory":
return (await self.to_arrow()).to_pandas(**kwargs)
if blob_mode == "bytes" and blob_v2_column_paths(schema):
return await self.query().to_pandas(blob_mode=blob_mode, **kwargs)
return (await self._to_lance()).to_pandas(blob_mode=blob_mode, **kwargs)
async def to_arrow(self) -> pa.Table:
@@ -4952,7 +4803,6 @@ class AsyncTable:
on_bad_vectors: Optional[OnBadVectorsType] = None,
fill_value: Optional[float] = None,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
) -> AddResult:
"""Add more data to the [Table](Table).
@@ -4977,12 +4827,6 @@ class AsyncTable:
progress: callable or tqdm-like, optional
A callback or tqdm-compatible progress bar. See
:meth:`Table.add` for details.
write_parallelism: int, optional
Number of partitions to write in parallel. Higher values increase
throughput but also peak memory use, since each partition buffers
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
"""
schema = await self.schema()
@@ -5014,12 +4858,7 @@ class AsyncTable:
data = to_scannable(data)
progress, owns = _normalize_progress(progress)
try:
return await self._inner.add(
data,
mode or "append",
progress=progress,
write_parallelism=write_parallelism,
)
return await self._inner.add(data, mode or "append", progress=progress)
except RuntimeError as e:
if "Cast error" in str(e):
raise ValueError(e)
@@ -5431,15 +5270,10 @@ class AsyncTable:
async_query = self._sync_query_to_async(query)
return await async_query.explain_plan(verbose)
async def _analyze_plan(
self,
query: Query,
*,
distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate",
) -> str:
async def _analyze_plan(self, query: Query) -> str:
# This method is used by the sync table
async_query = self._sync_query_to_async(query)
return await async_query.analyze_plan(distributed_metrics)
return await async_query.analyze_plan()
async def _output_schema(self, query: Query) -> pa.Schema:
async_query = self._sync_query_to_async(query)
@@ -5813,24 +5647,6 @@ class AsyncTable:
"""
return AsyncTakeQuery(self._inner.take_row_ids(row_ids), self)
async def blob_columns(self) -> list[str]:
return await self._inner.blob_columns()
async def fetch_blobs(
self, column: str, row_ids: Union[list[int], pa.Table]
) -> pa.LargeBinaryArray:
return await self._inner.fetch_blobs(
column, _normalize_blob_row_ids(row_ids, column)
)
async def fetch_blob_files(
self, column: str, row_ids: Union[list[int], pa.Table]
) -> "list[Optional[BlobFile]]":
handles = await self._inner.fetch_blob_files(
column, _normalize_blob_row_ids(row_ids, column)
)
return _wrap_blob_files(handles)
@property
def tags(self) -> AsyncTags:
"""Tag management for the dataset.
@@ -5932,24 +5748,6 @@ class AsyncTable:
"""
return await self._inner.list_indices()
async def tokenize(
self,
query: str,
*,
column: Optional[str] = None,
index_name: Optional[str] = None,
) -> Iterable[FtsToken]:
"""
Tokenize a query using the tokenizer configured on an FTS index.
Specify exactly one of ``column`` or ``index_name``.
Model-backed tokenizers such as ``jieba/*`` and ``lindera/*`` are
rebuilt in the client process from index metadata. For remote tables,
this means the same tokenizer model files must also exist locally.
"""
return await self._inner.tokenize(query, column=column, index_name=index_name)
async def index_stats(self, index_name: str) -> Optional[IndexStatistics]:
"""
Retrieve statistics about an index
@@ -6267,24 +6065,6 @@ class Branches:
"""Delete a branch."""
LOOP.run(self._table.branches.delete(name))
def diff(self, from_branch: str) -> Dict[str, Any]:
"""Diff a branch against main."""
return LOOP.run(self._table.branches.diff(from_branch))
def merge(self, from_branch: str, dry_run: bool = False) -> Dict[str, Any]:
"""Merge a branch into main, or dry-run.
Parameters
----------
from_branch: str
Branch to merge from.
dry_run: bool, default False
When True, only preview. When False, attempt the merge.
A rejected merge returns ``status="rejected"`` instead of raising.
"""
return LOOP.run(self._table.branches.merge(from_branch, dry_run))
def _wrap(
self, async_table: "AsyncTable", version: Optional[int] = None
) -> "Table":
@@ -6414,14 +6194,3 @@ class AsyncBranches:
async def delete(self, name: str) -> None:
"""Delete a branch."""
await self._table.branches.delete(name)
async def diff(self, from_branch: str) -> Dict[str, Any]:
"""Diff a branch against main."""
return await self._table.branches.diff(from_branch)
async def merge(self, from_branch: str, dry_run: bool = False) -> Dict[str, Any]:
"""Merge a branch into main, or dry-run.
A rejected merge returns ``status="rejected"`` instead of raising.
"""
return await self._table.branches.merge(from_branch, dry_run)

Some files were not shown because too many files have changed in this diff Show More