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

.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

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"]

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.

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)
+}

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())