Access OSS Tables through the Python SDK

Manage namespaces and the catalog

Catalog request/response models are all imported from lance_namespace:

from lance_namespace import (
    CreateNamespaceRequest, ListNamespacesRequest, DescribeNamespaceRequest,
    NamespaceExistsRequest, DropNamespaceRequest,
    DeclareTableRequest, CreateTableRequest, DescribeTableRequest,
    TableExistsRequest, ListTablesRequest, RenameTableRequest,
    DeregisterTableRequest, DropTableRequest,
)

DB = "my_db"

# Namespace (database) operations
ns.create_namespace(CreateNamespaceRequest(id=[DB]))
ns.namespace_exists(NamespaceExistsRequest(id=[DB]))
ns.describe_namespace(DescribeNamespaceRequest(id=[DB]))

ns.list_namespaces(ListNamespacesRequest(id=[]))            # id=[] means root

# List tables (returns bare table names, e.g. ["t1"])
ns.list_tables(ListTablesRequest(id=[DB]))

ns.list_all_tables(ListTablesRequest(id=[]))                # Across all namespaces

describe_table supports three optional parameters to obtain richer metadata:

DescribeTableRequest(id=[DB, "t1"], check_declared=True)         # Populates is_only_declared
DescribeTableRequest(id=[DB, "t1"], with_table_uri=True)         # Additionally returns table_uri
DescribeTableRequest(id=[DB, "t1"], load_detailed_metadata=True) # Returns version / schema / stats

Create a table and write data

Use lance.write_dataset to create a table and write data, passing in namespace_client and table_id:

import lance
import pyarrow as pa

TABLE_ID = ["my_db", "my_table"]

# Create table + write
lance.write_dataset(
    pa.table({"id": [1, 2, 3], "name": ["a", "b", "c"]}),
    namespace_client=ns, table_id=TABLE_ID,
    mode="create", storage_options=STORAGE_OPTIONS,
)

# Append
lance.write_dataset(
    pa.table({"id": [4, 5], "name": ["d", "e"]}),
    namespace_client=ns, table_id=TABLE_ID,
    mode="append", storage_options=STORAGE_OPTIONS,
)

mode="create" creates a new table; mode="append" / "overwrite" require the table to already exist.

You can also create a table directly through the catalog with an Arrow IPC byte stream (the server writes the data, so no data plane credentials are required):

import io
tbl = pa.table({"id": [1, 2, 3]})
sink = io.BytesIO()
with pa.ipc.new_stream(sink, tbl.schema) as w:
    w.write_table(tbl)
ns.create_table(CreateTableRequest(id=["my_db", "t2"]), sink.getvalue())

Read data and time travel

Open a table to read the current version, or specify a historical version number to travel back to the data state before a given change.

# Read the current version
ds = lance.dataset(namespace_client=ns, table_id=TABLE_ID, storage_options=STORAGE_OPTIONS)
print(ds.count_rows(), ds.version)
print(ds.to_table().to_pydict())

# Time travel: read a historical version
old = lance.dataset(namespace_client=ns, table_id=TABLE_ID, version=1,
                    storage_options=STORAGE_OPTIONS)
print(old.count_rows())

Row-level DML (update / delete / merge)

Update or delete specific rows by predicate, or perform an upsert on a join key: matched rows are updated and unmatched rows are inserted.

ds = lance.dataset(namespace_client=ns, table_id=TABLE_ID, storage_options=STORAGE_OPTIONS)

# Update: updates is {column_name: SQL expression}; string literals need inner single quotes (e.g. "'updated'").
ds.update({"name": "'updated'"}, where="id = 1")           # Set name to the string "updated"
ds.delete("id = 5")                                        # Delete (predicate)

# merge (upsert)
ds.merge_insert("id") \
  .when_matched_update_all() \
  .when_not_matched_insert_all() \
  .execute(pa.table({"id": [3, 6], "name": ["c-new", "f"]}))

Scalar index, vector index, and search

