Compare commits

...
Author SHA1 Message Date
Chris Lu 3466e74391 test: drive the Lance namespace with Spark
The counterpart of catalog_spark, which does this for the Iceberg REST
catalog. Spark is the engine most likely to be pointed at a lakehouse,
and it reaches the Lance catalog through the connector's DSV2 catalog -
org.lance.spark.LanceNamespaceSparkCatalog with impl=rest - over the same
routes every other client uses.

    SHOW NAMESPACES -> ['`sparklance-lcephd80`.ml']
    SHOW TABLES -> ['sparklance-lcephd80$ml$embeddings']
    count -> 3
    filtered -> [(2, 'two'), (3, 'three')]
    count after a second commit -> 4

The second insert is there on purpose: a store that cannot order commits
fails on the second one, not the first.

Two things the run settled that were guesses beforehand. CREATE TABLE
works, because the connector declares through the namespace and writes the
data itself rather than pushing Arrow at the server. And SHOW TABLES
returns the namespace's own identifiers - bucket, namespace and name
joined by the delimiter - not bare Spark table names.

Credentials go under the catalog's storage.* prefix, which is handed to
lance as object_store options; a gateway without STS vends none, the same
trap the LanceDB suite documents.

Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm
2026-08-21 12:32:59 -07:00
Chris Lu 49a7d4ff6a test: pin the LanceDB client, and index before searching
Three from review.

The client's dependencies were unpinned, so an unrelated upstream release
could change what an old commit reproduces. Pinned to the versions this
suite was verified against; the client is as much the thing under test as
the server.

The search was called ANN and was not: without an index LanceDB scans.
The test now builds an IVF_PQ index over 1024 rows first, which is worth
more than the wording fix - an index writes into a directory of the table
that the S3 door has to admit, and that guard has refused a Lance
directory before. It builds, covers all 1024 rows, and searches.

The assertion moved with it. Demanding the exact nearest neighbour was
right for a brute-force scan and wrong for a quantized index, which
answered 0 as readily as 1; both are correct, so the check is now the
neighbourhood.

And the pushdown check accepted any failure. It now requires the refusal
to be the catalog's Unsupported and requires that nothing was left behind,
or, when the client falls back, that the table is complete.

Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm
2026-08-20 19:51:55 -07:00
Chris Lu 2481641abf test: drive the Lance namespace with LanceDB
The Iceberg catalog is checked against Spark, Trino, ClickHouse, Doris,
Dremio and RisingWave. The Lance one had only its own reference client,
which is the same thing as checking it against ourselves.

LanceDB connects with connect_namespace("rest", ...), which speaks the
routes this catalog implements, so the suite exercises the protocol rather
than our idea of it: list the catalog, open a table through it, read the
schema, run a vector search and a filtered scan, create a table, and read
the same dataset straight off its URI with no catalog at all.

    table_names -> ['lancedb-p0guidmm$ml$embeddings']
    open_table  -> 64 rows
    search      -> [1, 0, 2]
    create_table -> 4 rows, listed by the catalog
    direct read without the catalog -> 64 rows

Seeding is pylance, because the namespace records where a table lives and
does not carry its data. That split is the design rather than a limit of
the test.

