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