Build a B-tree index on a regular column to speed up point lookups, an IVF_PQ index on a vector column for approximate nearest neighbor search, and an inverted index on a text column for full-text search.

import numpy as np

# Write a table with a vector column (IVF_PQ training has data-volume requirements; the example writes 70,000 rows).
dim = 128
n = 70000
rng = np.random.default_rng(42)
data = pa.table({
    "id": pa.array(range(n), pa.int64()),
    "text": pa.array([f"doc number {i} category {i % 7}" for i in range(n)], pa.string()),
    "vector": pa.FixedSizeListArray.from_arrays(
        pa.array(rng.random((n, dim), dtype=np.float32).reshape(-1), pa.float32()), dim),
})
lance.write_dataset(data, namespace_client=ns, table_id=["my_db", "vecs"],
                    mode="create", storage_options=STORAGE_OPTIONS)
vds = lance.dataset(namespace_client=ns, table_id=["my_db", "vecs"],
                    storage_options=STORAGE_OPTIONS)

# Scalar index (BTREE)
vds.create_scalar_index("id", index_type="BTREE")
vds.to_table(filter="id = 12345")            # Point lookup

# Vector index (IVF_PQ) + vector search
vds.create_index("vector", index_type="IVF_PQ", num_partitions=16, num_sub_vectors=16)
qv = rng.random((dim,), dtype=np.float32)
res = vds.to_table(nearest={"column": "vector", "q": qv, "k": 5})
print(res["id"].to_pylist(), [round(d, 4) for d in res["_distance"].to_pylist()])

# Full-text search (INVERTED inverted index)
vds.create_scalar_index("text", index_type="INVERTED")
hits = vds.to_table(full_text_query="category", limit=5)   # Result includes the _score relevance column
print(hits.column_names)                                    # ['id', 'text', 'vector', '_score']
# You can also specify the search columns: full_text_query={"columns": ["text"], "query": "category"}

Schema evolution

Add columns, drop columns, rename columns, or change column types without rewriting the entire table; this only produces new version metadata.

# Add a column (value is a SQL expression; you can also pass a BatchUDF / Arrow schema)
vds.add_columns({"score": "CAST(id AS DOUBLE)"})
# Drop a column
vds.drop_columns(["score"])
# Rename (alter_columns takes path + target attribute)
vds.alter_columns({"path": "text", "name": "content"})
# Change column type (int32 -> int64)
vds.add_columns({"n32": "CAST(id AS INT)"})                   # n32 is int32
vds.alter_columns({"path": "n32", "data_type": pa.int64()})   # int32 -> int64

Small-file compaction and version cleanup

Compact small files produced by writes to improve read performance, incrementally update indexes to cover new data, and clean up historical version files that are no longer referenced to release storage.

# Compact small files
vds.optimize.compact_files()
# Incrementally train / merge indexes (newly written data is brought into existing indexes)
vds.optimize.optimize_indices()

# Clean up historical version files no longer referenced
vds.cleanup_old_versions()

Drop tables and namespaces

Drop a table (including its data; irreversible) or only deregister the table metadata while keeping the underlying data. A namespace can be dropped only after all tables beneath it have been cleared.

# Drop tables (data is irreversible)
for table_name in ["vecs", "t2"]:
    ns.drop_table(DropTableRequest(id=["my_db", table_name]))

# For my_table, choose one of the two:
# Option 1: Only deregister catalog metadata; underlying data is kept
ns.deregister_table(DeregisterTableRequest(id=["my_db", "my_table"]))
# Option 2: If you are sure the data is no longer needed, use instead:
# ns.drop_table(DropTableRequest(id=["my_db", "my_table"]))

# After the catalog no longer contains the table, drop the namespace
ns.drop_namespace(DropNamespaceRequest(id=["my_db"]))

Error handling

Catch catalog call failures by the standard exception types under lance_namespace.errors and perform idempotent retries or user-friendly prompts.

from lance_namespace import NamespaceAlreadyExistsError

try:
    ns.create_namespace(CreateNamespaceRequest(id=["my_db"]))
except NamespaceAlreadyExistsError:
    pass