test(s3tables): add an OLake Iceberg catalog integration test (#11441)

OLake (github.com/datazip-inc/olake) is a CDC and ingestion engine that writes
Apache Iceberg. It covers two paths none of the existing catalog suites reach.

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. That makes it the client class weed/s3api/iceberg's metadata
compliance backfill exists to serve: the one that fails with "Cannot parse
missing long current-snapshot-id" when spec-required keys are omitted.

And it produces equality deletes. Its upsert path commits operation=overwrite
with an equality-delete file and a delete manifest. ClickHouse, Doris, Trino,
Spark and DuckDB all only append, so nothing else in this directory exercises
a delete manifest at all.

Six subtests: the destination check reaches SUCCEEDED and actually loads the
REST catalog; discover enumerates the source; a full sync commits a snapshot;
PyIceberg reads back what the Java writer committed; an update plus a re-sync
records an overwrite carrying equality deletes with a delete manifest in the
current snapshot; and the catalog does not rewrite manifests the official Java
writer produced, which gives the manifest-repair path a negative test to go
with the ClickHouse positive one.

What it deliberately does not assert is a delete-applied read. PyIceberg
refuses to scan a table carrying equality deletes (apache/iceberg#6568) while
reading its metadata fine, and an engine that can apply them costs a
multi-gigabyte image. Recording the commit correctly is the catalog's
contract; applying deletes on read is the engine's. The README says so, says
the read half was verified by hand once with StarRocks, and warns against
later "upgrading" this to a PyIceberg rows read — which would either fail or,
if PyIceberg ever starts skipping deletes instead of raising, pass by not
looking.

The workflow job asserts the suite actually ran rather than trusting a green
exit, for the same reason: at least one top-level PASS and zero SKIP.

No product change — the destination config is the generic catalog_type=rest
with the standard OAuth2 client-credentials flow, and s3_path_style is not even
set, since OLake turns it on itself whenever s3_endpoint is non-empty.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

.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

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"]

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.

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())

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")
+}