SELECT¶

Query data from Lance tables using SQL or DataFrames.

Basic Queries¶

SELECT * FROM users;

# Load table as DataFrame
users_df = spark.table("users")

# Use DataFrame operations
filtered_users = users_df.filter("age > 25").select("name", "email")
filtered_users.show()

// Load table as DataFrame
val usersDF = spark.table("users")

// Use DataFrame operations
val filteredUsers = usersDF.filter("age > 25").select("name", "email")
filteredUsers.show()

// Load table as DataFrame
Dataset<Row> usersDF = spark.table("users");

// Use DataFrame operations
Dataset<Row> filteredUsers = usersDF.filter("age > 25").select("name", "email");
filteredUsers.show();

Select Specific Columns¶

SELECT id, name, email FROM users;

Query with WHERE Clause¶

SELECT * FROM users WHERE age > 25;

Aggregate Queries¶

An ungrouped query with a single COUNT(*) or non-distinct COUNT(column) can be pushed down to Lance. COUNT(column) excludes nulls using IS NOT NULL and returns partial counts for Spark to combine, without transferring column values to Spark.

SELECT COUNT(email) FROM users WHERE age > 25;

Column counts currently require a top-level column and enabled filter pushdown. Nested fields, expressions, distinct counts, multiple aggregates, grouping, and queries with filters that Lance cannot evaluate remain in Spark. Column names containing backticks also remain in Spark.

SELECT department, COUNT(*) as employee_count
FROM users
GROUP BY department;

Join Queries¶

SELECT u.name, p.title
FROM users u
JOIN projects p ON u.id = p.user_id;

Querying Blob Columns¶

When querying tables with blob columns, the blob data itself is not materialized by default. Instead, you can access blob metadata through virtual columns. For each blob column, Lance provides two virtual columns:

• <column_name>__blob_pos - The byte position of the blob in the blob file
• <column_name>__blob_size - The size of the blob in bytes

These virtual columns can be used for:

• Monitoring blob storage statistics
• Filtering rows by blob size
• Implementing custom blob retrieval logic
• Verifying successful blob writes

SELECT id, title, content__blob_pos, content__blob_size
FROM documents
WHERE id = 1;

# Read table with blob column
documents_df = spark.table("documents")

# Access blob metadata using virtual columns
blob_metadata = documents_df.select(
    "id",
    "title",
    "content__blob_pos",
    "content__blob_size"
)
blob_metadata.show()

# Filter by blob size
large_blobs = documents_df.filter("content__blob_size > 1000000")
large_blobs.select("id", "title", "content__blob_size").show()

// Read table with blob column
val documentsDF = spark.table("documents")

// Access blob metadata using virtual columns
val blobMetadata = documentsDF.select(
  "id",
  "title",
  "content__blob_pos",
  "content__blob_size"
)
blobMetadata.show()

// Filter by blob size
val largeBlobs = documentsDF.filter("content__blob_size > 1000000")
largeBlobs.select("id", "title", "content__blob_size").show()

// Read table with blob column
Dataset<Row> documentsDF = spark.table("documents");

// Access blob metadata using virtual columns
Dataset<Row> blobMetadata = documentsDF.select(
    "id",
    "title",
    "content__blob_pos",
    "content__blob_size"
);
blobMetadata.show();

// Filter by blob size
Dataset<Row> largeBlobs = documentsDF.filter("content__blob_size > 1000000");
largeBlobs.select("id", "title", "content__blob_size").show();

Time Travel¶

Lance supports time travel queries, allowing you to query historical versions of your data using either a specific version number or a timestamp.

Query by Version¶

Use VERSION AS OF to query a specific version of the table:

-- Query version 5 of the table
SELECT * FROM users VERSION AS OF 5;

-- Query specific columns from a version
SELECT id, name FROM users VERSION AS OF 3;

# Query a specific version using DataFrame API
df = spark.read \
    .format("lance") \
    .option("version", "5") \
    .load("/path/to/dataset.lance")
df.show()

// Query a specific version using DataFrame API
val df = spark.read
    .format("lance")
    .option("version", "5")
    .load("/path/to/dataset.lance")
df.show()

// Query a specific version using DataFrame API
Dataset<Row> df = spark.read()
    .format("lance")
    .option("version", "5")
    .load("/path/to/dataset.lance");
df.show();

The version read option applies after Spark analyzes the current table schema. Select compatible columns, or use SQL VERSION AS OF to analyze the historical schema. Missing columns or incompatible types fail during planning.

Query by Tag¶

Use VERSION AS OF with a quoted tag name to query the snapshot referenced by a tag:

-- Query the snapshot referenced by the release_candidate tag
SELECT * FROM users VERSION AS OF 'release_candidate';

-- Query specific columns from a tagged snapshot
SELECT id, name FROM users VERSION AS OF 'v1.0';

tagged = spark.read.option("tag", "release_candidate").table("users")

tagged_path = spark.read \
    .format("lance") \
    .option("tag", "release_candidate") \
    .load("/path/to/dataset.lance")

val tagged = spark.read.option("tag", "release_candidate").table("users")

Dataset<Row> tagged = spark.read()
    .option("tag", "release_candidate")
    .table("users");

The explicit tag read option also accepts numeric tag names. Do not combine it with branch or version. Read options apply at scan time, after Spark analyzes the current table schema. You can select columns compatible with the tagged snapshot; missing columns or incompatible types (including nested fields) fail during planning. Use SQL VERSION AS OF to analyze the historical schema instead.

Tag queries are read-only; mutating commands are rejected. Besides DML and schema changes, this covers CREATE INDEX, DROP INDEX, OPTIMIZE, VACUUM, SET UNENFORCED PRIMARY KEY, and branch and tag DDL. Branch identifiers behave the same way.

