test: drive the Lance namespace with Spark (#10864)

* 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 * test: verify the Lance table bucket was actually created weed shell prints a command's own failure and still exits 0, so the harness would go on to blame Spark for a bucket that was never made. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm * test: bound the Docker probe An unhealthy daemon makes docker version hang, and the probe runs before the test has a timeout of its own. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm * test: keep the ivy cache under the user's cache directory It is mounted into a container running as root, so a shared temp path lets another local user pre-create it and choose what Spark loads. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm * test: assert the vector column's type, not only its name A column that came back as array<double> or array<string> would still be called vector and still pass. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm * test: read the dataset off its location for real The catalog being optional is the property that lets duckdb and pandas read these tables; it was asserted in a comment and printed, never exercised. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm * test: do not persist credentials in the Spark Lance checkout The job only uploads a log on failure; nothing in it pushes. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm * test: say the hosts in the README are placeholders The suite passes dynamically allocated host.docker.internal ports. Claude-Session: https://claude.ai/code/session_01Rkp1Mw5E89Jp6dzJFYiMrm
2026-09-08 15:41:15 +02:00 · 2026-08-21 23:49:09 -07:00
parent 4af6798639
commit 34bb444f33
4 changed files with 629 additions and 0 deletions
@@ -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.
@@ -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<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 - the type
+    #    as well as the name, since a column that came back as ARRAY<DOUBLE> or
+    #    ARRAY<STRING> 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<float>",
+    )
+    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())
@@ -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)
+}