diff --git a/.github/workflows/s3-tables-tests.yml b/.github/workflows/s3-tables-tests.yml index ddbf01292..c215e32af 100644 --- a/.github/workflows/s3-tables-tests.yml +++ b/.github/workflows/s3-tables-tests.yml @@ -6,6 +6,7 @@ on: - 'weed/s3api/**' - 'weed/filer/**' - 'weed/server/**' + - 'weed/worker/tasks/iceberg/**' - 'test/s3tables/**' - 'go.mod' - 'go.sum' @@ -944,6 +945,74 @@ jobs: path: test/s3tables/catalog_lancedb/test-output.log retention-days: 3 + table-lifecycle-tests: + name: Table Lifecycle Integration Tests + runs-on: ubuntu-22.04 + timeout-minutes: 40 + + steps: + - name: Check out code + uses: actions/checkout@v7 + + - name: Set up Go + uses: actions/setup-go@v7 + with: + go-version-file: 'go.mod' + id: go + + - name: Configure Docker Hub mirror + run: | + echo '{"registry-mirrors": ["https://mirror.gcr.io"]}' | sudo tee /etc/docker/daemon.json + sudo systemctl restart docker + + - name: Pre-pull images + run: | + pull() { for i in 1 2 3; do docker pull "$1" && return 0; sleep 15; done; return 1; } + pull python:3.11-slim + pull duckdb/duckdb:latest + + - name: Run go mod tidy + run: go mod tidy + + - name: Build SeaweedFS + run: | + cd weed && go build -buildvcs=false . + + - name: Run Table Lifecycle Integration Tests + timeout-minutes: 35 + working-directory: test/s3tables/lifecycle + env: + # The Rust worker's own tests cover its handlers; a cold build of the + # lance crate costs more here than the layer it would be checking. + WEED_LANCE_MAINTENANCE: library + run: | + set -x + set -o pipefail + go test -v -timeout 30m . 2>&1 | tee test-output.log || { + echo "Table lifecycle integration tests failed" + exit 1 + } + + - name: Show test output on failure + if: failure() + working-directory: test/s3tables/lifecycle + run: | + echo "=== Test Output ===" + if [ -f test-output.log ]; then + tail -200 test-output.log + fi + + echo "=== Process information ===" + ps aux | grep -E "(weed|test|docker)" || true + + - name: Upload test logs on failure + if: failure() + uses: actions/upload-artifact@v7 + with: + name: table-lifecycle-test-logs + path: test/s3tables/lifecycle/test-output.log + retention-days: 3 + s3-tables-build-verification: name: S3 Tables Build Verification runs-on: ubuntu-22.04 diff --git a/seaweed-worker/crates/lance/tests/common/mod.rs b/seaweed-worker/crates/lance/tests/common/mod.rs new file mode 100644 index 000000000..31f603a5b --- /dev/null +++ b/seaweed-worker/crates/lance/tests/common/mod.rs @@ -0,0 +1,97 @@ +//! What every integration test in this crate needs to drive a handler: a sender +//! that keeps what the handler sent, and the two settings that make a handler +//! usable against a test gateway. +//! +//! Each test binary compiles this module on its own and uses part of it. +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Mutex; + +use anyhow::Result; +use seaweed_worker_core::pb::{ + config_value::Kind, ActivityEvent, ConfigValue, DetectionComplete, DetectionProposals, + JobCompleted, JobProgressUpdate, JobProposal, ObjectObservation, WorkerObservations, +}; +use seaweed_worker_core::{DetectionSender, ExecutionSender}; + +/// Keeps everything a handler reports, so a test can assert on it afterwards. +#[derive(Default)] +pub struct Recorder { + pub proposals: Mutex>, + pub observations: Mutex>, + pub completed: Mutex>, +} + +impl DetectionSender for Recorder { + fn send_proposals(&self, proposals: DetectionProposals) -> Result<()> { + self.proposals.lock().unwrap().extend(proposals.proposals); + Ok(()) + } + fn send_complete(&self, _complete: DetectionComplete) -> Result<()> { + Ok(()) + } + fn send_activity(&self, _activity: ActivityEvent) -> Result<()> { + Ok(()) + } + fn send_observations(&self, observations: WorkerObservations) -> Result<()> { + self.observations + .lock() + .unwrap() + .extend(observations.observations); + Ok(()) + } +} + +impl ExecutionSender for Recorder { + fn send_progress(&self, _progress: JobProgressUpdate) -> Result<()> { + Ok(()) + } + fn send_completed(&self, completed: JobCompleted) -> Result<()> { + self.completed.lock().unwrap().push(completed); + Ok(()) + } +} + +/// The gateway to run against, or nothing, in which case the caller skips: +/// these handlers rewrite real files, and there is nothing to learn from them +/// against a fake. +pub fn namespace_url() -> Option { + std::env::var("WEED_LANCE_NAMESPACE") + .ok() + .filter(|s| !s.is_empty()) +} + +pub fn int_config(name: &str, value: i64) -> HashMap { + let mut values = HashMap::new(); + values.insert( + name.to_string(), + ConfigValue { + kind: Some(Kind::Int64Value(value)), + }, + ); + values +} + +/// What to reach the data with where the namespace vends nothing of its own - +/// a gateway without STS vends no credentials, and one bound to a wildcard +/// address vends no endpoint either. Taken from the environment so a harness +/// can point these tests at a gateway that does check what it is given. +pub fn fallback() -> weed_lance_worker::dataset::FallbackOptions { + let mut options = weed_lance_worker::dataset::FallbackOptions::new(); + let from_env = + |name: &str, default: &str| std::env::var(name).unwrap_or_else(|_| default.to_string()); + options.insert( + "aws_access_key_id".to_string(), + from_env("AWS_ACCESS_KEY_ID", "any"), + ); + options.insert( + "aws_secret_access_key".to_string(), + from_env("AWS_SECRET_ACCESS_KEY", "any"), + ); + if let Ok(endpoint) = std::env::var("AWS_ENDPOINT_URL") { + options.insert("aws_endpoint".to_string(), endpoint); + options.insert("allow_http".to_string(), "true".to_string()); + } + options +} diff --git a/seaweed-worker/crates/lance/tests/compaction.rs b/seaweed-worker/crates/lance/tests/compaction.rs index 3c83107ed..5de4cb379 100644 --- a/seaweed-worker/crates/lance/tests/compaction.rs +++ b/seaweed-worker/crates/lance/tests/compaction.rs @@ -5,82 +5,26 @@ //! nothing to learn from it against a fake. use std::collections::HashMap; -use std::sync::Mutex; use anyhow::Result; use seaweed_worker_core::pb::{ - config_value::Kind, ConfigValue, DetectionComplete, DetectionProposals, ExecuteJobRequest, - JobCompleted, JobProgressUpdate, JobProposal, JobSpec, RunDetectionRequest, + config_value::Kind, ExecuteJobRequest, JobSpec, RunDetectionRequest, }; -use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler, PreviewProvider}; +use seaweed_worker_core::{JobHandler, PreviewProvider}; use weed_lance_worker::catalog::NamespaceClient; use weed_lance_worker::jobs::cleanup::CleanupVersionsHandler; use weed_lance_worker::jobs::compact::{CompactHandler, JOB_TYPE}; use weed_lance_worker::jobs::indices::OptimizeIndicesHandler; use weed_lance_worker::preview::LancePreview; -#[derive(Default)] -struct Recorder { - proposals: Mutex>, - observations: Mutex>, - completed: Mutex>, -} - -impl DetectionSender for Recorder { - fn send_proposals(&self, proposals: DetectionProposals) -> Result<()> { - self.proposals.lock().unwrap().extend(proposals.proposals); - Ok(()) - } - fn send_complete(&self, _complete: DetectionComplete) -> Result<()> { - Ok(()) - } - fn send_activity(&self, _activity: seaweed_worker_core::pb::ActivityEvent) -> Result<()> { - Ok(()) - } - fn send_observations( - &self, - observations: seaweed_worker_core::pb::WorkerObservations, - ) -> Result<()> { - self.observations - .lock() - .unwrap() - .extend(observations.observations); - Ok(()) - } -} - -impl ExecutionSender for Recorder { - fn send_progress(&self, _progress: JobProgressUpdate) -> Result<()> { - Ok(()) - } - fn send_completed(&self, completed: JobCompleted) -> Result<()> { - self.completed.lock().unwrap().push(completed); - Ok(()) - } -} +mod common; +use common::{fallback, int_config, namespace_url, Recorder}; /// These tests drive one live gateway and one shared catalog: `list_all_tables` /// sweeps everything, so a table another test is writing shows up in this test's /// detection. Rust runs a binary's tests concurrently, so take a lock. static GATEWAY: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); -fn namespace_url() -> Option { - std::env::var("WEED_LANCE_NAMESPACE") - .ok() - .filter(|s| !s.is_empty()) -} - -fn int_config(name: &str, value: i64) -> HashMap { - let mut values = HashMap::new(); - values.insert( - name.to_string(), - ConfigValue { - kind: Some(Kind::Int64Value(value)), - }, - ); - values -} - /// Declares a table through the namespace and writes `fragments` one-row /// appends into it, so a test brings its own state instead of depending on /// whatever a previous run left behind. @@ -205,15 +149,12 @@ async fn compacts_a_fragmented_table() { eprintln!("WEED_LANCE_NAMESPACE is unset, skipping"); return; }; - let mut fallback = weed_lance_worker::dataset::FallbackOptions::new(); - fallback.insert("aws_access_key_id".to_string(), "any".to_string()); - fallback.insert("aws_secret_access_key".to_string(), "any".to_string()); // Seeded here rather than by a script, so the test is repeatable: a previous // run compacts the table it depended on. let encoded = seed_fragmented_table(&url, "compactme", 12) .await .expect("seed a fragmented table"); - let handler = CompactHandler::new(url).with_fallback(fallback); + let handler = CompactHandler::new(url).with_fallback(fallback()); let recorder = Recorder::default(); let request = RunDetectionRequest { @@ -288,13 +229,6 @@ async fn compacts_a_fragmented_table() { eprintln!("compaction result: {summary}"); } -fn fallback() -> weed_lance_worker::dataset::FallbackOptions { - let mut options = weed_lance_worker::dataset::FallbackOptions::new(); - options.insert("aws_access_key_id".to_string(), "any".to_string()); - options.insert("aws_secret_access_key".to_string(), "any".to_string()); - options -} - /// A table with more versions than the floor is proposed, and running the job /// reports what it removed. The compaction test above leaves one behind. #[tokio::test] diff --git a/seaweed-worker/crates/lance/tests/lifecycle.rs b/seaweed-worker/crates/lance/tests/lifecycle.rs new file mode 100644 index 000000000..e0d25ba31 --- /dev/null +++ b/seaweed-worker/crates/lance/tests/lifecycle.rs @@ -0,0 +1,123 @@ +//! Maintains one table someone else wrote, and nothing else. +//! +//! The Go suite in test/s3tables/lifecycle drives a table through its whole +//! life - declared through the namespace, written by a Lance client, maintained, +//! read again, dropped - and this is the maintenance step of it, so that step +//! goes through the handlers a deployed worker runs rather than through the two +//! lance calls they wrap. The table to work on comes in as WEED_LANCE_TABLE +//! because the table is the Go test's, seeded and checked there. +//! +//! Skipped, like the rest of this crate's integration tests, unless +//! WEED_LANCE_NAMESPACE names a live gateway. + +use std::collections::HashMap; + +use seaweed_worker_core::pb::{ConfigValue, ExecuteJobRequest, JobSpec, RunDetectionRequest}; +use seaweed_worker_core::JobHandler; +use weed_lance_worker::jobs::cleanup::{self, CleanupVersionsHandler}; +use weed_lance_worker::jobs::compact::{CompactHandler, JOB_TYPE as COMPACT_JOB_TYPE}; + +mod common; +use common::{fallback, int_config, namespace_url, Recorder}; + +fn table() -> Option { + std::env::var("WEED_LANCE_TABLE") + .ok() + .filter(|s| !s.is_empty()) +} + +/// Compacts the named table's fragments, then drops the versions compaction +/// left behind. Both go through the handler's own detect-then-execute path: a +/// proposal the worker would not have made is not one worth running. +#[tokio::test] +async fn maintains_the_named_table() { + let (Some(url), Some(name)) = (namespace_url(), table()) else { + eprintln!("WEED_LANCE_NAMESPACE or WEED_LANCE_TABLE is unset, skipping"); + return; + }; + + // A table written as a fragment per append is the case this exists for, so + // anything above one fragment is worth merging here. + let mut config = int_config("min_fragments", 2); + config.extend(int_config("target_rows_per_fragment", 1_048_576)); + let compact = CompactHandler::new(url.clone()).with_fallback(fallback()); + run(&compact, COMPACT_JOB_TYPE, &name, config, true).await; + + // Keep the current version and retain nothing else, so the versions + // compaction superseded are actually removed rather than counted. + let mut config = int_config("min_versions_to_keep", 1); + config.extend(int_config("retain_hours", 0)); + let cleanup = CleanupVersionsHandler::new(url).with_fallback(fallback()); + run(&cleanup, cleanup::JOB_TYPE, &name, config, true).await; +} + +/// Detects, finds the proposal for `name`, and executes it. The config carries +/// every key both halves read; each ignores what it does not know. +async fn run( + handler: &H, + job_type: &str, + name: &str, + config: HashMap, + required: bool, +) { + let recorder = Recorder::default(); + handler + .detect( + &RunDetectionRequest { + request_id: format!("detect-{job_type}"), + job_type: job_type.to_string(), + worker_config_values: config.clone(), + ..Default::default() + }, + &recorder, + ) + .await + .unwrap_or_else(|err| panic!("{job_type} detection failed: {err}")); + + let proposals = recorder.proposals.lock().unwrap().clone(); + let Some(proposal) = proposals.iter().find(|p| p.summary.contains(name)) else { + assert!( + !required, + "{job_type} proposed nothing for {name}, out of {} proposals", + proposals.len() + ); + eprintln!("{job_type}: nothing to do for {name}"); + return; + }; + + handler + .execute( + &ExecuteJobRequest { + request_id: format!("execute-{job_type}"), + job: Some(JobSpec { + job_id: format!("job-{job_type}"), + job_type: job_type.to_string(), + parameters: proposal.parameters.clone(), + ..Default::default() + }), + worker_config_values: config, + ..Default::default() + }, + &recorder, + ) + .await + .unwrap_or_else(|err| panic!("{job_type} execution failed: {err}")); + + let completed = recorder.completed.lock().unwrap().clone(); + let result = completed + .last() + .unwrap_or_else(|| panic!("{job_type} reported no completion")); + assert!( + result.success, + "{job_type} reported failure: {}", + result.error_message + ); + eprintln!( + "{job_type}: {}", + result + .result + .as_ref() + .map(|r| r.summary.clone()) + .unwrap_or_default() + ); +} diff --git a/test/s3tables/lifecycle/Dockerfile.lance b/test/s3tables/lifecycle/Dockerfile.lance new file mode 100644 index 000000000..519b9b93f --- /dev/null +++ b/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"] diff --git a/test/s3tables/lifecycle/Dockerfile.pyiceberg b/test/s3tables/lifecycle/Dockerfile.pyiceberg new file mode 100644 index 000000000..6c1c2903e --- /dev/null +++ b/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"] diff --git a/test/s3tables/lifecycle/README.md b/test/s3tables/lifecycle/README.md new file mode 100644 index 000000000..2e36bc27c --- /dev/null +++ b/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. diff --git a/test/s3tables/lifecycle/cluster_test.go b/test/s3tables/lifecycle/cluster_test.go new file mode 100644 index 000000000..5c5236596 --- /dev/null +++ b/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) +} diff --git a/test/s3tables/lifecycle/iceberg_lifecycle.py b/test/s3tables/lifecycle/iceberg_lifecycle.py new file mode 100644 index 000000000..8b4545b78 --- /dev/null +++ b/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() diff --git a/test/s3tables/lifecycle/iceberg_lifecycle_test.go b/test/s3tables/lifecycle/iceberg_lifecycle_test.go new file mode 100644 index 000000000..2c34d7b5c --- /dev/null +++ b/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) + } +} diff --git a/test/s3tables/lifecycle/lance_lifecycle.py b/test/s3tables/lifecycle/lance_lifecycle.py new file mode 100644 index 000000000..60d62604c --- /dev/null +++ b/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()) diff --git a/test/s3tables/lifecycle/lance_lifecycle_test.go b/test/s3tables/lifecycle/lance_lifecycle_test.go new file mode 100644 index 000000000..ea076eef5 --- /dev/null +++ b/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) +}