Query by Branch¶

Read the current head of a named branch. Do not combine branch with tag or version.

SELECT * FROM catalog.db.users.branch_audit;

audit = spark.read.option("branch", "audit").table("catalog.db.users")

audit_path = spark.read \
    .format("lance") \
    .option("branch", "audit") \
    .load("/path/to/dataset.lance")

val audit = spark.read.option("branch", "audit").table("catalog.db.users")

val auditPath = spark.read
    .format("lance")
    .option("branch", "audit")
    .load("/path/to/dataset.lance")

Dataset<Row> audit = spark.read()
    .option("branch", "audit")
    .table("catalog.db.users");

Dataset<Row> auditPath = spark.read()
    .format("lance")
    .option("branch", "audit")
    .load("/path/to/dataset.lance");

If a table named users.branch_audit already exists, Spark reads that table. Otherwise the last segment is a branch on the parent table. Do not add VERSION AS OF or TIMESTAMP AS OF. Branch identifiers are read-only; mutating commands are rejected.

.option("branch").table(...) applies the branch at scan time. Spark still analyzes with the table schema. Use table.branch_name when the branch schema differs from the table.

Query by Timestamp¶

Use TIMESTAMP AS OF to query the table as it existed at a specific point in time:

-- Query the table as it was at a specific timestamp
SELECT * FROM users TIMESTAMP AS OF '2024-01-15 10:30:00';

-- Query with timestamp in epoch microseconds
SELECT * FROM users TIMESTAMP AS OF 1705314600000000;

# Query by timestamp using SQL
spark.sql("SELECT * FROM users TIMESTAMP AS OF '2024-01-15 10:30:00'").show()

// Query by timestamp using SQL
spark.sql("SELECT * FROM users TIMESTAMP AS OF '2024-01-15 10:30:00'").show()

// Query by timestamp using SQL
spark.sql("SELECT * FROM users TIMESTAMP AS OF '2024-01-15 10:30:00'").show();

Read Options¶

These options control how data is read from Lance datasets. They can be set using the .option() method when reading data.

Runtime Join Filtering¶

On Spark 4.1 and newer, Spark can push join dynamic partition pruning filters into Lance scans. Lance accepts IN filters on projected _rowaddr and _rowid columns, active sharding columns, and top-level scalar columns backed by a compatible scalar or zonemap index. Columns that Spark reads with lossy conversions are skipped (Timestamp(NANOSECOND), Date(MILLISECOND), and UInt64).

Runtime filters prune fragments from the pinned snapshot and also get passed to the native Lance scanner, while Spark still runs the full join. Multiple runtime filters combine with AND. Null keys are ignored, an empty or all-null key set returns an empty scan, and fragments not covered by a partially built zonemap index are kept safely. A filter with a negative _rowid or _rowaddr key is left to Spark.

You can control runtime filtering with a few read options:

• pushDownFilters=false disables runtime filtering completely.
• use_scalar_index=false skips native scalar index lookups, while keeping _rowaddr, sharding, and zonemap fragment pruning enabled.
• runtime_filter_max_keys and runtime_filter_max_bytes limit driver memory and task payload size. Since every Spark task carries a copy of the accepted filter, keep the fragment count in mind when raising these limits. You can also set them at the catalog level (for example spark.sql.catalog.lance.runtime_filter_max_keys=200000), and per-read .option() calls override them. If a filter hits a limit and gets skipped, rejectedRuntimeFilterCount increments in the scan metadata and the query still runs normally.

With Spark's default dynamicPartitionPruning.reuseBroadcastOnly=true, runtime filters only reach Lance when the other side of the join is broadcast and has a selective filter. Spark drops the filter on shuffle joins or when Lance is the broadcast side.

You can also use a _rowid runtime filter when another table (like an Iceberg table) stores Lance row IDs and joins back to Lance:

SELECT /*+ BROADCAST(o) */ i.image_bytes
FROM lance.default.images i
JOIN iceberg.db.observations o ON i._rowid = o.lance_row_id
WHERE o.ingestion_date = '2026-05-01';

Without stable row IDs, _rowid equals _rowaddr, so Lance prunes fragments. With enable_stable_row_ids, a row ID does not identify its fragment, so every fragment is scanned and the native scanner filters the matching rows.

User-Defined Types¶

Spark UDT columns (e.g. MLlib's VectorUDT) round-trip transparently. On write the column is unwrapped to the UDT's sqlType (a struct) for storage, and the UDT's fully qualified class name is stamped onto the Arrow field's user metadata under the __udt key. On read, that marker is consulted and the schema is rewrapped to the original UDT automatically. No configuration needed; the user-facing class (e.g. Vector for VectorUDT) is materialized in rows as usual.

Example: Reading with Options¶

-- Read a specific version using table options
SELECT * FROM lance.`/path/to/dataset.lance` VERSION AS OF 10;

# Reading with options
df = spark.read \
    .format("lance") \
    .option("batch_size", "1024") \
    .option("version", "5") \
    .load("/path/to/dataset.lance")

// Reading with options
val df = spark.read
    .format("lance")
    .option("batch_size", "1024")
    .option("version", "5")
    .load("/path/to/dataset.lance")

// Reading with options
Dataset<Row> df = spark.read()
    .format("lance")
    .option("batch_size", "1024")
    .option("version", "5")
    .load("/path/to/dataset.lance");

Example: Tuning Batch Size for Performance¶

# Larger batch size for better throughput on large scans
df = spark.read \
    .format("lance") \
    .option("batch_size", "4096") \
    .load("/path/to/dataset.lance")

// Larger batch size for better throughput on large scans
val df = spark.read
    .format("lance")
    .option("batch_size", "4096")
    .load("/path/to/dataset.lance")