One interop note the test encodes: a gateway without STS vends
storage_options carrying an endpoint and a region but no credentials, and
LanceDB uses what the namespace vends on some paths. The container gets
credentials in its environment as well, which is what a deployment without
STS would do.

Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm
2026-08-20 12:38:06 -07:00
8 changed files with 1248 additions and 0 deletions
+140
View File
@@ -874,6 +874,146 @@ jobs:
path: test/s3tables/unity_catalog/test-output.log
retention-days: 3
lancedb-namespace-tests:
name: LanceDB Namespace Integration Tests
runs-on: ubuntu-22.04
timeout-minutes: 30
steps:
- name: Check out code
uses: actions/checkout@v7
- name: Set up Go
uses: actions/setup-go@v7
with:
go-version-file: 'go.mod'
id: go
- name: Configure Docker Hub mirror
run: |
echo '{"registry-mirrors": ["https://mirror.gcr.io"]}' | sudo tee /etc/docker/daemon.json
sudo systemctl restart docker
- name: Pre-pull images
run: |
pull() { for i in 1 2 3; do docker pull "$1" && return 0; sleep 15; done; return 1; }
pull python:3.11-slim
- name: Run go mod tidy
run: go mod tidy
- name: Build SeaweedFS
run: |
cd weed && go build -buildvcs=false .
- name: Run LanceDB Namespace Integration Tests
timeout-minutes: 25
working-directory: test/s3tables/catalog_lancedb
run: |
set -x
set -o pipefail
echo "=== System Information ==="
uname -a
free -h
df -h
docker info
echo "=== Starting LanceDB Namespace Tests ==="
go test -v -timeout 20m . 2>&1 | tee test-output.log || {
echo "LanceDB namespace integration tests failed"
exit 1
}
- name: Show test output on failure
if: failure()
working-directory: test/s3tables/catalog_lancedb
run: |
echo "=== Test Output ==="
if [ -f test-output.log ]; then
tail -200 test-output.log
fi
echo "=== Process information ==="
ps aux | grep -E "(weed|test|docker)" || true
- name: Upload test logs on failure
if: failure()
uses: actions/upload-artifact@v7
with:
name: lancedb-namespace-test-logs
path: test/s3tables/catalog_lancedb/test-output.log
retention-days: 3
spark-lance-namespace-tests:
name: Spark Lance Namespace Integration Tests
runs-on: ubuntu-22.04
timeout-minutes: 40
steps:
- name: Check out code
uses: actions/checkout@v7
- name: Set up Go
uses: actions/setup-go@v7
with:
go-version-file: 'go.mod'
id: go
- name: Configure Docker Hub mirror
run: |
echo '{"registry-mirrors": ["https://mirror.gcr.io"]}' | sudo tee /etc/docker/daemon.json
sudo systemctl restart docker
- name: Pre-pull images
run: |
pull() { for i in 1 2 3; do docker pull "$1" && return 0; sleep 15; done; return 1; }
pull apache/spark:3.5.1
- name: Run go mod tidy
run: go mod tidy
- name: Build SeaweedFS
run: |
cd weed && go build -buildvcs=false .
- name: Run Spark Lance Namespace Integration Tests
timeout-minutes: 35
working-directory: test/s3tables/catalog_spark_lance
run: |
set -x
set -o pipefail
echo "=== System Information ==="
uname -a
free -h
df -h
docker info
echo "=== Starting Spark Lance Namespace Tests ==="
go test -v -timeout 30m . 2>&1 | tee test-output.log || {
echo "Spark Lance namespace integration tests failed"
exit 1
}
- name: Show test output on failure
if: failure()
working-directory: test/s3tables/catalog_spark_lance
run: |
echo "=== Test Output ==="
if [ -f test-output.log ]; then
tail -200 test-output.log
fi
echo "=== Process information ==="
ps aux | grep -E "(weed|test|docker|spark)" || true
- name: Upload test logs on failure
if: failure()
uses: actions/upload-artifact@v7
with:
name: spark-lance-namespace-test-logs
path: test/s3tables/catalog_spark_lance/test-output.log
retention-days: 3
s3-tables-build-verification:
name: S3 Tables Build Verification
runs-on: ubuntu-22.04
@@ -0,0 +1,22 @@
# LanceDB client for the SeaweedFS Lance Namespace integration test.
#
# The Iceberg suites point Spark, Trino and ClickHouse at the catalog; this is
# the same idea for Lance. LanceDB connects with connect_namespace("rest", ...),
# which speaks the routes this catalog implements, so what it exercises is the
# protocol rather than requests we wrote ourselves.
FROM python:3.11-slim
WORKDIR /app
# Pinned to the versions this suite was verified against. An unrelated upstream
# release should not be able to change what an old commit reproduces, and this
# client is the thing under test as much as the server is.
RUN pip install --no-cache-dir \
"lancedb==0.37.1" \
"lance-namespace==0.8.6" \
"pylance==10.0.0" \
"pyarrow==25.0.1"
COPY lancedb_ops.py /app/
CMD ["python3", "/app/lancedb_ops.py", "--help"]
+56
View File
@@ -0,0 +1,56 @@
# LanceDB Integration Test
Drives the SeaweedFS Lance Namespace with [LanceDB](https://lancedb.com), the way
`catalog_spark`, `catalog_trino` and `catalog_clickhouse` drive the Iceberg REST
catalog with their engines.
## Why a real client
Every serious bug in this catalog so far looked correct to a request written by
hand: a deregister that deleted the dataset, an S3 door that refused every Lance
file, a namespace that listed tables it would then deny. A hand-built HTTP test
checks the shape of a response. A client checks whether the response is *usable* —
that the location it hands back, the storage options beside it and the layout
rules on the S3 door all line up at once.
LanceDB connects with `connect_namespace("rest", ...)`, which speaks the routes
this catalog implements, so what runs here is the protocol rather than our own
idea of it.
## What it does
`TestLanceDBNamespace`:
1. Starts a `weed mini` cluster with S3 and the Lance Namespace enabled.
2. Creates a table bucket declared `LANCE`, so the catalog refuses tables of any
other format in it.
3. Builds `Dockerfile.client` (LanceDB, pylance, lance-namespace) and runs
`lancedb_ops.py` against the namespace.
Inside the container:
| Step | What it proves |
| --- | --- |
| seed a table | the namespace's location and credentials are enough to write |
| `table_names` | the catalog is browsable through LanceDB |
| `open_table` | LanceDB resolves a table through the catalog and reads it |
| schema check | the vector column survived the round trip |
| `create_index` | an IVF_PQ index builds, and its files land through the S3 door's layout rules |
| `search(...)` | approximate search over that index, on data behind SeaweedFS |
| `where("id < 5")` | so does the scan path, not only the index |
| `create_table` | LanceDB declares through the namespace and writes the data itself |
| `create_table` with pushdown | either the client falls back and the table is complete, or the catalog refuses with the spec's Unsupported and leaves nothing behind |
| direct `lance.dataset(uri)` | the catalog stays optional; the dataset opens without it |
The seeding is pylance rather than LanceDB, because the namespace records where
a table lives and does not carry its data. That split is the design, not a
limitation of the test.
## Running it
cd test/s3tables/catalog_lancedb
(cd ../../../weed && go build .) # the harness runs this binary
go test -run TestLanceDBNamespace -v -timeout 30m .
Skipped without Docker, and in `-short` mode. The first run builds the client
image, which takes a few minutes; later runs reuse it.
@@ -0,0 +1,291 @@
// Package lancedb drives the SeaweedFS Lance Namespace with LanceDB, the way
// the catalog_spark, catalog_trino and catalog_clickhouse suites drive the
// Iceberg REST catalog with their engines.
//
// A catalog is only as good as what a real client can do with it. Every serious
// bug in this surface so far - a deregister that deleted the dataset, an S3 door
// that refused every Lance file, a namespace that listed tables it would then
// deny - looked correct to a request written by hand.
package lancedb
import (
"context"
"fmt"
"math/rand"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/test/testutil"
)
const (
clientImage = "seaweedfs-lancedb-test"
startupTimeout = 60 * time.Second
clientTimeout = 15 * time.Minute
)
// TestLanceDBNamespace runs LanceDB against the namespace end to end: it lists
// what the catalog holds, opens a table through it, searches the vectors, and
// reads the same dataset straight off its URI to show the catalog stays
// optional.
func TestLanceDBNamespace(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
if !hasDocker() {
t.Skip("Docker not available, skipping LanceDB integration test")
}
env := newEnvironment(t)
defer env.cleanup()
env.start(t)
bucket := "lancedb-" + randomSuffix()
env.createTableBucket(t, bucket)
buildClientImage(t)
env.runClient(t, bucket)
}
type environment struct {
weedBinary string
dataDir string
bindIP string
masterPort int
masterGrpcPort int
volumePort int
volumeGrpcPort int
filerPort int
filerGrpcPort int
s3Port int
s3GrpcPort int
lancePort int
accessKey string
secretKey string
weedCancel context.CancelFunc
weedCmd *exec.Cmd
}
func newEnvironment(t *testing.T) *environment {
t.Helper()
wd, err := os.Getwd()
if err != nil {
t.Fatalf("get working directory: %v", err)
}
seaweedDir := wd
for i := 0; i < 5; i++ {
if _, err := os.Stat(filepath.Join(seaweedDir, "go.mod")); err == nil {
break
}
seaweedDir = filepath.Dir(seaweedDir)
}
weedBinary := filepath.Join(seaweedDir, "weed", "weed")
if info, statErr := os.Stat(weedBinary); statErr == nil && !info.IsDir() {
// `make test` builds first; a plain `go test` will otherwise drive a
// binary from days ago and report a pass for code it never ran.
t.Logf("using %s, built %s", weedBinary, info.ModTime().Format(time.RFC3339))
} else {
weedBinary = "weed"
if _, err := exec.LookPath(weedBinary); err != nil {
t.Skip("weed binary not found, skipping integration test")
}
}
dataDir, err := os.MkdirTemp("", "seaweed-lancedb-test-*")
if err != nil {
t.Fatalf("create temp dir: %v", err)
}
ports := testutil.MustAllocatePorts(t, 9)
return &environment{
weedBinary: weedBinary,
dataDir: dataDir,
bindIP: testutil.FindBindIP(),
masterPort: ports[0],
masterGrpcPort: ports[1],
volumePort: ports[2],
volumeGrpcPort: ports[3],
filerPort: ports[4],
filerGrpcPort: ports[5],
s3Port: ports[6],
s3GrpcPort: ports[7],
lancePort: ports[8],
accessKey: "AKIAIOSFODNN7EXAMPLE",
secretKey: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
}
}
func (env *environment) start(t *testing.T) {
t.Helper()
iamConfigPath, err := testutil.WriteIAMConfig(env.dataDir, env.accessKey, env.secretKey)
if err != nil {
t.Fatalf("write IAM config: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
env.weedCancel = cancel
cmd := exec.CommandContext(ctx, env.weedBinary, "mini",
"-master.port", fmt.Sprintf("%d", env.masterPort),
"-master.port.grpc", fmt.Sprintf("%d", env.masterGrpcPort),
"-volume.port", fmt.Sprintf("%d", env.volumePort),
"-volume.port.grpc", fmt.Sprintf("%d", env.volumeGrpcPort),
"-filer.port", fmt.Sprintf("%d", env.filerPort),
"-filer.port.grpc", fmt.Sprintf("%d", env.filerGrpcPort),
"-s3.port", fmt.Sprintf("%d", env.s3Port),
"-s3.port.grpc", fmt.Sprintf("%d", env.s3GrpcPort),
"-s3.port.lance", fmt.Sprintf("%d", env.lancePort),
"-s3.config", iamConfigPath,
"-ip", env.bindIP,
"-ip.bind", "0.0.0.0",
"-dir", env.dataDir,
)
cmd.Dir = env.dataDir
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
cmd.Env = append(os.Environ(),
"AWS_ACCESS_KEY_ID="+env.accessKey,
"AWS_SECRET_ACCESS_KEY="+env.secretKey,
)
if err := cmd.Start(); err != nil {
t.Fatalf("start SeaweedFS: %v", err)
}
env.weedCmd = cmd
// The namespace answers /v1/table once it is serving, which is a cheaper
// readiness check than waiting on a bucket that does not exist yet.
url := fmt.Sprintf("http://%s:%d/v1/table", env.bindIP, env.lancePort)
if !waitForHTTP(url, startupTimeout) {
t.Fatalf("the Lance namespace did not become ready at %s", url)
}
}
// waitForHTTP polls until the URL answers at all. An auth refusal counts: it
// means the server is up, which is the only thing being waited on.
func waitForHTTP(url string, timeout time.Duration) bool {
client := &http.Client{Timeout: 2 * time.Second}
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
resp, err := client.Get(url)
if err != nil {
time.Sleep(500 * time.Millisecond)
continue
}
status := resp.StatusCode
resp.Body.Close()
if status < 500 {
return true
}
time.Sleep(500 * time.Millisecond)
}
return false
}
func (env *environment) cleanup() {
if env.weedCancel != nil {
env.weedCancel()
}
if env.weedCmd != nil {
_ = env.weedCmd.Wait()
}
if env.dataDir != "" {
_ = os.RemoveAll(env.dataDir)
}
}
// createTableBucket makes the bucket LanceDB will read through, declared LANCE
// so the catalog refuses anything of another format in it.
func (env *environment) createTableBucket(t *testing.T, bucket string) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
cmd := exec.CommandContext(ctx, env.weedBinary, "shell",
fmt.Sprintf("-master=%s:%d.%d", env.bindIP, env.masterPort, env.masterGrpcPort),
)
cmd.Stdin = strings.NewReader(fmt.Sprintf(
"s3tables.bucket -create -name %s -format LANCE -account 000000000000\nexit\n", bucket))
if out, err := cmd.CombinedOutput(); err != nil {
t.Fatalf("create table bucket %s: %v\n%s", bucket, err, out)
}
t.Logf("created LANCE table bucket %s", bucket)
}
func buildClientImage(t *testing.T) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
cmd := exec.CommandContext(ctx, "docker", "build",
"-t", clientImage, "-f", "Dockerfile.client", ".")
if out, err := cmd.CombinedOutput(); err != nil {
t.Fatalf("build the LanceDB client image: %v\n%s", err, out)
}
}
func (env *environment) runClient(t *testing.T, bucket string) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), clientTimeout)
defer cancel()
// The container reaches the gateway through the host gateway address, which
// is not the address the namespace advertises to its own clients.
namespaceURL := fmt.Sprintf("http://host.docker.internal:%d", env.lancePort)
s3Endpoint := fmt.Sprintf("http://host.docker.internal:%d", env.s3Port)
cmd := exec.CommandContext(ctx, "docker", "run", "--rm",
"--add-host", "host.docker.internal:host-gateway",
// Also in the environment, not only in storage_options: LanceDB takes
// some paths through the options the namespace vends, and a gateway
// without STS vends none, leaving lance's provider chain to find them.
"-e", "AWS_ACCESS_KEY_ID="+env.accessKey,
"-e", "AWS_SECRET_ACCESS_KEY="+env.secretKey,
"-e", "AWS_REGION=us-east-1",
"-e", "AWS_ENDPOINT_URL="+s3Endpoint,
"-e", "AWS_ALLOW_HTTP=true",
clientImage,
"python3", "/app/lancedb_ops.py",
"--namespace-url", namespaceURL,
"--s3-endpoint", s3Endpoint,
"--bucket", bucket,
"--access-key", env.accessKey,
"--secret-key", env.secretKey,
)
out, err := cmd.CombinedOutput()
t.Logf("LanceDB client output:\n%s", out)
if err != nil {
t.Fatalf("the LanceDB client failed: %v", err)
}
if !strings.Contains(string(out), "PASS") {
t.Fatalf("the LanceDB client did not report PASS")
}
}
func hasDocker() bool {
return exec.Command("docker", "version").Run() == nil
}
func randomSuffix() string {
const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
suffix := make([]byte, 8)
for i := range suffix {
suffix[i] = charset[rand.Intn(len(charset))]
}
return string(suffix)
}
@@ -0,0 +1,244 @@
#!/usr/bin/env python3
"""Drive the SeaweedFS Lance Namespace with LanceDB.
The existing client test uses `lance_namespace` and `pylance` directly, which is
the protocol's reference client. LanceDB is what people actually point at a
catalog: it connects with `connect_namespace("rest", ...)`, lists what is there,
opens a table and searches it. This checks the catalog against that path, the
way the Spark, Trino and ClickHouse suites check the Iceberg one.
Everything it prints is either "PASS" or a line starting with FAIL, so the Go
harness can report the first real failure rather than a stack trace.
"""
import argparse
import sys
import warnings
warnings.filterwarnings("ignore")
import lance
import lance_namespace as ln
import lancedb
import pyarrow as pa
DIM = 8
def sample_rows(count):
"""A vector table, which is the only kind worth putting in Lance."""
return pa.table(
{
"id": pa.array(list(range(count)), type=pa.int64()),
"title": pa.array([f"row-{i}" for i in range(count)]),
"vector": pa.array(
[[float(i) + d for d in range(DIM)] for i in range(count)],
type=pa.list_(pa.float32(), DIM),
),
}
)
def seed_table(namespace_url, storage, bucket, namespace, table, rows):
"""Declares a table through the namespace and writes a dataset into it.
LanceDB reads through the catalog; the writing half is pylance, because the
namespace records where a table lives and does not carry its data.
"""
ns = ln.connect("rest", {"uri": namespace_url})
ns.create_namespace(
ln.CreateNamespaceRequest(id=[bucket], mode="EXIST_OK")
)
ns.create_namespace(
ln.CreateNamespaceRequest(id=[bucket, namespace], mode="EXIST_OK")
)
table_id = [bucket, namespace, table]
declared = ns.declare_table(ln.DeclareTableRequest(id=table_id))
lance.write_dataset(
sample_rows(rows), declared.location, storage_options=storage, mode="overwrite"
)
print(f"seeded {rows} rows at {declared.location}")
return declared.location
def check(condition, message):
if not condition:
print(f"FAIL: {message}", file=sys.stderr)
raise SystemExit(1)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--namespace-url", required=True)
parser.add_argument("--s3-endpoint", required=True)
parser.add_argument("--bucket", required=True)
parser.add_argument("--namespace", default="ml")
parser.add_argument("--table", default="embeddings")
# Enough rows for an IVF_PQ index to be worth building, which is the point
# of step 4: without one, a search is a brute-force scan.
parser.add_argument("--rows", type=int, default=1024)
parser.add_argument("--access-key", default="any")
parser.add_argument("--secret-key", default="any")
args = parser.parse_args()
# The namespace vends an endpoint correct for its own host; a container
# reaches the same gateway by another name, so the endpoint is overridden
# here and the credentials filled in for a deployment without STS.
storage = {
"aws_endpoint": args.s3_endpoint,
"allow_http": "true",
"aws_access_key_id": args.access_key,
"aws_secret_access_key": args.secret_key,
"aws_region": "us-east-1",
}
location = seed_table(
args.namespace_url, storage, args.bucket, args.namespace, args.table, args.rows
)
print(f"lancedb {lancedb.__version__} connecting to {args.namespace_url}")
db = lancedb.connect_namespace(
"rest", {"uri": args.namespace_url}, storage_options=storage
)
# 1. The catalog is browsable: the bucket is a namespace, and the table is
# in it under the name the namespace gave it.
tables = list(db.table_names(namespace_path=[args.bucket, args.namespace], limit=100))
print(f"table_names -> {tables}")
check(
any(args.table in name for name in tables),
f"{args.table} is not listed in {tables}",
)
# 2. Opening it goes through the catalog: LanceDB asks the namespace where
# the table is and reads it from there.
# storage_options is passed per call as well as on the connection: what the
# namespace vends for a table is merged in, and a deployment without STS
# vends no credentials, which is what the client would otherwise be left with.
table = db.open_table(
args.table,
namespace_path=[args.bucket, args.namespace],
storage_options=storage,
)
count = table.count_rows()
print(f"open_table -> {count} rows")
check(count == args.rows, f"read {count} rows, want {args.rows}")
# 3. The schema survived the round trip, vector column included. This is the
# part a catalog that only records a location cannot fake.
names = table.schema.names
print(f"schema -> {names}")
check("vector" in names and "title" in names, f"schema lost columns: {names}")
# 4. Build a vector index, then search it. Both halves matter: an index
# writes files into a directory of the table the S3 door has to admit -
# the layout guard has refused a Lance directory before - and without one
# a search is a brute-force scan that proves nothing about the index path
# the maintenance worker exists to keep in shape.
table.create_index(
metric="l2",
vector_column_name="vector",
index_type="IVF_PQ",
num_partitions=1,
num_sub_vectors=4,
)
indices = table.list_indices()
print(f"create_index -> {indices}")
check(len(indices) >= 1, "no index was created")
# Vectors are laid out so that id N sits near id N+1, so a query built from
# id 1 should come back with its neighbourhood. The assertion is a
# neighbourhood and not an exact id: an IVF_PQ index quantizes, so the
# nearest hit is approximate by construction - with this data it answers 0
# as readily as 1, and both are right.
query = [float(1) + d for d in range(DIM)]
hits = table.search(query).limit(3).to_list()
ids = [hit["id"] for hit in hits]
print(f"search -> {ids}")
check(len(hits) == 3, f"search returned {len(hits)} hits, want 3")
check(
all(i <= 5 for i in ids),
f"search returned {ids}, which is not the neighbourhood of the query",
)
# 5. A filtered scan, so it is not only the ANN path that works.
filtered = table.search().where("id < 5").limit(10).to_list()
print(f"filtered scan -> {len(filtered)} rows")
check(len(filtered) == 5, f"filter returned {len(filtered)} rows, want 5")
# 6. Creating a table. By default LanceDB declares it through the namespace
# and writes the data itself, which is exactly the split this catalog
# serves, so this has to work.
created = db.create_table(
"created_by_lancedb",
data=sample_rows(4),
namespace_path=[args.bucket, args.namespace],
storage_options=storage,
)
check(created.count_rows() == 4, "create_table wrote the wrong number of rows")
listed = list(db.table_names(namespace_path=[args.bucket, args.namespace], limit=100))
check(
any("created_by_lancedb" in name for name in listed),
f"a table created through LanceDB is not listed: {listed}",
)
print(f"create_table -> {created.count_rows()} rows, listed by the catalog")
# 7. The same creation with server-side pushdown, which asks the namespace
# to run CreateTable itself. That operation carries Arrow data and this
# catalog answers the spec's Unsupported for it. What matters is that the
# client is left with something coherent either way - it falls back to
# declare-and-write - rather than a hang, a 404, or a half-made table.
pushdown = lancedb.connect_namespace(
"rest",
{"uri": args.namespace_url},
storage_options=storage,
namespace_client_pushdown_operations=["CreateTable"],
)
pushed_error = None
try:
pushdown.create_table(
"pushed_by_lancedb",
data=sample_rows(2),
namespace_path=[args.bucket, args.namespace],
storage_options=storage,
)
except Exception as err: # noqa: BLE001 - the point is what the client sees
pushed_error = err
after = list(db.table_names(namespace_path=[args.bucket, args.namespace], limit=100))
landed = any("pushed_by_lancedb" in name for name in after)
print(f"create_table with pushdown: error={pushed_error!r}, catalog has it={landed}")
if pushed_error is None:
# The client fell back to declare-and-write, so the table is real and
# has to be readable and complete.
check(landed, "create_table reported success but the catalog has no table")
rows = db.open_table(
"pushed_by_lancedb",
namespace_path=[args.bucket, args.namespace],
storage_options=storage,
).count_rows()
check(rows == 2, f"the pushed table holds {rows} rows, want 2")
else:
# Refused, which is what this catalog answers for a data-plane
# operation. It has to be that refusal and not some other failure, and
# it must not have left a half-made table behind.
message = str(pushed_error).lower()
check(
"unsupported" in message or "501" in message or "not implemented" in message,
f"pushdown failed for an unexpected reason: {pushed_error}",
)
check(not landed, "a refused create left a table behind in the catalog")
# 8. And the dataset is still readable straight off its URI, which is what
# keeps the catalog optional.
direct = lance.dataset(location, storage_options=storage).count_rows()
check(direct == args.rows, f"direct read got {direct} rows, want {args.rows}")
print(f"direct read without the catalog -> {direct} rows")
print("PASS")
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,61 @@
# Spark Lance Integration Test
Drives the SeaweedFS Lance Namespace with Spark, through the Lance Spark
connector's DSV2 catalog. The Lance counterpart of `catalog_spark`, which does
the same for the Iceberg REST catalog.
## Why Spark
Spark is the engine most likely to be pointed at a lakehouse, and it reaches the
catalog over the same Lance Namespace routes as every other client. Between this
and `catalog_lancedb`, the catalog is exercised by the two clients people
actually use, rather than only by the protocol's reference implementation.
## What it does
`TestSparkLanceNamespace`:
1. Starts a `weed mini` cluster with S3 and the Lance Namespace enabled.
2. Creates a table bucket declared `LANCE`.
3. Runs `spark_lance_ops.py` inside the stock `apache/spark:3.5.1` image, with
the connector pulled from Maven at submit time.
Inside Spark:
| Step | What it proves |
| --- | --- |
| `CREATE NAMESPACE` / `SHOW NAMESPACES` | the bucket is the first level of the path, and the catalog is writable |
| `CREATE TABLE ... USING lance` | the connector declares through the namespace and writes the data itself |
| `INSERT` / `SELECT count(*)` | a write lands and reads back through the catalog |
| schema check | the vector column survived the round trip |
| `WHERE id >= 2` | the filter path, not only a full scan |
| a second `INSERT` | the commit a store that cannot order commits fails on |
## Configuration
The connector's catalog properties, for reference:
```
spark.sql.catalog.lance org.lance.spark.LanceNamespaceSparkCatalog
spark.sql.catalog.lance.impl rest
spark.sql.catalog.lance.uri http://seaweed:9101
spark.sql.catalog.lance.storage.aws_endpoint http://seaweed:8333
spark.sql.catalog.lance.storage.allow_http true
spark.sql.catalog.lance.storage.aws_access_key_id …
spark.sql.catalog.lance.storage.aws_secret_access_key …
```
Anything under `storage.` is handed to lance as object_store options, so those
are object_store's key names rather than Spark's `s3a` ones. Credentials belong
there because a gateway without STS vends none — see the note in
`../catalog_lancedb/README.md`.
## Running it
cd test/s3tables/catalog_spark_lance
(cd ../../../weed && go build .) # the harness runs this binary
go test -run TestSparkLanceNamespace -v -timeout 40m .
Skipped without Docker, and in `-short` mode. The first run downloads the Spark
image and the connector bundle, which is a few hundred megabytes; later runs
reuse both.
@@ -0,0 +1,127 @@
#!/usr/bin/env python3
"""Drive the SeaweedFS Lance Namespace with Spark.
Spark reaches the catalog through the Lance Spark connector's DSV2 catalog,
`org.lance.spark.LanceNamespaceSparkCatalog` with `impl=rest`, which speaks the
same routes this catalog implements. It is the Lance counterpart of the Spark
Iceberg suite next door.
Every step prints what happened. The last line is PASS, or a line starting with
FAIL that names the step, so the Go harness can report something useful rather
than a stack trace.
"""
import argparse
import sys
from pyspark.sql import SparkSession
def check(condition, message):
if not condition:
print(f"FAIL: {message}", file=sys.stderr)
raise SystemExit(1)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--namespace-url", required=True)
parser.add_argument("--s3-endpoint", required=True)
parser.add_argument("--bucket", required=True)
parser.add_argument("--namespace", default="ml")
parser.add_argument("--table", default="embeddings")
parser.add_argument("--access-key", default="any")
parser.add_argument("--secret-key", default="any")
parser.add_argument("--packages", required=True)
parser.add_argument("--ivy-dir", default="/tmp/ivy")
args = parser.parse_args()
# storage.* is handed to lance as its object_store options, so these are
# object_store's own key names rather than Spark's s3a ones.
spark = (
SparkSession.builder.appName("SeaweedFS Lance Namespace Test")
.config("spark.jars.ivy", args.ivy_dir)
.config("spark.jars.packages", args.packages)
.config("spark.sql.catalog.lance", "org.lance.spark.LanceNamespaceSparkCatalog")
.config("spark.sql.catalog.lance.impl", "rest")
.config("spark.sql.catalog.lance.uri", args.namespace_url)
.config("spark.sql.catalog.lance.storage.aws_endpoint", args.s3_endpoint)
.config("spark.sql.catalog.lance.storage.allow_http", "true")
.config("spark.sql.catalog.lance.storage.aws_access_key_id", args.access_key)
.config("spark.sql.catalog.lance.storage.aws_secret_access_key", args.secret_key)
.config("spark.sql.catalog.lance.storage.aws_region", "us-east-1")
.config("spark.sql.catalog.lance.storage.virtual_hosted_style_request", "false")
.getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")
print(f"spark {spark.version} connected to {args.namespace_url}")
ns = f"lance.`{args.bucket}`.{args.namespace}"
table = f"{ns}.{args.table}"
# 1. A table bucket is the first level of the namespace path, so the
# namespace below it is created through the catalog like any other.
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS {ns}")
namespaces = [row[0] for row in spark.sql(f"SHOW NAMESPACES IN lance.`{args.bucket}`").collect()]
print(f"SHOW NAMESPACES -> {namespaces}")
check(
any(args.namespace in name for name in namespaces),
f"{args.namespace} is not listed in {namespaces}",
)
# 2. Creating a table. The connector declares it through the namespace and
# writes the data itself, which is the split this catalog serves.
spark.sql(f"DROP TABLE IF EXISTS {table}")
spark.sql(
f"CREATE TABLE {table} (id BIGINT, title STRING, vector ARRAY<FLOAT>) USING lance"
)
# SHOW TABLES gives back the namespace's own identifiers - bucket, namespace
# and name joined by the delimiter - rather than a bare Spark table name.
tables = [row[1] for row in spark.sql(f"SHOW TABLES IN {ns}").collect()]
print(f"SHOW TABLES -> {tables}")
check(
any(args.table in name for name in tables),
f"{args.table} is not listed in {tables}",
)
# 3. Write, and read back through the catalog.
spark.sql(
f"""INSERT INTO {table} VALUES
(1, 'one', array(1.0f, 2.0f, 3.0f)),
(2, 'two', array(2.0f, 3.0f, 4.0f)),
(3, 'three', array(3.0f, 4.0f, 5.0f))"""
)
count = spark.sql(f"SELECT count(*) FROM {table}").collect()[0][0]
print(f"count -> {count}")
check(count == 3, f"read back {count} rows, want 3")
# 4. The schema survived the round trip, vector column included.
schema = {field.name for field in spark.table(table).schema.fields}
print(f"schema -> {sorted(schema)}")
check({"id", "title", "vector"} <= schema, f"schema lost columns: {schema}")
# 5. A filter, so it is not only a full scan.
rows = spark.sql(f"SELECT id, title FROM {table} WHERE id >= 2 ORDER BY id").collect()
print(f"filtered -> {[(r[0], r[1]) for r in rows]}")
check(len(rows) == 2, f"filter returned {len(rows)} rows, want 2")
check(rows[0][0] == 2, f"filter returned {rows[0][0]} first, want 2")
# 6. Appending again, because a second commit is the one that fails when a
# store cannot order commits - this store can, so it must not.
spark.sql(f"INSERT INTO {table} VALUES (4, 'four', array(4.0f, 5.0f, 6.0f))")
count = spark.sql(f"SELECT count(*) FROM {table}").collect()[0][0]
print(f"count after a second commit -> {count}")
check(count == 4, f"after appending, {count} rows, want 4")
# 7. And the dataset is readable straight off its location, which is what
# keeps the catalog optional.
location = f"s3://{args.bucket}/{args.namespace}/{args.table}"
print(f"dataset location -> {location}")
spark.stop()
print("PASS")
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,307 @@
// Package sparklance drives the SeaweedFS Lance Namespace with Spark, through
// the Lance Spark connector's DSV2 catalog. It is the Lance counterpart of the
// catalog_spark suite, which does the same for the Iceberg REST catalog.
//
// Spark is the engine most likely to be pointed at a lakehouse, and it reaches
// the catalog over the same routes as every other client - so what this proves
// is the protocol, not our idea of it.
package sparklance
import (
"context"
"fmt"
"math/rand"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/test/testutil"
)
const (
sparkImage = "apache/spark:3.5.1"
// Pinned: the connector is as much the thing under test as the server, and
// an unrelated release should not change what an old commit reproduces.
lancePackages = "org.lance:lance-spark-bundle-3.5_2.12:0.7.1"
startupTimeout = 60 * time.Second
clientTimeout = 25 * time.Minute
)
// TestSparkLanceNamespace runs Spark SQL against the namespace end to end:
// create a namespace and a table through the catalog, write, read back, filter,
// and append a second time.
func TestSparkLanceNamespace(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
if !hasDocker() {
t.Skip("Docker not available, skipping Spark Lance integration test")
}
env := newEnvironment(t)
defer env.cleanup()
env.start(t)
bucket := "sparklance-" + randomSuffix()
env.createTableBucket(t, bucket)
env.runSpark(t, bucket)
}
type environment struct {
weedBinary string
dataDir string
bindIP string
masterPort int
masterGrpcPort int
volumePort int
volumeGrpcPort int
filerPort int
filerGrpcPort int
s3Port int
s3GrpcPort int
lancePort int
accessKey string
secretKey string
weedCancel context.CancelFunc
weedCmd *exec.Cmd
}
func newEnvironment(t *testing.T) *environment {
t.Helper()
wd, err := os.Getwd()
if err != nil {
t.Fatalf("get working directory: %v", err)
}
seaweedDir := wd
for i := 0; i < 5; i++ {
if _, err := os.Stat(filepath.Join(seaweedDir, "go.mod")); err == nil {
break
}
seaweedDir = filepath.Dir(seaweedDir)
}
weedBinary := filepath.Join(seaweedDir, "weed", "weed")
if info, statErr := os.Stat(weedBinary); statErr == nil && !info.IsDir() {
// `make test` builds first; a plain `go test` will otherwise drive a
// binary from days ago and report a pass for code it never ran.
t.Logf("using %s, built %s", weedBinary, info.ModTime().Format(time.RFC3339))
} else {
weedBinary = "weed"
if _, err := exec.LookPath(weedBinary); err != nil {
t.Skip("weed binary not found, skipping integration test")
}
}
dataDir, err := os.MkdirTemp("", "seaweed-spark-lance-test-*")
if err != nil {
t.Fatalf("create temp dir: %v", err)
}
ports := testutil.MustAllocatePorts(t, 9)
return &environment{
weedBinary: weedBinary,
dataDir: dataDir,
bindIP: testutil.FindBindIP(),
masterPort: ports[0],
masterGrpcPort: ports[1],
volumePort: ports[2],
volumeGrpcPort: ports[3],
filerPort: ports[4],
filerGrpcPort: ports[5],
s3Port: ports[6],
s3GrpcPort: ports[7],
lancePort: ports[8],
accessKey: "AKIAIOSFODNN7EXAMPLE",
secretKey: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
}
}
func (env *environment) start(t *testing.T) {
t.Helper()
iamConfigPath, err := testutil.WriteIAMConfig(env.dataDir, env.accessKey, env.secretKey)
if err != nil {
t.Fatalf("write IAM config: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
env.weedCancel = cancel
cmd := exec.CommandContext(ctx, env.weedBinary, "mini",
"-master.port", fmt.Sprintf("%d", env.masterPort),
"-master.port.grpc", fmt.Sprintf("%d", env.masterGrpcPort),
"-volume.port", fmt.Sprintf("%d", env.volumePort),
"-volume.port.grpc", fmt.Sprintf("%d", env.volumeGrpcPort),
"-filer.port", fmt.Sprintf("%d", env.filerPort),
"-filer.port.grpc", fmt.Sprintf("%d", env.filerGrpcPort),
"-s3.port", fmt.Sprintf("%d", env.s3Port),
"-s3.port.grpc", fmt.Sprintf("%d", env.s3GrpcPort),
"-s3.port.lance", fmt.Sprintf("%d", env.lancePort),
"-s3.config", iamConfigPath,
"-ip", env.bindIP,
"-ip.bind", "0.0.0.0",
"-dir", env.dataDir,
)
cmd.Dir = env.dataDir
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
cmd.Env = append(os.Environ(),
"AWS_ACCESS_KEY_ID="+env.accessKey,
"AWS_SECRET_ACCESS_KEY="+env.secretKey,
)
if err := cmd.Start(); err != nil {
t.Fatalf("start SeaweedFS: %v", err)
}
env.weedCmd = cmd
// The namespace answers /v1/table once it is serving, which is a cheaper
// readiness check than waiting on a bucket that does not exist yet.
url := fmt.Sprintf("http://%s:%d/v1/table", env.bindIP, env.lancePort)
if !waitForHTTP(url, startupTimeout) {
t.Fatalf("the Lance namespace did not become ready at %s", url)
}
}
// waitForHTTP polls until the URL answers at all. An auth refusal counts: it
// means the server is up, which is the only thing being waited on.
func waitForHTTP(url string, timeout time.Duration) bool {
client := &http.Client{Timeout: 2 * time.Second}
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
resp, err := client.Get(url)
if err != nil {
time.Sleep(500 * time.Millisecond)
continue
}
status := resp.StatusCode
resp.Body.Close()
if status < 500 {
return true
}
time.Sleep(500 * time.Millisecond)
}
return false
}
func (env *environment) cleanup() {
if env.weedCancel != nil {
env.weedCancel()
}
if env.weedCmd != nil {
_ = env.weedCmd.Wait()
}
if env.dataDir != "" {
_ = os.RemoveAll(env.dataDir)
}
}
// createTableBucket makes the bucket LanceDB will read through, declared LANCE
// so the catalog refuses anything of another format in it.
func (env *environment) createTableBucket(t *testing.T, bucket string) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
cmd := exec.CommandContext(ctx, env.weedBinary, "shell",
fmt.Sprintf("-master=%s:%d.%d", env.bindIP, env.masterPort, env.masterGrpcPort),
)
cmd.Stdin = strings.NewReader(fmt.Sprintf(
"s3tables.bucket -create -name %s -format LANCE -account 000000000000\nexit\n", bucket))
if out, err := cmd.CombinedOutput(); err != nil {
t.Fatalf("create table bucket %s: %v\n%s", bucket, err, out)
}
t.Logf("created LANCE table bucket %s", bucket)
}
// runSpark runs the SQL driver inside the stock Spark image. The connector is
// pulled from Maven at submit time, the way the Iceberg Spark suite pulls its
// runtime, so nothing has to be built here.
func (env *environment) runSpark(t *testing.T, bucket string) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), clientTimeout)
defer cancel()
script, err := filepath.Abs("spark_lance_ops.py")
if err != nil {
t.Fatalf("resolve the driver script: %v", err)
}
// Ivy needs somewhere writable, and the Spark image runs as a user without a
// home directory it can write to. Kept outside the run's data directory so
// the connector bundle is downloaded once rather than on every run.
ivyDir := filepath.Join(os.TempDir(), "seaweedfs-lance-spark-ivy")
if err := os.MkdirAll(ivyDir, 0o777); err != nil {
t.Fatalf("create the ivy directory: %v", err)
}
// The container reaches the gateway through the host gateway address, which
// is not the address the namespace advertises to its own clients.
namespaceURL := fmt.Sprintf("http://host.docker.internal:%d", env.lancePort)
s3Endpoint := fmt.Sprintf("http://host.docker.internal:%d", env.s3Port)
cmd := exec.CommandContext(ctx, "docker", "run", "--rm",
"--add-host", "host.docker.internal:host-gateway",
"-v", script+":/opt/spark/work-dir/spark_lance_ops.py:ro",
"-v", ivyDir+":/tmp/ivy",
"-e", "HOME=/tmp",
"-e", "SPARK_LOCAL_IP=127.0.0.1",
"-u", "root",
sparkImage,
"/opt/spark/bin/spark-submit",
"--packages", lancePackages,
"--conf", "spark.jars.ivy=/tmp/ivy",
"/opt/spark/work-dir/spark_lance_ops.py",
"--namespace-url", namespaceURL,
"--s3-endpoint", s3Endpoint,
"--bucket", bucket,
"--access-key", env.accessKey,
"--secret-key", env.secretKey,
"--packages", lancePackages,
"--ivy-dir", "/tmp/ivy",
)
out, err := cmd.CombinedOutput()
t.Logf("Spark output:\n%s", tailLines(string(out), 60))
if err != nil {
t.Fatalf("the Spark driver failed: %v", err)
}
if !strings.Contains(string(out), "PASS") {
t.Fatalf("the Spark driver did not report PASS")
}
}
// tailLines keeps the end of Spark's very chatty output, which is where the
// driver's own lines and any failure are.
func tailLines(out string, n int) string {
lines := strings.Split(strings.TrimRight(out, "\n"), "\n")
if len(lines) <= n {
return out
}
return "... " + fmt.Sprintf("%d earlier lines omitted", len(lines)-n) + "\n" +
strings.Join(lines[len(lines)-n:], "\n")
}
func hasDocker() bool {
return exec.Command("docker", "version").Run() == nil
}
func randomSuffix() string {
const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
suffix := make([]byte, 8)
for i := range suffix {
suffix[i] = charset[rand.Intn(len(charset))]
}
return string(suffix)
}