diff --git a/.github/workflows/s3-tables-tests.yml b/.github/workflows/s3-tables-tests.yml index 3a2667a8a..19c1395fe 100644 --- a/.github/workflows/s3-tables-tests.yml +++ b/.github/workflows/s3-tables-tests.yml @@ -439,6 +439,117 @@ jobs: path: test/s3tables/catalog_clickhouse/test-output.log retention-days: 3 + olake-iceberg-catalog-tests: + name: OLake Iceberg Catalog Integration Tests (${{ matrix.tag }}) + runs-on: ubuntu-22.04 + timeout-minutes: 30 + strategy: + fail-fast: false + matrix: + include: + # Pinned baseline, and latest so new OLake releases are exercised + # without a code change. OLake's Iceberg writer is a Java sidecar + # whose Iceberg version moves independently of the Go release, so + # the latest leg is the one that catches library drift. + - olake-image: olakego/source-postgres:v0.10.1 + tag: "v0.10.1" + - olake-image: olakego/source-postgres:latest + tag: latest + + 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 ${{ matrix.olake-image }} + pull postgres:16 + pull python:3.11-slim + + - name: Run go mod tidy + run: go mod tidy + + - name: Install SeaweedFS + run: | + go install -buildvcs=false ./weed + + - name: Run OLake Iceberg Catalog Integration Tests + timeout-minutes: 25 + working-directory: test/s3tables/catalog_olake + env: + OLAKE_IMAGE: ${{ matrix.olake-image }} + run: | + set -x + set -o pipefail + echo "=== System Information ===" + uname -a + free -h + df -h + docker info + echo "=== Starting OLake Iceberg Catalog Tests ===" + + go test -v -timeout 20m . 2>&1 | tee test-output.log || { + echo "OLake Iceberg catalog integration tests failed" + exit 1 + } + + # The suite skips itself when Docker is unavailable, so a green job is not + # by itself evidence that anything ran. Assert execution explicitly. + - name: Assert the suite actually ran + working-directory: test/s3tables/catalog_olake + run: | + log=test-output.log + if [ ! -f "$log" ]; then + echo "::error::no test-output.log; the suite did not run" + exit 1 + fi + passes=$(grep -c '^--- PASS' "$log" || true) + skips=$(grep -c '^--- SKIP' "$log" || true) + echo "top-level PASS=$passes SKIP=$skips" + if [ "$skips" -gt 0 ]; then + echo "::error::the OLake suite skipped $skips top-level test(s); the environment it needs was not provisioned, so this job proves nothing" + grep '^--- SKIP' "$log" | head -20 + exit 1 + fi + if [ "$passes" -lt 1 ]; then + echo "::error::the OLake suite recorded no passing top-level test" + exit 1 + fi + + - name: Show test output on failure + if: failure() + working-directory: test/s3tables/catalog_olake + 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|olake|postgres)" || true + echo "=== Containers ===" + docker ps -a | head -30 || true + + - name: Upload test logs on failure + if: failure() + uses: actions/upload-artifact@v7 + with: + name: olake-iceberg-catalog-test-logs-${{ matrix.tag }} + path: test/s3tables/catalog_olake/test-output.log + retention-days: 3 + polaris-integration-tests: name: Polaris Integration Tests runs-on: ubuntu-22.04 diff --git a/test/s3tables/catalog_olake/Dockerfile.reader b/test/s3tables/catalog_olake/Dockerfile.reader new file mode 100644 index 000000000..63d38077c --- /dev/null +++ b/test/s3tables/catalog_olake/Dockerfile.reader @@ -0,0 +1,14 @@ +# Strict Iceberg reader used by the OLake integration test. PyIceberg is +# deliberate here: it rejects manifests that omit the spec's field ids and +# parquet without either field ids or a name mapping, so a passing read proves +# the catalog served something every engine can consume -- not just something +# OLake itself can read back. +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install --no-cache-dir "pyiceberg[s3fs]==0.11.1" "pyarrow==25.0.0" + +COPY inspect_table.py /app/ + +ENTRYPOINT ["python3", "/app/inspect_table.py"] diff --git a/test/s3tables/catalog_olake/README.md b/test/s3tables/catalog_olake/README.md new file mode 100644 index 000000000..2e0f75272 --- /dev/null +++ b/test/s3tables/catalog_olake/README.md @@ -0,0 +1,104 @@ +# OLake Iceberg Catalog Integration Test + +An integration test for [OLake](https://github.com/datazip-inc/olake) against +SeaweedFS's Iceberg REST Catalog, in the same shape as `catalog_clickhouse`. + +## Why OLake, given we already test five engines + +Two things here are covered by nothing else in this directory. + +**It is a strict Java Iceberg client.** OLake does not write Iceberg from Go — +its Go process spawns a Java sidecar over gRPC and writes through the official +Apache Iceberg library, because the Go library has no equality deletes and CDC +needs them. So this test exercises the client class that +`weed/s3api/iceberg/metadata_compliance.go` exists to serve: the one +that fails with *"Cannot parse missing long current-snapshot-id"* when the +catalog omits spec-required keys that `iceberg-go` strips via `omitempty`. + +**It produces equality deletes.** OLake is a CDC tool. Its upsert path commits +`operation=overwrite` with an equality-delete file, and a delete manifest +alongside the data manifests. ClickHouse, StarRocks, Doris and DuckDB all only +append. + +## What it asserts + +`TestOLakeIcebergCatalog` runs six subtests against a `weed mini` cluster with +a pre-created table bucket and a Postgres source: + +| Subtest | What a failure means | +|---|---| +| `CheckDestination` | `olake check` did not reach `SUCCEEDED`, or it passed without ever loading `org.apache.iceberg.rest.RESTSessionCatalog` — the second case means the destination was never actually contacted. | +| `Discover` | OLake could not enumerate the source table, or wrote no `streams.json`. | +| `FullSyncAppendsRows` | The sync read fewer than the three seeded rows, or committed no Iceberg snapshot. | +| `StrictReaderSeesRows` | PyIceberg could not read back what the Java writer committed, or the values differ. This is the data path, not just metadata. | +| `UpsertProducesEqualityDelete` | After an `UPDATE` and a re-sync, no snapshot recorded an overwrite carrying equality deletes, or the current snapshot has no delete manifest. | +| `CompliantWriterNeedsNoRepair` | The catalog rewrote a manifest the official Iceberg Java writer produced. That is a regression in the repair gate, not a problem with OLake. | + +## The one thing this test deliberately does not check + +It does **not** assert that a reader sees the updated row and no duplicate. + +PyIceberg refuses to scan a table carrying equality deletes +([apache/iceberg#6568](https://github.com/apache/iceberg/issues/6568)) while +reading its metadata perfectly well — so a rows-mode read after the upsert would +raise, not pass. The alternative is an engine that applies equality deletes, +which for StarRocks means a 3 GB image and 12 GB of RAM in CI. + +The line drawn instead: **recording the commit correctly is the catalog's +contract; applying deletes on read is the query engine's.** The metadata +assertions cover our half. + +This was verified once by hand outside CI, on 2026-09-23, with StarRocks 4.1.4 +attached to the same catalog: after the upsert it read 3 rows / 3 distinct ids +with `id=1` showing the updated value and `_op_type=u`. If someone later wants +that inside the gate, add a reader that supports equality deletes — **do not** +"upgrade" this test to a PyIceberg rows-mode read after the upsert. It would not +pass; and if PyIceberg ever starts silently skipping deletes instead of raising, +it would pass by not looking. + +## What the config proves + +Nothing in the destination config is SeaweedFS-specific: + +```json +{ + "catalog_type": "rest", + "rest_catalog_url": "http://HOST:ICEBERG_PORT", + "iceberg_s3_path": "s3://olake-tables", + "s3_endpoint": "http://HOST:S3_PORT", + "rest_auth_type": "oauth2", + "oauth2_uri": "http://HOST:ICEBERG_PORT/v1/oauth/tokens", + "credential": "ACCESS_KEY:SECRET_KEY" +} +``` + +`catalog_type` is the generic `rest`, auth is the standard OAuth2 +client-credentials flow, and `s3_path_style` does not even need setting — +OLake turns it on by itself whenever `s3_endpoint` is non-empty. As of +OLake v0.10.1 this works with no change on either side. + +## Running it + +```sh +go test ./test/s3tables/catalog_olake/ -run TestOLakeIcebergCatalog -v -timeout 25m +``` + +Needs Docker and a `weed` binary (at `weed/weed` under the repo root, or on +`PATH`). Takes about 35 seconds. It skips rather than fails when Docker is +absent, and `SEAWEEDFS_SKIP_OLAKE_TESTS=1` skips it outright. + +Overrides: `OLAKE_IMAGE` (default `olakego/source-postgres:latest`), +`POSTGRES_IMAGE` (default `postgres:16`). + +## In CI + +Runs as `olake-iceberg-catalog-tests` in `.github/workflows/s3-tables-tests.yml`, +on a matrix of a pinned image plus `latest` — the same shape the ClickHouse job +uses, and for the same reason: OLake's Iceberg writer is a Java sidecar whose +library version moves independently of the Go release, so the `latest` leg is +what catches drift in the client rather than in OLake itself. + +The job asserts the suite actually ran — at least one top-level `--- PASS` and +zero `--- SKIP` — rather than trusting a green exit. This suite skips itself +when Docker is unavailable, and a skipped suite reporting success is how a gate +quietly stops being one. diff --git a/test/s3tables/catalog_olake/inspect_table.py b/test/s3tables/catalog_olake/inspect_table.py new file mode 100644 index 000000000..0f8edca22 --- /dev/null +++ b/test/s3tables/catalog_olake/inspect_table.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Inspect an Iceberg table written by OLake, through the SeaweedFS REST catalog. + +Two modes, because a strict reader cannot do both: + + rows -- scan the table and print "id,region,amount" per row, ordered by + id. Only valid while the table has no equality deletes. + snapshots -- print one line per snapshot with its operation and delete-file + counters, plus the manifest content kinds of the current + snapshot. + +The split exists because PyIceberg refuses to scan a table carrying equality +deletes (apache/iceberg#6568) while reading its metadata perfectly well. OLake +is a CDC tool, so its upsert path produces exactly those deletes -- asserting +the commit landed is the catalog's concern, and applying deletes on read is the +query engine's. +""" + +import argparse +import sys + +from pyiceberg.catalog import load_catalog + + +def main() -> int: + p = argparse.ArgumentParser() + p.add_argument("mode", choices=["rows", "snapshots"]) + p.add_argument("--catalog-url", required=True) + p.add_argument("--warehouse", required=True) + p.add_argument("--prefix", required=True) + p.add_argument("--s3-endpoint", required=True) + p.add_argument("--access-key", required=True) + p.add_argument("--secret-key", required=True) + p.add_argument("--region", default="us-east-1") + p.add_argument("--namespace", action="append", required=True) + p.add_argument("--table", required=True) + args = p.parse_args() + + catalog = load_catalog( + "rest", + **{ + "type": "rest", + "uri": args.catalog_url, + "warehouse": args.warehouse, + "prefix": args.prefix, + "credential": f"{args.access_key}:{args.secret_key}", + "s3.access-key-id": args.access_key, + "s3.secret-access-key": args.secret_key, + "s3.endpoint": args.s3_endpoint, + "s3.region": args.region, + "s3.path-style-access": "true", + }, + ) + + table = catalog.load_table(tuple(args.namespace) + (args.table,)) + + if args.mode == "rows": + data = table.scan().to_arrow().to_pydict() + # amount is decimal(10,2) in Postgres but arrives here as a float, whose + # repr drops trailing zeros (120.5, not 120.50). Format it to the source + # scale so the expected values in the Go test stay readable. + for row_id, region, amount in sorted( + zip(data["id"], data["region"], data["amount"]) + ): + print("%s,%s,%.2f" % (row_id, region, float(amount))) + return 0 + + print("format-version=%d" % table.metadata.format_version) + ids = table.metadata.schemas[-1].identifier_field_ids + print("identifier-field-ids=%s" % ",".join(str(i) for i in ids)) + for snap in table.metadata.snapshots: + s = snap.summary + print( + "snapshot operation=%s total-delete-files=%s added-delete-files=%s " + "added-equality-deletes=%s total-records=%s" + % ( + s.operation, + s.get("total-delete-files", "0"), + s.get("added-delete-files", "0"), + s.get("added-equality-deletes", "0"), + s.get("total-records", "0"), + ) + ) + current = table.current_snapshot() + kinds = [str(m.content).rsplit(".", 1)[-1] for m in current.manifests(table.io)] + print("current-manifest-kinds=%s" % ",".join(kinds)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test/s3tables/catalog_olake/olake_catalog_test.go b/test/s3tables/catalog_olake/olake_catalog_test.go new file mode 100644 index 000000000..de5e2327a --- /dev/null +++ b/test/s3tables/catalog_olake/olake_catalog_test.go @@ -0,0 +1,623 @@ +// Package catalog_olake provides an integration test for OLake +// (github.com/datazip-inc/olake) against the SeaweedFS Iceberg REST Catalog. +// +// OLake matters here for two reasons that no other engine in this directory +// covers. First, it does not write Iceberg from Go: it spawns a Java sidecar +// over gRPC and writes through the official Apache Iceberg library, so this is +// a strict Java client -- the class that iceberg/metadata_compliance.go exists +// to serve. Second, it is a CDC tool, so its upsert path emits equality +// deletes, which neither the ClickHouse nor the Doris test exercises. +package catalog_olake + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/testutil" +) + +const ( + olakeDefaultImage = "olakego/source-postgres:latest" + postgresDefaultImage = "postgres:16" + readerImage = "seaweedfs-olake-reader:test" + + tableBucketName = "olake-tables" + sourceDatabase = "olakedb" + sourceUser = "olake" + sourcePassword = "olakepw" + + // OLake derives the destination namespace from the source: it joins the + // connector name, database and schema. Asserting the derived name rather + // than configuring one keeps the test honest about what OLake actually + // does with our catalog. + expectedNamespace = "postgres_olakedb_public" + expectedTable = "orders" + + postgresStartTimeout = 90 * time.Second + olakeRunTimeout = 6 * time.Minute +) + +type TestEnvironment struct { + seaweedDir string + weedBinary string + dataDir string + configDir string + bindIP string + + masterPort int + masterGrpcPort int + volumePort int + volumeGrpcPort int + filerPort int + filerGrpcPort int + s3Port int + s3GrpcPort int + icebergPort int + postgresPort int + + accessKey string + secretKey string + + weedProcess *exec.Cmd + weedCancel func() + postgresContainer string +} + +func TestOLakeIcebergCatalog(t *testing.T) { + requireOLakeRuntime(t) + + env := NewTestEnvironment(t) + defer env.Cleanup(t) + + env.StartSeaweedFS(t) + env.startPostgres(t) + env.seedSource(t) + + buildReaderImage(t) + env.writeOLakeConfigs(t) + + t.Run("CheckDestination", func(t *testing.T) { + out := env.runOLake(t, "check", + "--config", "/mnt/config/source.json", + "--destination", "/mnt/config/destination.json") + if !strings.Contains(out, `"status":"SUCCEEDED"`) { + t.Fatalf("olake check did not report SUCCEEDED.\n%s", tailLines(out, 40)) + } + // The Java sidecar, not the Go process, is what talks to our catalog. + // If this line is missing the check passed without exercising the + // Iceberg REST path at all. + if !strings.Contains(out, "org.apache.iceberg.rest.RESTSessionCatalog") { + t.Errorf("check passed but never loaded the Iceberg REST catalog; "+ + "the destination may not have been contacted.\n%s", tailLines(out, 40)) + } + }) + + t.Run("Discover", func(t *testing.T) { + out := env.runOLake(t, "discover", + "--config", "/mnt/config/source.json", + "--destination", "/mnt/config/destination.json") + if !strings.Contains(out, `"stream_name":"`+expectedTable+`"`) { + t.Fatalf("discover did not report the %s stream.\n%s", expectedTable, tailLines(out, 20)) + } + if _, err := os.Stat(filepath.Join(env.configDir, "streams.json")); err != nil { + t.Fatalf("discover did not write streams.json: %v", err) + } + }) + + t.Run("FullSyncAppendsRows", func(t *testing.T) { + out := env.runOLake(t, "sync", + "--config", "/mnt/config/source.json", + "--destination", "/mnt/config/destination.json", + "--streams", "/mnt/config/streams.json", + "--state", "/mnt/config/state.json") + if !strings.Contains(out, "Total records read: 3") { + t.Fatalf("sync did not read the three seeded rows.\n%s", tailLines(out, 30)) + } + if !strings.Contains(out, "Committed snapshot") { + t.Fatalf("sync never committed an Iceberg snapshot.\n%s", tailLines(out, 30)) + } + }) + + t.Run("StrictReaderSeesRows", func(t *testing.T) { + rows := env.readTable(t, "rows") + want := []string{"1,us-east,120.50", "2,us-west,87.20", "3,eu-west,210.00"} + got := nonEmptyLines(rows) + if len(got) != len(want) { + t.Fatalf("PyIceberg read %d rows, want %d.\n%s", len(got), len(want), rows) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("row %d = %q, want %q", i, got[i], want[i]) + } + } + }) + + t.Run("UpsertProducesEqualityDelete", func(t *testing.T) { + // Push the row past the incremental cursor so the next sync re-emits it + // as an update rather than skipping it. + env.execSQL(t, "UPDATE orders SET amount=999.99, region='ap-south' WHERE id=1") + + out := env.runOLake(t, "sync", + "--config", "/mnt/config/source.json", + "--destination", "/mnt/config/destination.json", + "--streams", "/mnt/config/streams.json", + "--state", "/mnt/config/state.json") + if !strings.Contains(out, "delete files") { + t.Fatalf("re-sync committed no delete files.\n%s", tailLines(out, 30)) + } + + meta := env.readTable(t, "snapshots") + if !strings.Contains(meta, "format-version=2") { + t.Errorf("expected Iceberg format-version 2, got:\n%s", meta) + } + if !strings.Contains(meta, "identifier-field-ids=") || + strings.Contains(meta, "identifier-field-ids=\n") { + t.Errorf("table carries no identifier fields, so OLake could not "+ + "have upserted:\n%s", meta) + } + + var sawOverwrite bool + for _, line := range nonEmptyLines(meta) { + if !strings.HasPrefix(line, "snapshot operation=") { + continue + } + if strings.Contains(line, "operation=Operation.OVERWRITE") && + !strings.Contains(line, "added-equality-deletes=0") { + sawOverwrite = true + } + } + if !sawOverwrite { + t.Errorf("no snapshot recorded an overwrite carrying equality "+ + "deletes:\n%s", meta) + } + if !strings.Contains(meta, "current-manifest-kinds=") || + !strings.Contains(meta, "DELETES") { + t.Errorf("current snapshot has no delete manifest:\n%s", meta) + } + // Deliberately NOT asserted here: that a reader sees the updated value + // and no duplicate row. PyIceberg refuses to scan a table carrying + // equality deletes (apache/iceberg#6568), so a rows-mode read would + // fail rather than pass, and swapping in an engine that can apply them + // costs this test a multi-gigabyte image. Applying deletes on read is + // the engine's contract; recording the commit correctly is ours, and + // that is what the assertions above cover. See the README. + }) + + t.Run("CompliantWriterNeedsNoRepair", func(t *testing.T) { + names := env.listTableMetadata(t) + if len(names) == 0 { + t.Fatalf("no metadata files found for %s.%s", expectedNamespace, expectedTable) + } + for _, n := range names { + if strings.HasPrefix(n, "repaired-") { + t.Errorf("catalog repaired a manifest written by OLake (%s); "+ + "the official Iceberg Java writer is expected to be "+ + "spec-compliant, so this is a regression in the repair "+ + "gate rather than in OLake", n) + } + } + }) +} + +func NewTestEnvironment(t *testing.T) *TestEnvironment { + t.Helper() + + wd, err := os.Getwd() + if err != nil { + t.Fatalf("Failed to get working directory: %v", err) + } + + seaweedDir := wd + for i := 0; i < 6; 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, err := os.Stat(weedBinary); err != nil || info.IsDir() { + weedBinary = filepath.Join(seaweedDir, "weed", "weed", "weed") + if info, err := os.Stat(weedBinary); err != nil || info.IsDir() { + weedBinary = "weed" + if _, err := exec.LookPath(weedBinary); err != nil { + t.Skip("weed binary not found, skipping integration test") + } + } + } + + dataDir, err := os.MkdirTemp("", "seaweed-olake-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + configDir := filepath.Join(dataDir, "olake") + if err := os.MkdirAll(configDir, 0755); err != nil { + t.Fatalf("Failed to create config dir: %v", err) + } + + // 9 for the mini cluster, 1 for the Postgres source mapped on the host. + ports := testutil.MustAllocatePorts(t, 10) + + return &TestEnvironment{ + seaweedDir: seaweedDir, + weedBinary: weedBinary, + dataDir: dataDir, + configDir: configDir, + 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], + icebergPort: ports[8], + postgresPort: ports[9], + accessKey: "AKIAIOSFODNN7EXAMPLE", + secretKey: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + } +} + +func (env *TestEnvironment) StartSeaweedFS(t *testing.T) { + t.Helper() + + iamConfigPath, err := testutil.WriteIAMConfig(env.dataDir, env.accessKey, env.secretKey) + if err != nil { + t.Fatalf("Failed to create IAM config: %v", err) + } + + cmd := exec.Command(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.config", iamConfigPath, + // Pre-create the table bucket the way an operator would, rather than + // reaching for the S3 Tables control plane from the test. + "-tableBucket", tableBucketName, + "-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("Failed to start SeaweedFS: %v", err) + } + env.weedProcess = cmd + env.weedCancel = func() { + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + } + + url := fmt.Sprintf("http://%s:%d/v1/config", env.bindIP, env.icebergPort) + if !waitForService(url, 45*time.Second) { + t.Fatalf("Iceberg REST API did not become ready at %s", url) + } +} + +func (env *TestEnvironment) startPostgres(t *testing.T) { + t.Helper() + + name := fmt.Sprintf("olake-pg-%d", env.postgresPort) + _ = exec.Command("docker", "rm", "-f", name).Run() + + cmd := exec.Command("docker", "run", "-d", "--name", name, + "-e", "POSTGRES_USER="+sourceUser, + "-e", "POSTGRES_PASSWORD="+sourcePassword, + "-e", "POSTGRES_DB="+sourceDatabase, + "-p", fmt.Sprintf("%d:5432", env.postgresPort), + postgresImage(), + ) + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("Failed to start Postgres: %v\n%s", err, out) + } + env.postgresContainer = name + + deadline := time.Now().Add(postgresStartTimeout) + for time.Now().Before(deadline) { + probe := exec.Command("docker", "exec", name, + "pg_isready", "-U", sourceUser, "-d", sourceDatabase) + if err := probe.Run(); err == nil { + return + } + time.Sleep(2 * time.Second) + } + t.Fatalf("Postgres did not become ready within %s", postgresStartTimeout) +} + +func (env *TestEnvironment) seedSource(t *testing.T) { + t.Helper() + env.execSQL(t, ` +CREATE TABLE orders ( + id int PRIMARY KEY, + region text, + amount numeric(10,2), + order_ts timestamp +); +INSERT INTO orders VALUES + (1,'us-east',120.50,'2026-07-27 09:15'), + (2,'us-west', 87.20,'2026-07-27 09:20'), + (3,'eu-west',210.00,'2026-07-27 09:31'); +ALTER TABLE orders REPLICA IDENTITY FULL;`) +} + +func (env *TestEnvironment) execSQL(t *testing.T, sqlText string) { + t.Helper() + cmd := exec.Command("docker", "exec", env.postgresContainer, + "psql", "-v", "ON_ERROR_STOP=1", "-U", sourceUser, "-d", sourceDatabase, + "-c", sqlText) + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("psql failed: %v\n%s", err, out) + } +} + +// writeOLakeConfigs writes the source and destination configs. The destination +// is the point of this test: nothing in it is SeaweedFS-specific. catalog_type +// is the generic "rest", auth is the standard OAuth2 client-credentials flow, +// and path-style access is not even set here because OLake turns it on by +// itself whenever s3_endpoint is non-empty. +func (env *TestEnvironment) writeOLakeConfigs(t *testing.T) { + t.Helper() + + source := map[string]any{ + "host": env.bindIP, + "port": env.postgresPort, + "database": sourceDatabase, + "username": sourceUser, + "password": sourcePassword, + "jdbc_url_params": map[string]any{}, + "ssl": map[string]any{"mode": "disable"}, + "update_method": map[string]any{"type": "Standalone"}, + "max_threads": 2, + "retry_count": 0, + } + + catalogURL := fmt.Sprintf("http://%s:%d", env.bindIP, env.icebergPort) + destination := map[string]any{ + "type": "ICEBERG", + "writer": map[string]any{ + "catalog_type": "rest", + "rest_catalog_url": catalogURL, + "catalog_name": "olake", + "iceberg_s3_path": "s3://" + tableBucketName, + "s3_endpoint": fmt.Sprintf("http://%s:%d", env.bindIP, env.s3Port), + "s3_use_ssl": false, + "s3_path_style": true, + "aws_region": "us-east-1", + "aws_access_key": env.accessKey, + "aws_secret_key": env.secretKey, + "rest_auth_type": "oauth2", + "oauth2_uri": catalogURL + "/v1/oauth/tokens", + "credential": env.accessKey + ":" + env.secretKey, + }, + } + + writeJSON(t, filepath.Join(env.configDir, "source.json"), source) + writeJSON(t, filepath.Join(env.configDir, "destination.json"), destination) + if err := os.WriteFile(filepath.Join(env.configDir, "state.json"), []byte("{}\n"), 0644); err != nil { + t.Fatalf("Failed to write state.json: %v", err) + } +} + +func (env *TestEnvironment) runOLake(t *testing.T, args ...string) string { + t.Helper() + + full := append([]string{ + "run", "--rm", + "-v", dockerMount(env.configDir) + ":/mnt/config", + olakeImage(), + }, args...) + + cmd := exec.Command("docker", full...) + done := make(chan struct{}) + var out []byte + var err error + go func() { + out, err = cmd.CombinedOutput() + close(done) + }() + select { + case <-done: + case <-time.After(olakeRunTimeout): + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + t.Fatalf("olake %s did not finish within %s", args[0], olakeRunTimeout) + } + if err != nil { + t.Fatalf("olake %s failed: %v\n%s", args[0], err, tailLines(string(out), 40)) + } + return string(out) +} + +func (env *TestEnvironment) readTable(t *testing.T, mode string) string { + t.Helper() + + cmd := exec.Command("docker", "run", "--rm", readerImage, mode, + "--catalog-url", fmt.Sprintf("http://%s:%d", env.bindIP, env.icebergPort), + "--warehouse", "s3://"+tableBucketName, + "--prefix", tableBucketName, + "--s3-endpoint", fmt.Sprintf("http://%s:%d", env.bindIP, env.s3Port), + "--access-key", env.accessKey, + "--secret-key", env.secretKey, + "--namespace", expectedNamespace, + "--table", expectedTable, + ) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("reader (%s) failed: %v\n%s", mode, err, tailLines(string(out), 30)) + } + return string(out) +} + +// listTableMetadata lists the table's metadata directory through the filer so +// the repair assertion looks at real objects rather than at what the catalog +// reports about itself. +func (env *TestEnvironment) listTableMetadata(t *testing.T) []string { + t.Helper() + + url := fmt.Sprintf("http://%s:%d/buckets/%s/%s/%s/metadata/?limit=200", + env.bindIP, env.filerPort, tableBucketName, expectedNamespace, expectedTable) + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + t.Fatalf("Failed to build filer request: %v", err) + } + req.Header.Set("Accept", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("filer listing failed: %v", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + t.Fatalf("filer listing returned %d: %s", resp.StatusCode, body) + } + + var parsed struct { + Entries []struct { + FullPath string `json:"FullPath"` + } `json:"Entries"` + } + if err := json.Unmarshal(body, &parsed); err != nil { + t.Fatalf("Failed to parse filer listing: %v\n%s", err, body) + } + + names := make([]string, 0, len(parsed.Entries)) + for _, e := range parsed.Entries { + names = append(names, e.FullPath[strings.LastIndex(e.FullPath, "/")+1:]) + } + return names +} + +func (env *TestEnvironment) Cleanup(t *testing.T) { + t.Helper() + + if env.postgresContainer != "" { + _ = exec.Command("docker", "rm", "-f", env.postgresContainer).Run() + } + if env.weedCancel != nil { + env.weedCancel() + } + if env.weedProcess != nil { + _ = env.weedProcess.Wait() + } + if env.dataDir != "" { + _ = os.RemoveAll(env.dataDir) + } +} + +func buildReaderImage(t *testing.T) { + t.Helper() + + wd, err := os.Getwd() + if err != nil { + t.Fatalf("Failed to get working directory: %v", err) + } + cmd := exec.Command("docker", "build", + "-f", filepath.Join(wd, "Dockerfile.reader"), + "-t", readerImage, wd) + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("Failed to build reader image: %v\n%s", err, tailLines(string(out), 30)) + } +} + +func olakeImage() string { + if v := os.Getenv("OLAKE_IMAGE"); v != "" { + return v + } + return olakeDefaultImage +} + +func postgresImage() string { + if v := os.Getenv("POSTGRES_IMAGE"); v != "" { + return v + } + return postgresDefaultImage +} + +func requireOLakeRuntime(t *testing.T) { + t.Helper() + if os.Getenv("SEAWEEDFS_SKIP_OLAKE_TESTS") != "" { + t.Skip("SEAWEEDFS_SKIP_OLAKE_TESTS set") + } + if _, err := exec.LookPath("docker"); err != nil { + t.Skip("docker not available, skipping OLake integration test") + } + if err := exec.Command("docker", "info").Run(); err != nil { + t.Skip("docker daemon not reachable, skipping OLake integration test") + } +} + +func waitForService(url string, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + client := &http.Client{Timeout: 3 * time.Second} + for time.Now().Before(deadline) { + resp, err := client.Get(url) + if err == nil { + resp.Body.Close() + if resp.StatusCode < 500 { + return true + } + } + time.Sleep(time.Second) + } + return false +} + +func writeJSON(t *testing.T, path string, v any) { + t.Helper() + body, err := json.MarshalIndent(v, "", " ") + if err != nil { + t.Fatalf("Failed to marshal %s: %v", path, err) + } + if err := os.WriteFile(path, append(body, '\n'), 0644); err != nil { + t.Fatalf("Failed to write %s: %v", path, err) + } +} + +// dockerMount normalises a host path for a -v bind mount. Docker Desktop +// accepts forward slashes on Windows; the native separator it does not. +func dockerMount(path string) string { + return strings.ReplaceAll(path, `\`, "/") +} + +func nonEmptyLines(s string) []string { + var out []string + for _, line := range strings.Split(s, "\n") { + if trimmed := strings.TrimSpace(line); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} + +func tailLines(s string, n int) string { + lines := strings.Split(strings.TrimRight(s, "\n"), "\n") + if len(lines) > n { + lines = lines[len(lines)-n:] + } + return strings.Join(lines, "\n") +}