test: take a table through its whole life, for Iceberg and Lance (#10862)

* lance worker: share the integration tests' scaffolding

The recorder that keeps what a handler sent, the config builder and the
storage-option fallback all lived inside compaction.rs, so a second test
binary would have had to copy them. They move to tests/common.

The fallback now reads AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and
AWS_ENDPOINT_URL from the environment, defaulting to what it used before.
A harness can then point these tests at a gateway that checks what it is
given rather than one that accepts anything.

* lance worker: maintain one named table, for a harness to drive

Compacts and cleans up whatever WEED_LANCE_TABLE names, through the
handlers' own detect-then-execute path: a proposal the worker would not
have made is not one worth running.

The existing tests seed the tables they check. This one deliberately does
not, so a harness that has already written a table and knows what is in it
can have the real handlers maintain it and then read it back.

* test: take a table through its whole life, for Iceberg and Lance

Created in the catalog, filled by a real client, maintained by the worker,
read again, dropped. The step nothing was checking is the read after
maintenance: compaction once rewrote every dictionary-encoded column onto
a single value and shipped, because the maintenance tests were thorough
about sequence numbers, manifest entries and metadata versions and none of
them opened the parquet file the worker had just written.

So the assertion is a tally - row count, the cardinality of each
dictionary-encoded column, and an md5 over whole rows - taken before
maintenance and again after, required to be equal. The cardinalities name
the failure that happened; the digest catches a rewrite that keeps every
column's cardinality and hands the values to the wrong rows. A compaction
that merged nothing fails rather than passes, or the read afterwards is
checking a file the worker never wrote.

The Iceberg half runs two clients. DuckDB is the one the corruption was
reported against and the only one here that writes the deprecated
PLAIN_DICTIONARY encoding, which parquet-go normalizes away on write, so a
Go writer cannot produce it. PyIceberg writes the modern spelling. Pinning
parquet-go back to v0.30.1 fails the DuckDB half and passes the PyIceberg
one, which is why both are here.

Lance maintenance lives in the Rust worker, so it runs there where cargo
is installed and through the two lance calls those handlers wrap where it
is not. WEED_LANCE_MAINTENANCE picks one instead of letting the test guess.

* ci: run the table lifecycle tests

CI maintains the Lance table through the lance library rather than the
worker: a cold build of the lance crate costs more than the glue it would
be checking, and the worker's own tests cover its handlers.

The suite drives the Iceberg maintenance worker, so a change to it now
triggers this workflow too.

* test: let the lifecycle harness fail instead of skipping

Setup failures all exited zero, so a cluster that would not come up, or a
port allocation that lost, reported a green run for code nothing had
executed. That is the failure mode this whole directory exists to close,
and it was in the harness itself.

Only a checkout without a weed binary skips now, and it runs the tests so
each one says so rather than the package quietly passing. Everything else
fails.

The filer existence probe gets a deadline while I am here: it ran without
one, so an unresponsive filer would hang the suite past every timeout the
clients have.

* test: make the lifecycle checks check what they claim to

Three of them could pass without having looked.

The DuckDB skip matched "syntax error", "not implemented" and "Failed to
load" anywhere in the output, in any phase. A parse error in the SQL this
test generates, or a refusal from our own catalog, would have taken the
only coverage of the PLAIN_DICTIONARY encoding out of CI and left it
green. It now matches the extension failing to install, and only in the
phase that installs it. Everything past LOAD is ours and fails.

The digests covered id, category and value. Compaction rewrites the whole
row, so a defect confined to ts, or to a Lance vector, changed nothing
either side of maintenance. Every persisted column goes in now, ts as
microseconds so no timezone sits between the two runs.

The Lance drop check caught every exception as proof the dataset was
gone. pylance turns credential and transport failures into the same
ValueError, so it only accepts the message that means not found.

* docs: say up front which maintenance path the Lance half takes

The opening summary said the worker maintains both tables. It maintains
the Iceberg one always and the Lance one only where cargo is installed,
which is not what CI does.

test/s3tables/lifecycle/Dockerfile.lance

@@ -0,0 +1,17 @@
+# Lance client for the Lance half of the lifecycle test.
+#
+# Pinned to the versions this suite was verified against, the way
+# catalog_lancedb pins its client: an unrelated upstream release should not be
+# able to change what an old commit reproduces.
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN pip install --no-cache-dir \
+    "lance-namespace==0.8.6" \
+    "pylance==10.0.0" \
+    "pyarrow==25.0.1"
+
+COPY lance_lifecycle.py /app/
+
+CMD ["python3", "/app/lance_lifecycle.py", "--help"]

test/s3tables/lifecycle/Dockerfile.pyiceberg

@@ -0,0 +1,10 @@
+# PyIceberg client for the Iceberg half of the lifecycle test.
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN pip install --no-cache-dir "pyiceberg[s3fs]" pyarrow
+
+COPY iceberg_lifecycle.py /app/
+
+CMD ["python3", "/app/iceberg_lifecycle.py", "--help"]

test/s3tables/lifecycle/README.md

@@ -0,0 +1,77 @@
+# Table Lifecycle Integration Tests
+
+One table, all the way through: created in the catalog, filled by a real client,
+maintained, read again, dropped. Once for Iceberg and once for Lance. The
+Iceberg half always maintains through the worker; the Lance half maintains
+through the Rust worker or through the lance library, depending on what the
+environment has - see below.
+
+## Why this suite exists
+
+[#10853](https://github.com/seaweedfs/seaweedfs/issues/10853) was a compaction
+that rewrote every dictionary-encoded column onto a single value. It shipped.
+The maintenance tests we had were thorough about the bookkeeping - sequence
+numbers, added and deleted manifest entries, metadata versions, the manifest
+list - and every one of them passed, because not one of them opened the parquet
+file the worker had just written.
+
+So the assertion here is the dull one nothing else was making: tally the table
+before maintenance, tally it again after, and require the two to be equal. The
+tally is a row count, the cardinality of each dictionary-encoded column, and an
+md5 over whole rows. The cardinalities name the failure that happened; the
+digest catches a rewrite that keeps every column's cardinality and hands the
+values to the wrong rows.
+
+The same shape covers Lance, because the exposure is the same: a compaction that
+merges fragments can hand back a table that reads without complaint and answers
+wrongly.
+
+## What runs
+
+`TestIcebergTableLifecycle` starts a `weed mini` cluster, declares an `ICEBERG`
+table bucket, and runs two clients against it:
+
+| Client | Why both |
+| --- | --- |
+| DuckDB | the client the bug was reported against, and the only one here that writes the deprecated `PLAIN_DICTIONARY` encoding - parquet-go normalizes it away on write, so a Go writer cannot produce it |
+| PyIceberg | writes `RLE_DICTIONARY`, the modern spelling, so between the two the merge is checked against both dictionary encodings in the spec |
+
+Between the write and the read, the test runs the worker's whole maintenance
+cycle in-process against the live filer: compact, expire snapshots, remove
+orphans, rewrite manifests. A compaction that merged nothing fails the test
+rather than passing it - otherwise the read afterwards is checking a file the
+worker never wrote.
+
+`TestLanceTableLifecycle` does the same against a `LANCE` bucket: declare
+through the namespace, write a fragment per append, maintain, read, drop.
+Maintenance goes through the Rust worker's own handlers where cargo is
+installed, and through the two lance calls those handlers wrap where it is not.
+`WEED_LANCE_MAINTENANCE=library|worker` picks one instead of letting the test
+guess; CI sets `library`, because a cold build of the lance crate costs more
+than the layer it would be checking.
+
+Both tests finish by dropping the table and checking the data actually left the
+filer, which is the half of a lifecycle a catalog test never reaches.
+
+## Running it
+
+    cd test/s3tables/lifecycle
+    (cd ../../../weed && go build .)     # the harness runs this binary
+    go test -v -timeout 40m .
+
+Skipped without Docker and in `-short` mode. The first run builds the two client
+images and pulls `duckdb/duckdb:latest`; later runs reuse them. The DuckDB half
+skips itself, rather than failing, on an image whose iceberg extension cannot
+write through a REST catalog.
+
+To watch it catch the bug it was written for, pin parquet-go back to the version
+that had it:
+
+    go mod edit -require=github.com/parquet-go/parquet-go@v0.30.1 && go mod tidy
+    go test -run TestIcebergTableLifecycle/DuckDB -v .
+
+    maintenance collapsed the category column: 7 distinct values -> 1
+    maintenance collapsed the value column: 13 distinct values -> 1
+
+The PyIceberg half still passes there, which is the reason both clients are in
+this directory.

test/s3tables/lifecycle/cluster_test.go

@@ -0,0 +1,277 @@
+// Package lifecycle takes a table through everything that happens to it: made
+// through the catalog, filled by a real client, maintained by the worker, read
+// again, and dropped.
+//
+// The step that matters is the read after maintenance. A compaction once
+// rewrote every dictionary-encoded column onto a single value and went out in a
+// release, because the tests we had checked the bookkeeping - sequence numbers,
+// manifest entries, metadata versions - and none of them opened the file the
+// worker had just written. A tally taken before maintenance and the same tally
+// taken after is the whole idea, and it is the same idea for both formats:
+// Iceberg compaction merges parquet files, Lance compaction merges fragments,
+// and either can hand back a table that reads without complaint and answers
+// wrongly.
+package lifecycle
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"math/rand"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+const (
+	// The keys the issue's own recipe uses. weed mini turns them into the
+	// admin identity, which is what the clients then sign with.
+	accessKey = "AKIAIOSFODNN7EXAMPLE"
+	secretKey = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+
+	startupTimeout = 60 * time.Second
+	clientTimeout  = 20 * time.Minute
+)
+
+// shared is the one cluster both formats run against. They use separate table
+// buckets, and a bucket holds one format only.
+var shared *environment
+
+// errNoBinary is the one setup failure worth skipping over.
+var errNoBinary = errors.New("weed binary not found")
+
+func TestMain(m *testing.M) {
+	flag.Parse()
+	if testing.Short() {
+		os.Exit(m.Run())
+	}
+
+	// A checkout without a weed binary cannot run this and says so. Anything
+	// else - ports, a cluster that will not come up - is a failure, because a
+	// suite that turns its own breakage into a green run is the thing this
+	// directory exists to stop.
+	env, err := newEnvironment()
+	if errors.Is(err, errNoBinary) {
+		fmt.Fprintf(os.Stderr, "SKIP: %v\n", err)
+		os.Exit(m.Run())
+	}
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FAIL: %v\n", err)
+		os.Exit(1)
+	}
+	if err := env.start(); err != nil {
+		fmt.Fprintf(os.Stderr, "FAIL: weed mini did not start: %v\n", err)
+		env.cleanup()
+		os.Exit(1)
+	}
+	shared = env
+
+	code := m.Run()
+	shared.cleanup()
+	os.Exit(code)
+}
+
+type environment struct {
+	weedBinary string
+	rootDir    string
+	testDir    string
+	dataDir    string
+
+	masterPort     int
+	masterGrpcPort int
+	volumePort     int
+	volumeGrpcPort int
+	filerPort      int
+	filerGrpcPort  int
+	s3Port         int
+	s3GrpcPort     int
+	icebergPort    int
+	lancePort      int
+
+	weedCancel context.CancelFunc
+	weedCmd    *exec.Cmd
+}
+
+func newEnvironment() (*environment, error) {
+	wd, err := os.Getwd()
+	if err != nil {
+		return nil, fmt.Errorf("get working directory: %w", 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() {
+		// A plain `go test` will otherwise drive a binary from days ago and
+		// report a pass for code it never ran.
+		fmt.Fprintf(os.Stderr, "using %s, built %s\n", weedBinary, info.ModTime().Format(time.RFC3339))
+	} else {
+		weedBinary = "weed"
+		if _, err := exec.LookPath(weedBinary); err != nil {
+			return nil, errNoBinary
+		}
+	}
+
+	dataDir, err := os.MkdirTemp("", "seaweed-lifecycle-*")
+	if err != nil {
+		return nil, fmt.Errorf("create temp dir: %w", err)
+	}
+
+	ports, err := testutil.AllocatePorts(10)
+	if err != nil {
+		return nil, fmt.Errorf("allocate ports: %w", err)
+	}
+	return &environment{
+		weedBinary:     weedBinary,
+		rootDir:        seaweedDir,
+		testDir:        wd,
+		dataDir:        dataDir,
+		masterPort:     ports[0],
+		masterGrpcPort: ports[1],
+		volumePort:     ports[2],
+		volumeGrpcPort: ports[3],
+		filerPort:      ports[4],
+		filerGrpcPort:  ports[5],
+		s3Port:         ports[6],
+		s3GrpcPort:     ports[7],
+		icebergPort:    ports[8],
+		lancePort:      ports[9],
+	}, nil
+}
+
+func (env *environment) start() error {
+	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.iceberg", fmt.Sprintf("%d", env.icebergPort),
+		"-s3.port.lance", fmt.Sprintf("%d", env.lancePort),
+		"-ip.bind", "0.0.0.0",
+		"-dir", env.dataDir,
+	)
+	cmd.Dir = env.dataDir
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	// mini makes its admin identity from these, so they are the keys the
+	// clients sign with rather than a separate IAM file to keep in step.
+	cmd.Env = append(os.Environ(),
+		"AWS_ACCESS_KEY_ID="+accessKey,
+		"AWS_SECRET_ACCESS_KEY="+secretKey,
+	)
+
+	if err := cmd.Start(); err != nil {
+		cancel()
+		return err
+	}
+	env.weedCmd = cmd
+
+	if !testutil.WaitForService(env.catalogURL()+"/v1/config", startupTimeout) {
+		cancel()
+		return fmt.Errorf("the Iceberg catalog never answered on port %d", env.icebergPort)
+	}
+	if !testutil.WaitForPort(env.lancePort, startupTimeout) {
+		cancel()
+		return fmt.Errorf("the Lance namespace never answered on port %d", env.lancePort)
+	}
+	return nil
+}
+
+func (env *environment) cleanup() {
+	if env.weedCancel != nil {
+		env.weedCancel()
+	}
+	if env.weedCmd != nil {
+		_ = env.weedCmd.Wait()
+	}
+	if env.dataDir != "" {
+		_ = os.RemoveAll(env.dataDir)
+	}
+}
+
+func (env *environment) catalogURL() string {
+	return fmt.Sprintf("http://127.0.0.1:%d", env.icebergPort)
+}
+
+// A container reaches the same gateway by another name than the host does.
+func (env *environment) containerURL(port int) string {
+	return fmt.Sprintf("http://host.docker.internal:%d", port)
+}
+
+// createTableBucket makes a bucket declared for one format, so the catalog
+// refuses tables of any other kind in it.
+func (env *environment) createTableBucket(t *testing.T, bucket, format string) {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, env.weedBinary, "shell",
+		fmt.Sprintf("-master=127.0.0.1:%d.%d", env.masterPort, env.masterGrpcPort),
+	)
+	cmd.Stdin = strings.NewReader(fmt.Sprintf(
+		"s3tables.bucket -create -name %s -format %s -account 000000000000\nexit\n", bucket, format))
+	if out, err := cmd.CombinedOutput(); err != nil {
+		t.Fatalf("create the %s table bucket %s: %v\n%s", format, bucket, err, out)
+	}
+}
+
+// filerClient dials the filer the maintenance handlers talk to.
+func (env *environment) filerClient(t *testing.T) filer_pb.SeaweedFilerClient {
+	t.Helper()
+
+	conn, err := grpc.NewClient(fmt.Sprintf("127.0.0.1:%d", env.filerGrpcPort),
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("dial the filer: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return filer_pb.NewSeaweedFilerClient(conn)
+}
+
+// entryExists tells "the catalog forgot the table" from "the data is gone".
+func (env *environment) entryExists(t *testing.T, path string) bool {
+	t.Helper()
+
+	client := &http.Client{Timeout: 30 * time.Second}
+	resp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d%s", env.filerPort, path))
+	if err != nil {
+		t.Fatalf("filer GET %s: %v", path, err)
+	}
+	defer resp.Body.Close()
+	io.Copy(io.Discard, resp.Body)
+	return resp.StatusCode == http.StatusOK
+}
+
+func randomSuffix() string {
+	const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
+	suffix := make([]byte, 8)
+	for i := range suffix {
+		suffix[i] = charset[rand.Intn(len(charset))]
+	}
+	return string(suffix)
+}

test/s3tables/lifecycle/iceberg_lifecycle.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""PyIceberg half of the table lifecycle, run one phase per invocation.
+
+The Go test calls this three times - write, verify, drop - and runs the
+maintenance worker between the first two. Splitting it that way is the whole
+point: a tally taken before compaction and the same tally taken after are the
+only thing that catches a merge that rewrote every dictionary-encoded column
+onto a single value and said nothing.
+"""
+
+import argparse
+import hashlib
+import json
+import sys
+import time
+
+import pyarrow as pa
+from pyiceberg.catalog import load_catalog
+from pyiceberg.exceptions import NamespaceAlreadyExistsError, NoSuchTableError
+from pyiceberg.schema import Schema
+from pyiceberg.types import LongType, NestedField, StringType, TimestamptzType
+
+# Few enough distinct values in the two string columns that any writer worth
+# the name dictionary-encodes them, which is the encoding that broke.
+CATEGORIES = 7
+VALUES = 13
+ROWS_PER_BATCH = 4000
+BATCHES = 3
+
+SCHEMA = Schema(
+    NestedField(1, "id", LongType(), required=True),
+    NestedField(2, "category", StringType(), required=True),
+    NestedField(3, "value", StringType(), required=True),
+    NestedField(4, "ts", TimestamptzType(), required=True),
+)
+
+ARROW_SCHEMA = pa.schema(
+    [
+        pa.field("id", pa.int64(), nullable=False),
+        pa.field("category", pa.string(), nullable=False),
+        pa.field("value", pa.string(), nullable=False),
+        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
+    ]
+)
+
+
+def batch(start, count):
+    ids = list(range(start, start + count))
+    return pa.Table.from_pydict(
+        {
+            "id": ids,
+            "category": [f"cat-{i % CATEGORIES}" for i in ids],
+            "value": [f"v-{i % VALUES}" for i in ids],
+            # An hour apart, so the rows spread over months the way a real
+            # table's do without needing a partition spec to prove it.
+            "ts": [1772323200000000 + i * 3600000000 for i in ids],
+        },
+        schema=ARROW_SCHEMA,
+    )
+
+
+def tally(table):
+    """Row count, per-column cardinality, and a digest of every row.
+
+    The cardinalities catch a column collapsed onto one dictionary entry; the
+    digest catches everything else, including a merge that keeps the right
+    number of distinct values while handing them to the wrong rows. Every
+    column goes into it, not just the two the cardinalities watch - compaction
+    rewrites the whole row. ts goes in as microseconds so no timezone sits
+    between the two runs.
+    """
+    scanned = table.scan().to_arrow()
+    categories = scanned.column("category").to_pylist()
+    values = scanned.column("value").to_pylist()
+    rows = [
+        f"{i}|{t}|{c}|{v}"
+        for i, t, c, v in zip(
+            scanned.column("id").to_pylist(),
+            scanned.column("ts").cast(pa.int64()).to_pylist(),
+            categories,
+            values,
+            strict=True,
+        )
+    ]
+    digest = hashlib.md5(
+        "\n".join(sorted(rows)).encode(), usedforsecurity=False
+    ).hexdigest()
+    return {
+        "rows": scanned.num_rows,
+        "categories": len(set(categories)),
+        "values": len(set(values)),
+        "digest": digest,
+    }
+
+
+def connect(args):
+    properties = {
+        "type": "rest",
+        "uri": args.catalog_url,
+        "warehouse": f"s3://{args.bucket}/",
+        "prefix": args.bucket,
+        "s3.endpoint": args.s3_endpoint,
+        "s3.access-key-id": args.access_key,
+        "s3.secret-access-key": args.secret_key,
+        "s3.region": "us-east-1",
+        "s3.path-style-access": "true",
+    }
+    last = None
+    for attempt in range(10):
+        try:
+            return load_catalog("rest", **properties)
+        except Exception as err:  # the gateway may still be coming up
+            last = err
+            print(f"connect attempt {attempt + 1} failed: {err}", file=sys.stderr)
+            time.sleep(2)
+    raise last
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--phase", required=True, choices=["write", "verify", "drop"])
+    parser.add_argument("--catalog-url", required=True)
+    parser.add_argument("--s3-endpoint", required=True)
+    parser.add_argument("--bucket", required=True)
+    parser.add_argument("--namespace", required=True)
+    parser.add_argument("--table", required=True)
+    parser.add_argument("--access-key", required=True)
+    parser.add_argument("--secret-key", required=True)
+    args = parser.parse_args()
+
+    catalog = connect(args)
+    identifier = f"{args.namespace}.{args.table}"
+
+    if args.phase == "write":
+        try:
+            catalog.create_namespace(args.namespace)
+        except NamespaceAlreadyExistsError:
+            pass
+        table = catalog.create_table(identifier, schema=SCHEMA)
+        # One append per batch, so compaction has several files to merge
+        # rather than one it would leave alone.
+        for i in range(BATCHES):
+            table.append(batch(i * ROWS_PER_BATCH + 1, ROWS_PER_BATCH))
+        table = catalog.load_table(identifier)
+        print(json.dumps(tally(table)))
+        return
+
+    if args.phase == "verify":
+        print(json.dumps(tally(catalog.load_table(identifier))))
+        return
+
+    catalog.drop_table(identifier)
+    try:
+        catalog.load_table(identifier)
+    except NoSuchTableError:
+        pass
+    else:
+        raise SystemExit("the table is still in the catalog after a drop")
+    catalog.drop_namespace(args.namespace)
+
+
+if __name__ == "__main__":
+    main()

test/s3tables/lifecycle/iceberg_lifecycle_test.go

@@ -0,0 +1,389 @@
+package lifecycle
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/iceberg"
+)
+
+const (
+	pyicebergImage = "seaweedfs-lifecycle-pyiceberg"
+	duckdbImage    = "duckdb/duckdb:latest"
+
+	// The phase that installs the extension, and so the only one whose failure
+	// can mean the image rather than the code.
+	firstPhase = "write"
+
+	// Three appends of this many rows, all inside one month so they land in
+	// one partition and compaction has something to merge. The two string
+	// columns hold few enough distinct values to be dictionary-encoded, which
+	// is the encoding the merge destroyed.
+	rowsPerBatch = 4000
+	batches      = 3
+	categories   = 7
+	values       = 13
+)
+
+// tally is what a client reports about a table. Both halves of this suite
+// speak it, and the point of the whole suite is that the one taken before
+// maintenance equals the one taken after.
+type tally struct {
+	Rows       int    `json:"rows"`
+	Categories int    `json:"categories"`
+	Values     int    `json:"values"`
+	Digest     string `json:"digest"`
+	// Lance only: a compaction that merged nothing would otherwise let this
+	// test pass without having tested anything.
+	Fragments int `json:"fragments"`
+	Version   int `json:"version"`
+}
+
+func TestIcebergTableLifecycle(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	if shared == nil {
+		t.Skip("no cluster")
+	}
+	if !testutil.HasDocker() {
+		t.Skip("Docker not available")
+	}
+
+	// DuckDB is the client the bug was reported against, and the only one
+	// here that writes the deprecated PLAIN_DICTIONARY encoding a Go writer
+	// will not produce. PyIceberg writes the modern one, so between them the
+	// merge is checked against both dictionary encodings in the spec.
+	t.Run("DuckDB", testIcebergLifecycleWithDuckDB)
+	t.Run("PyIceberg", testIcebergLifecycleWithPyIceberg)
+}
+
+func testIcebergLifecycleWithDuckDB(t *testing.T) {
+	env := shared
+	bucket := "lifecycle-duckdb-" + randomSuffix()
+	namespace, table := "sales", "events"
+	env.createTableBucket(t, bucket, "ICEBERG")
+
+	inserts := make([]string, 0, batches)
+	for i := 0; i < batches; i++ {
+		inserts = append(inserts, fmt.Sprintf(
+			"INSERT INTO cat.%s.%s SELECT g, TIMESTAMPTZ '2026-03-01 00:00:00' + INTERVAL (g %% 600) MINUTE, "+
+				"'cat-' || (g %% %d), 'v-' || (g %% %d) FROM generate_series(%d, %d) t(g);",
+			namespace, table, categories, values, i*rowsPerBatch+1, (i+1)*rowsPerBatch))
+	}
+
+	before := env.duckdb(t, bucket, "write", strings.Join([]string{
+		fmt.Sprintf("CREATE SCHEMA cat.%s;", namespace),
+		// Partitioned, the way the table in the report was. Compaction bins
+		// per partition, so this is also what puts more than one bin in play.
+		fmt.Sprintf("CREATE TABLE cat.%s.%s(id int, ts timestamptz, category text, value text) PARTITIONED BY (month(ts));", namespace, table),
+		strings.Join(inserts, "\n"),
+		duckdbTallySQL(namespace, table),
+	}, "\n"))
+	assertSeeded(t, before)
+
+	env.maintainIcebergTable(t, bucket, namespace+"/"+table)
+
+	after := env.duckdb(t, bucket, "verify", duckdbTallySQL(namespace, table))
+	assertSameData(t, before, after)
+
+	env.duckdb(t, bucket, "drop", fmt.Sprintf("DROP TABLE cat.%s.%s;\nDROP SCHEMA cat.%s;", namespace, table, namespace))
+	if env.entryExists(t, fmt.Sprintf("/buckets/%s/%s/%s/", bucket, namespace, table)) {
+		t.Fatal("the dropped table's data is still on disk")
+	}
+}
+
+func testIcebergLifecycleWithPyIceberg(t *testing.T) {
+	env := shared
+	bucket := "lifecycle-pyiceberg-" + randomSuffix()
+	namespace, table := "sales", "events"
+	env.createTableBucket(t, bucket, "ICEBERG")
+	buildClientImage(t, pyicebergImage, "Dockerfile.pyiceberg")
+
+	before := env.pyiceberg(t, "write", bucket, namespace, table)
+	assertSeeded(t, before)
+
+	env.maintainIcebergTable(t, bucket, namespace+"/"+table)
+
+	after := env.pyiceberg(t, "verify", bucket, namespace, table)
+	assertSameData(t, before, after)
+
+	env.pyiceberg(t, "drop", bucket, namespace, table)
+	if env.entryExists(t, fmt.Sprintf("/buckets/%s/%s/%s/", bucket, namespace, table)) {
+		t.Fatal("the dropped table's data is still on disk")
+	}
+}
+
+// maintainIcebergTable runs the worker's whole maintenance cycle against the
+// live filer, in the order a scheduled worker would: merge the small files,
+// drop the snapshots that referenced them, sweep what nothing references any
+// more, then fold the manifests together.
+func (env *environment) maintainIcebergTable(t *testing.T, bucket, tablePath string) {
+	t.Helper()
+
+	client := env.filerClient(t)
+	handler := iceberg.NewHandler(nil)
+	config := iceberg.Config{
+		TargetFileSizeBytes:  256 << 20,
+		MinInputFiles:        2,
+		MaxCommitRetries:     3,
+		SnapshotRetentionMs:  1,
+		MaxSnapshotsToKeep:   1,
+		OrphanOlderThanHours: 1,
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	operations := []struct {
+		name string
+		run  func() (string, map[string]int64, error)
+	}{
+		{"compact", func() (string, map[string]int64, error) {
+			return handler.CompactDataFiles(ctx, client, bucket, tablePath, config)
+		}},
+		{"expire", func() (string, map[string]int64, error) {
+			return handler.ExpireSnapshots(ctx, client, bucket, tablePath, config)
+		}},
+		{"orphans", func() (string, map[string]int64, error) {
+			return handler.RemoveOrphans(ctx, client, bucket, tablePath, config)
+		}},
+		{"manifests", func() (string, map[string]int64, error) {
+			return handler.RewriteManifests(ctx, client, bucket, tablePath, config)
+		}},
+	}
+	for _, operation := range operations {
+		result, metrics, err := operation.run()
+		if err != nil {
+			t.Fatalf("%s: %v", operation.name, err)
+		}
+		t.Logf("%s: %s %v", operation.name, result, metrics)
+
+		// A compaction that merged nothing leaves the read below checking a
+		// file the worker never wrote, which proves nothing at all.
+		if operation.name == "compact" && metrics[iceberg.MetricFilesMerged] < batches {
+			t.Fatalf("compaction merged %d files, want all %d written by the client: %s",
+				metrics[iceberg.MetricFilesMerged], batches, result)
+		}
+	}
+}
+
+// duckdb runs a script against the catalog and returns whatever tally it
+// printed. The prelude is the reporter's own ATTACH, SigV4 and all.
+func (env *environment) duckdb(t *testing.T, bucket, phase, body string) tally {
+	t.Helper()
+
+	script := fmt.Sprintf(`INSTALL iceberg;
+LOAD iceberg;
+CREATE SECRET s3_secret (TYPE S3, KEY_ID '%s', SECRET '%s', ENDPOINT 'host.docker.internal:%d', URL_STYLE 'path', USE_SSL false);
+ATTACH 's3://%s' AS cat (TYPE ICEBERG, ENDPOINT '%s', AUTHORIZATION_TYPE SigV4, SECRET 's3_secret', SIGV4_SERVICE 's3', SIGV4_REGION 'us-east-1', ACCESS_DELEGATION_MODE 'none', READ_ONLY false);
+%s
+`, accessKey, secretKey, env.s3Port, bucket, env.containerURL(env.icebergPort), body)
+
+	name := fmt.Sprintf("duckdb-%s-%s.sql", bucket, phase)
+	if err := os.WriteFile(filepath.Join(env.dataDir, name), []byte(script), 0644); err != nil {
+		t.Fatalf("write the DuckDB script: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), clientTimeout)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, "docker", "run", "--rm",
+		"-v", fmt.Sprintf("%s:/test", env.dataDir),
+		"--add-host", "host.docker.internal:host-gateway",
+		"--entrypoint", "duckdb",
+		duckdbImage,
+		"-init", "/test/"+name,
+		"-c", "SELECT 1",
+	)
+	out, err := cmd.CombinedOutput()
+	t.Logf("DuckDB %s:\n%s", phase, out)
+	if err != nil {
+		if phase == firstPhase && lacksIcebergExtension(string(out)) {
+			t.Skipf("this DuckDB image has no iceberg extension: %v", err)
+		}
+		t.Fatalf("DuckDB %s: %v", phase, err)
+	}
+	if !strings.Contains(body, "TALLY") {
+		return tally{}
+	}
+	return parseDuckDBTally(t, string(out))
+}
+
+func duckdbTallySQL(namespace, table string) string {
+	// The digest covers whole rows - every column, not just the two the
+	// cardinalities watch - because counting each column on its own passes a
+	// merge that keeps every column's cardinality and hands the values to the
+	// wrong rows. ts goes in as microseconds so the session's timezone cannot
+	// change how it renders between the two runs.
+	return fmt.Sprintf("SELECT 'TALLY ' || count(*) || ' ' || count(DISTINCT category) || ' ' || count(DISTINCT value)"+
+		" || ' ' || md5(string_agg(id || '|' || epoch_us(ts) || '|' || category || '|' || value, chr(10) ORDER BY id)) AS marker"+
+		" FROM cat.%s.%s;",
+		namespace, table)
+}
+
+var duckdbTallyPattern = regexp.MustCompile(`TALLY (\d+) (\d+) (\d+) ([0-9a-f]{32})`)
+
+func parseDuckDBTally(t *testing.T, out string) tally {
+	t.Helper()
+
+	fields := duckdbTallyPattern.FindStringSubmatch(out)
+	if fields == nil {
+		t.Fatalf("DuckDB printed no tally:\n%s", out)
+	}
+	number := func(s string) int {
+		value, err := strconv.Atoi(s)
+		if err != nil {
+			t.Fatalf("parse %q: %v", s, err)
+		}
+		return value
+	}
+	return tally{
+		Rows:       number(fields[1]),
+		Categories: number(fields[2]),
+		Values:     number(fields[3]),
+		Digest:     fields[4],
+	}
+}
+
+// lacksIcebergExtension reports whether DuckDB never got as far as this
+// repository's code, which is the only failure worth skipping over.
+//
+// The extension ships separately from the image, so an environment that cannot
+// fetch it has nothing to say about compaction. Everything past LOAD - a parse
+// error on the ATTACH, a refusal from the catalog, a bad read - is ours, and
+// has to fail: this is the one test covering the PLAIN_DICTIONARY encoding, and
+// a skip nobody reads is how the corruption it was written for shipped.
+func lacksIcebergExtension(out string) bool {
+	for _, marker := range []string{
+		"iceberg extension is not available",
+		"Failed to download extension",
+		`Extension "iceberg" not found`,
+		"Unknown extension",
+	} {
+		if strings.Contains(out, marker) {
+			return true
+		}
+	}
+	return false
+}
+
+func (env *environment) pyiceberg(t *testing.T, phase, bucket, namespace, table string) tally {
+	t.Helper()
+
+	out := env.runClient(t, pyicebergImage, phase, "/app/iceberg_lifecycle.py",
+		"--catalog-url", env.containerURL(env.icebergPort),
+		"--s3-endpoint", env.containerURL(env.s3Port),
+		"--bucket", bucket,
+		"--namespace", namespace,
+		"--table", table,
+	)
+	if phase == "drop" {
+		return tally{}
+	}
+	return decodeTally(t, out)
+}
+
+// runClient runs one phase of a python client and returns its stdout.
+func (env *environment) runClient(t *testing.T, image, phase, script string, args ...string) string {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), clientTimeout)
+	defer cancel()
+
+	run := []string{"run", "--rm",
+		"--add-host", "host.docker.internal:host-gateway",
+		"-e", "AWS_ACCESS_KEY_ID=" + accessKey,
+		"-e", "AWS_SECRET_ACCESS_KEY=" + secretKey,
+		"-e", "AWS_REGION=us-east-1",
+		"-e", "AWS_ALLOW_HTTP=true",
+		image, "python3", script, "--phase", phase,
+		"--access-key", accessKey, "--secret-key", secretKey,
+	}
+	cmd := exec.CommandContext(ctx, "docker", append(run, args...)...)
+	var stderr strings.Builder
+	cmd.Stderr = &stderr
+	out, err := cmd.Output()
+	if stderr.Len() > 0 {
+		t.Logf("%s %s (stderr):\n%s", image, phase, stderr.String())
+	}
+	if err != nil {
+		t.Fatalf("%s %s: %v\n%s", image, phase, err, out)
+	}
+	t.Logf("%s %s: %s", image, phase, strings.TrimSpace(string(out)))
+	return string(out)
+}
+
+func decodeTally(t *testing.T, out string) tally {
+	t.Helper()
+
+	// The clients print the tally last; anything a library logged before it
+	// is not JSON and not ours.
+	lines := strings.Split(strings.TrimSpace(out), "\n")
+	var decoded tally
+	if err := json.Unmarshal([]byte(lines[len(lines)-1]), &decoded); err != nil {
+		t.Fatalf("decode the tally from %q: %v", out, err)
+	}
+	return decoded
+}
+
+func buildClientImage(t *testing.T, image, dockerfile string) {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, "docker", "build", "-t", image, "-f", dockerfile, ".")
+	cmd.Dir = shared.testDir
+	if out, err := cmd.CombinedOutput(); err != nil {
+		t.Fatalf("build %s: %v\n%s", image, err, out)
+	}
+}
+
+// assertSeeded checks the client wrote what this suite assumes it wrote. Every
+// comparison below is against these numbers, so a client that quietly seeded
+// one row would otherwise make the whole test vacuous.
+func assertSeeded(t *testing.T, seeded tally) {
+	t.Helper()
+
+	if want := rowsPerBatch * batches; seeded.Rows != want {
+		t.Fatalf("seeded %d rows, want %d", seeded.Rows, want)
+	}
+	if seeded.Categories != categories || seeded.Values != values {
+		t.Fatalf("seeded %d categories and %d values, want %d and %d",
+			seeded.Categories, seeded.Values, categories, values)
+	}
+}
+
+// assertSameData is the test. Maintenance rewrites files; it must not change a
+// single row, and the cardinalities are called out separately because that is
+// the shape the failure took: a dictionary column collapsed onto one entry,
+// read back without an error, and answered wrongly.
+func assertSameData(t *testing.T, before, after tally) {
+	t.Helper()
+
+	if after.Rows != before.Rows {
+		t.Errorf("maintenance changed the row count: %d -> %d", before.Rows, after.Rows)
+	}
+	if after.Categories != before.Categories {
+		t.Errorf("maintenance collapsed the category column: %d distinct values -> %d",
+			before.Categories, after.Categories)
+	}
+	if after.Values != before.Values {
+		t.Errorf("maintenance collapsed the value column: %d distinct values -> %d",
+			before.Values, after.Values)
+	}
+	if after.Digest != before.Digest {
+		t.Errorf("maintenance changed the rows: digest %s -> %s", before.Digest, after.Digest)
+	}
+}

test/s3tables/lifecycle/lance_lifecycle.py

@@ -0,0 +1,168 @@
+#!/usr/bin/env python3
+"""Lance half of the table lifecycle, run one phase per invocation.
+
+The Go test calls this for each step and maintains the table in between, so the
+tally taken before maintenance and the tally taken after are directly
+comparable. That comparison is the test: the Iceberg side of this suite exists
+because compaction once rewrote a table's dictionary columns onto a single
+value and every check we had still passed, and a Lance dataset is rewritten by
+the same kind of job.
+
+The maintain phase is a fallback. When the Rust worker can be built, the Go test
+runs its handlers against this table instead and skips this phase - what runs
+here are the same two lance calls the handlers make.
+"""
+
+import argparse
+import hashlib
+import json
+import sys
+import warnings
+from datetime import timedelta
+
+warnings.filterwarnings("ignore")
+
+import lance
+import lance_namespace as ln
+import pyarrow as pa
+
+# Low enough cardinality that these two columns are dictionary-encoded.
+CATEGORIES = 7
+VALUES = 13
+ROWS_PER_BATCH = 4000
+BATCHES = 3
+DIM = 8
+
+
+def rows(start, count):
+    ids = list(range(start, start + count))
+    return pa.table(
+        {
+            "id": pa.array(ids, type=pa.int64()),
+            "category": pa.array([f"cat-{i % CATEGORIES}" for i in ids]),
+            "value": pa.array([f"v-{i % VALUES}" for i in ids]),
+            "vector": pa.array(
+                [[float(i % 97) + d for d in range(DIM)] for i in ids],
+                type=pa.list_(pa.float32(), DIM),
+            ),
+        }
+    )
+
+
+def tally(dataset):
+    """What the table holds, in a form two runs can be compared by.
+
+    The cardinalities catch a column collapsed onto one value; the digest
+    catches a rewrite that keeps the values and moves them to the wrong rows.
+    Every column goes into the digest, the vectors included - compaction
+    rewrites whole fragments, so leaving a column out leaves a place for it to
+    go wrong unnoticed. Fragments come along because a compaction that merged
+    nothing would otherwise let this test pass without having tested anything.
+    """
+    scanned = dataset.to_table(columns=["id", "category", "value", "vector"])
+    ids = scanned.column("id").to_pylist()
+    categories = scanned.column("category").to_pylist()
+    values = scanned.column("value").to_pylist()
+    vectors = scanned.column("vector").to_pylist()
+    serialized = (
+        f"{i}|{c}|{v}|{w}"
+        for i, c, v, w in zip(ids, categories, values, vectors, strict=True)
+    )
+    digest = hashlib.md5(
+        "\n".join(sorted(serialized)).encode(), usedforsecurity=False
+    ).hexdigest()
+    return {
+        "rows": scanned.num_rows,
+        "categories": len(set(categories)),
+        "values": len(set(values)),
+        "digest": digest,
+        "fragments": len(dataset.get_fragments()),
+        "version": dataset.version,
+    }
+
+
+def resolve(args):
+    """Ask the namespace where the table lives, the way the worker does."""
+    namespace = ln.connect("rest", {"uri": args.namespace_url})
+    table_id = [args.bucket, args.namespace, args.table]
+    described = namespace.describe_table(ln.DescribeTableRequest(id=table_id))
+    return namespace, table_id, described.location, storage_options(args, described)
+
+
+def storage_options(args, described=None):
+    # The namespace vends an endpoint correct for its own host; this container
+    # reaches the same gateway by another name. Credentials are filled in
+    # because a deployment without STS vends none.
+    options = dict((described.storage_options or {}) if described else {})
+    options["aws_endpoint"] = args.s3_endpoint
+    options["allow_http"] = "true"
+    options.setdefault("aws_access_key_id", args.access_key)
+    options.setdefault("aws_secret_access_key", args.secret_key)
+    options.setdefault("aws_region", "us-east-1")
+    return options
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--phase", required=True, choices=["write", "maintain", "verify", "drop"])
+    parser.add_argument("--namespace-url", required=True)
+    parser.add_argument("--s3-endpoint", required=True)
+    parser.add_argument("--bucket", required=True)
+    parser.add_argument("--namespace", required=True)
+    parser.add_argument("--table", required=True)
+    parser.add_argument("--access-key", default="any")
+    parser.add_argument("--secret-key", default="any")
+    args = parser.parse_args()
+
+    if args.phase == "write":
+        namespace = ln.connect("rest", {"uri": args.namespace_url})
+        for parent in ([args.bucket], [args.bucket, args.namespace]):
+            namespace.create_namespace(ln.CreateNamespaceRequest(id=parent, mode="EXIST_OK"))
+        table_id = [args.bucket, args.namespace, args.table]
+        declared = namespace.declare_table(ln.DeclareTableRequest(id=table_id))
+        options = storage_options(args)
+        # A fragment per append, so the compaction that follows has something
+        # to merge.
+        for i in range(BATCHES):
+            lance.write_dataset(
+                rows(i * ROWS_PER_BATCH + 1, ROWS_PER_BATCH),
+                declared.location,
+                storage_options=options,
+                mode="overwrite" if i == 0 else "append",
+            )
+        print(json.dumps(tally(lance.dataset(declared.location, storage_options=options))))
+        return 0
+
+    if args.phase == "maintain":
+        _, _, location, options = resolve(args)
+        dataset = lance.dataset(location, storage_options=options)
+        dataset.optimize.compact_files()
+        dataset = lance.dataset(location, storage_options=options)
+        dataset.cleanup_old_versions(older_than=timedelta(seconds=0), delete_unverified=True)
+        print(json.dumps(tally(lance.dataset(location, storage_options=options))))
+        return 0
+
+    if args.phase == "verify":
+        _, _, location, options = resolve(args)
+        print(json.dumps(tally(lance.dataset(location, storage_options=options))))
+        return 0
+
+    namespace, table_id, location, options = resolve(args)
+    namespace.drop_table(ln.DropTableRequest(id=table_id))
+    try:
+        lance.dataset(location, storage_options=options)
+    except ValueError as err:
+        # pylance turns every load failure into a ValueError, so the message is
+        # the only thing separating a dataset that is gone from credentials
+        # that stopped working halfway through the test.
+        if "was not found" in str(err):
+            return 0
+        print(f"FAIL: reading the dropped dataset failed for another reason: {err}",
+              file=sys.stderr)
+        return 1
+    print("FAIL: the dataset is still readable after a drop", file=sys.stderr)
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

test/s3tables/lifecycle/lance_lifecycle_test.go

@@ -0,0 +1,130 @@
+package lifecycle
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+)
+
+const lanceImage = "seaweedfs-lifecycle-lance"
+
+// TestLanceTableLifecycle is the Iceberg test's counterpart. A Lance table is
+// declared through the namespace, written a fragment at a time, compacted, its
+// superseded versions dropped, and read again - and the read has to answer
+// exactly what the write put there. Compaction rewrites fragments the way
+// Iceberg compaction rewrites parquet files, and the failure that suite exists
+// for is the kind that reads back without an error and answers wrongly.
+func TestLanceTableLifecycle(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	if shared == nil {
+		t.Skip("no cluster")
+	}
+	if !testutil.HasDocker() {
+		t.Skip("Docker not available")
+	}
+
+	env := shared
+	bucket := "lifecycle-lance-" + randomSuffix()
+	namespace, table := "ml", "events"
+	env.createTableBucket(t, bucket, "LANCE")
+	buildClientImage(t, lanceImage, "Dockerfile.lance")
+
+	before := env.lance(t, "write", bucket, namespace, table)
+	assertSeeded(t, before)
+	if before.Fragments != batches {
+		t.Fatalf("the client wrote %d fragments, want one per append (%d)", before.Fragments, batches)
+	}
+
+	env.maintainLanceTable(t, bucket, namespace, table)
+
+	after := env.lance(t, "verify", bucket, namespace, table)
+	assertSameData(t, before, after)
+	// Without this the comparison above would pass on a table nothing touched.
+	if after.Fragments >= before.Fragments {
+		t.Fatalf("maintenance merged nothing: %d fragments before, %d after",
+			before.Fragments, after.Fragments)
+	}
+
+	env.lance(t, "drop", bucket, namespace, table)
+	if env.entryExists(t, fmt.Sprintf("/buckets/%s/%s/%s/", bucket, namespace, table)) {
+		t.Fatal("the dropped table's data is still on disk")
+	}
+}
+
+// maintainLanceTable compacts the table and drops what compaction superseded.
+//
+// Lance maintenance lives in the Rust worker, so where its toolchain is around
+// the handlers themselves do the work against this table. Where it is not, the
+// same two lance calls those handlers wrap run in the client container instead.
+// The lifecycle is checked either way; only the layer above the format changes.
+// WEED_LANCE_MAINTENANCE picks one - CI sets "library", because a cold build of
+// the lance crate costs more than the layer it is checking.
+func (env *environment) maintainLanceTable(t *testing.T, bucket, namespace, table string) {
+	t.Helper()
+
+	if !maintainWithWorker(t) {
+		env.lance(t, "maintain", bucket, namespace, table)
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), clientTimeout)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, "cargo", "test",
+		"-p", "weed-lance-worker", "--test", "lifecycle", "--", "--nocapture")
+	cmd.Dir = filepath.Join(env.rootDir, "seaweed-worker")
+	cmd.Env = append(cmd.Environ(),
+		fmt.Sprintf("WEED_LANCE_NAMESPACE=http://127.0.0.1:%d", env.lancePort),
+		fmt.Sprintf("WEED_LANCE_TABLE=%s$%s$%s", bucket, namespace, table),
+		"AWS_ACCESS_KEY_ID="+accessKey,
+		"AWS_SECRET_ACCESS_KEY="+secretKey,
+		"AWS_REGION=us-east-1",
+		fmt.Sprintf("AWS_ENDPOINT_URL=http://127.0.0.1:%d", env.s3Port),
+	)
+	out, err := cmd.CombinedOutput()
+	t.Logf("lance worker:\n%s", out)
+	if err != nil {
+		t.Fatalf("the Lance maintenance worker failed: %v", err)
+	}
+}
+
+// maintainWithWorker says whether to maintain through the Rust worker.
+func maintainWithWorker(t *testing.T) bool {
+	t.Helper()
+
+	switch os.Getenv("WEED_LANCE_MAINTENANCE") {
+	case "library":
+		t.Log("maintaining through the lance library, as WEED_LANCE_MAINTENANCE asks")
+		return false
+	case "worker":
+		return true
+	}
+	if _, err := exec.LookPath("cargo"); err != nil {
+		t.Log("cargo is not installed, maintaining through the lance library rather than the worker")
+		return false
+	}
+	return true
+}
+
+func (env *environment) lance(t *testing.T, phase, bucket, namespace, table string) tally {
+	t.Helper()
+
+	out := env.runClient(t, lanceImage, phase, "/app/lance_lifecycle.py",
+		"--namespace-url", env.containerURL(env.lancePort),
+		"--s3-endpoint", env.containerURL(env.s3Port),
+		"--bucket", bucket,
+		"--namespace", namespace,
+		"--table", table,
+	)
+	if phase == "drop" {
+		return tally{}
+	}
+	return decodeTally(t, out)
+}