diff --git a/.github/workflows/s3-tables-tests.yml b/.github/workflows/s3-tables-tests.yml index f146a3c29..3a2667a8a 100644 --- a/.github/workflows/s3-tables-tests.yml +++ b/.github/workflows/s3-tables-tests.yml @@ -1088,6 +1088,80 @@ jobs: path: test/s3tables/catalog_duckdb_lance/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 + with: + # The job uploads a test log on failure; nothing here needs to push, + # so do not leave a token in the checkout for it to pick up. + persist-credentials: false + + - 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 diff --git a/test/s3tables/catalog_spark_lance/README.md b/test/s3tables/catalog_spark_lance/README.md new file mode 100644 index 000000000..7ba408b80 --- /dev/null +++ b/test/s3tables/catalog_spark_lance/README.md @@ -0,0 +1,63 @@ +# 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. The host names below are placeholders — the +test passes dynamically allocated `host.docker.internal` ports, and a real +deployment uses whatever address the gateway answers on: + +``` +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. diff --git a/test/s3tables/catalog_spark_lance/spark_lance_ops.py b/test/s3tables/catalog_spark_lance/spark_lance_ops.py new file mode 100644 index 000000000..bb824e2ad --- /dev/null +++ b/test/s3tables/catalog_spark_lance/spark_lance_ops.py @@ -0,0 +1,153 @@ +#!/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 +from pyspark.sql.types import ArrayType, FloatType + + +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) 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 - the type + # as well as the name, since a column that came back as ARRAY or + # ARRAY would still be called "vector". + fields = {field.name: field.dataType for field in spark.table(table).schema.fields} + print(f"schema -> {[(n, t.simpleString()) for n, t in sorted(fields.items())]}") + check({"id", "title", "vector"} <= set(fields), f"schema lost columns: {sorted(fields)}") + vector = fields["vector"] + check( + isinstance(vector, ArrayType) and isinstance(vector.elementType, FloatType), + f"vector came back as {vector.simpleString()}, want array", + ) + check( + fields["id"].simpleString() == "bigint", + f"id came back as {fields['id'].simpleString()}, want bigint", + ) + + # 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, with no catalog + # in the path. This is the property that lets duckdb, pandas and + # DataFusion read these tables, so it is read rather than asserted by + # comment: the reader is given the storage options directly and never + # consults spark.sql.catalog.lance. + location = f"s3://{args.bucket}/{args.namespace}/{args.table}" + direct = ( + spark.read.format("lance") + .option("aws_endpoint", args.s3_endpoint) + .option("allow_http", "true") + .option("aws_access_key_id", args.access_key) + .option("aws_secret_access_key", args.secret_key) + .option("aws_region", "us-east-1") + .load(location) + ) + direct_count = direct.count() + print(f"direct read of {location} -> {direct_count} rows") + check(direct_count == 4, f"direct read got {direct_count} rows, want 4") + + spark.stop() + print("PASS") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test/s3tables/catalog_spark_lance/spark_lance_test.go b/test/s3tables/catalog_spark_lance/spark_lance_test.go new file mode 100644 index 000000000..d5d808096 --- /dev/null +++ b/test/s3tables/catalog_spark_lance/spark_lance_test.go @@ -0,0 +1,339 @@ +// 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)) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("create table bucket %s: %v\n%s", bucket, err, out) + } + // weed shell reports a command's own failure on stdout and still exits 0, so + // the exit code alone would let a missing bucket through and turn a setup + // failure into a confusing engine failure later. + if !env.tableBucketExists(t, bucket) { + t.Fatalf("table bucket %s was not created:\n%s", bucket, out) + } + t.Logf("created LANCE table bucket %s", bucket) +} + +// tableBucketExists asks the namespace, which lists table buckets at its root. +func (env *environment) tableBucketExists(t *testing.T, bucket string) bool { + t.Helper() + + url := fmt.Sprintf("http://%s:%d/v1/namespace/%s/exists", env.bindIP, env.lancePort, bucket) + resp, err := http.Post(url, "application/json", strings.NewReader("{}")) + if err != nil { + t.Fatalf("ask the namespace whether %s exists: %v", bucket, err) + } + defer resp.Body.Close() + return resp.StatusCode == http.StatusOK +} + +// 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, absErr := filepath.Abs("spark_lance_ops.py") + if absErr != nil { + t.Fatalf("resolve the driver script: %v", absErr) + } + + // 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 287MB connector bundle is downloaded once rather than on every run, + // and under the user's own cache rather than a shared temp path: this is + // mounted into a container running as root, so another local user must not + // be able to pre-create it and choose what Spark loads. + cacheRoot, err := os.UserCacheDir() + if err != nil { + cacheRoot = env.dataDir + } + ivyDir := filepath.Join(cacheRoot, "seaweedfs-lance-spark-ivy") + if err := os.MkdirAll(ivyDir, 0o755); 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") +} + +// hasDocker reports whether a Docker daemon answers. Bounded, because an +// unhealthy daemon makes `docker version` hang, and this runs before the test +// has a timeout of its own: better to skip than to eat the whole budget. +func hasDocker() bool { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + return exec.CommandContext(ctx, "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) +}