diff --git a/.github/workflows/s3-tables-tests.yml b/.github/workflows/s3-tables-tests.yml index 42b1d78c8..ddbf01292 100644 --- a/.github/workflows/s3-tables-tests.yml +++ b/.github/workflows/s3-tables-tests.yml @@ -874,6 +874,76 @@ 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 + s3-tables-build-verification: name: S3 Tables Build Verification runs-on: ubuntu-22.04 diff --git a/test/s3tables/catalog_lancedb/Dockerfile.client b/test/s3tables/catalog_lancedb/Dockerfile.client new file mode 100644 index 000000000..8bc2e5f0c --- /dev/null +++ b/test/s3tables/catalog_lancedb/Dockerfile.client @@ -0,0 +1,15 @@ +# 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 + +RUN pip install --no-cache-dir lancedb lance-namespace pylance pyarrow + +COPY lancedb_ops.py /app/ + +CMD ["python3", "/app/lancedb_ops.py", "--help"] diff --git a/test/s3tables/catalog_lancedb/README.md b/test/s3tables/catalog_lancedb/README.md new file mode 100644 index 000000000..db067dc16 --- /dev/null +++ b/test/s3tables/catalog_lancedb/README.md @@ -0,0 +1,54 @@ +# 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 | +| `search(...)` | ANN search works on data behind SeaweedFS | +| `where("id < 5")` | so does the scan path, not only the index | +| `create_table` | reports what the client sees for an operation this catalog does not serve | +| 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. diff --git a/test/s3tables/catalog_lancedb/lancedb_catalog_test.go b/test/s3tables/catalog_lancedb/lancedb_catalog_test.go new file mode 100644 index 000000000..85a31861f --- /dev/null +++ b/test/s3tables/catalog_lancedb/lancedb_catalog_test.go @@ -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) +} diff --git a/test/s3tables/catalog_lancedb/lancedb_ops.py b/test/s3tables/catalog_lancedb/lancedb_ops.py new file mode 100644 index 000000000..dd47622c3 --- /dev/null +++ b/test/s3tables/catalog_lancedb/lancedb_ops.py @@ -0,0 +1,209 @@ +#!/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") + parser.add_argument("--rows", type=int, default=64) + 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. A vector search, which is what the format exists for. + query = [float(1) + d for d in range(DIM)] + hits = table.search(query).limit(3).to_list() + print(f"search -> {[hit['id'] for hit in hits]}") + check(len(hits) == 3, f"search returned {len(hits)} hits, want 3") + check(hits[0]["id"] == 1, f"nearest neighbour was id={hits[0]['id']}, want 1") + + # 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_ok = True + 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_ok = False + print(f"create_table with pushdown: refused with {type(err).__name__}: {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: client ok={pushed_ok}, catalog has it={landed}") + check( + pushed_ok == landed, + f"the client and the catalog disagree: client ok={pushed_ok}, listed={landed}", + ) + if landed: + 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") + + # 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())