From 0dfaa103d0ba20ed9877776f4776724e9b537885 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 21 Aug 2026 15:16:11 -0700 Subject: [PATCH] test: take a table through its whole life, for Iceberg and Lance (#10862) * lance worker: share the integration tests' scaffolding The recorder that keeps what a handler sent, the config builder and the storage-option fallback all lived inside compaction.rs, so a second test binary would have had to copy them. They move to tests/common. The fallback now reads AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_ENDPOINT_URL from the environment, defaulting to what it used before. A harness can then point these tests at a gateway that checks what it is given rather than one that accepts anything. * lance worker: maintain one named table, for a harness to drive Compacts and cleans up whatever WEED_LANCE_TABLE names, through the handlers' own detect-then-execute path: a proposal the worker would not have made is not one worth running. The existing tests seed the tables they check. This one deliberately does not, so a harness that has already written a table and knows what is in it can have the real handlers maintain it and then read it back. * test: take a table through its whole life, for Iceberg and Lance Created in the catalog, filled by a real client, maintained by the worker, read again, dropped. The step nothing was checking is the read after maintenance: compaction once rewrote every dictionary-encoded column onto a single value and shipped, because the maintenance tests were thorough about sequence numbers, manifest entries and metadata versions and none of them opened the parquet file the worker had just written. So the assertion is a tally - row count, the cardinality of each dictionary-encoded column, and an md5 over whole rows - taken before maintenance and again after, required to be equal. The cardinalities name the failure that happened; the digest catches a rewrite that keeps every column's cardinality and hands the values to the wrong rows. A compaction that merged nothing fails rather than passes, or the read afterwards is checking a file the worker never wrote. The Iceberg half runs two clients. DuckDB is the one the corruption was reported against and the only one here that writes the deprecated PLAIN_DICTIONARY encoding, which parquet-go normalizes away on write, so a Go writer cannot produce it. PyIceberg writes the modern spelling. Pinning parquet-go back to v0.30.1 fails the DuckDB half and passes the PyIceberg one, which is why both are here. Lance maintenance lives in the Rust worker, so it runs there where cargo is installed and through the two lance calls those handlers wrap where it is not. WEED_LANCE_MAINTENANCE picks one instead of letting the test guess. * ci: run the table lifecycle tests CI maintains the Lance table through the lance library rather than the worker: a cold build of the lance crate costs more than the glue it would be checking, and the worker's own tests cover its handlers. The suite drives the Iceberg maintenance worker, so a change to it now triggers this workflow too. * test: let the lifecycle harness fail instead of skipping Setup failures all exited zero, so a cluster that would not come up, or a port allocation that lost, reported a green run for code nothing had executed. That is the failure mode this whole directory exists to close, and it was in the harness itself. Only a checkout without a weed binary skips now, and it runs the tests so each one says so rather than the package quietly passing. Everything else fails. The filer existence probe gets a deadline while I am here: it ran without one, so an unresponsive filer would hang the suite past every timeout the clients have. * test: make the lifecycle checks check what they claim to Three of them could pass without having looked. The DuckDB skip matched "syntax error", "not implemented" and "Failed to load" anywhere in the output, in any phase. A parse error in the SQL this test generates, or a refusal from our own catalog, would have taken the only coverage of the PLAIN_DICTIONARY encoding out of CI and left it green. It now matches the extension failing to install, and only in the phase that installs it. Everything past LOAD is ours and fails. The digests covered id, category and value. Compaction rewrites the whole row, so a defect confined to ts, or to a Lance vector, changed nothing either side of maintenance. Every persisted column goes in now, ts as microseconds so no timezone sits between the two runs. The Lance drop check caught every exception as proof the dataset was gone. pylance turns credential and transport failures into the same ValueError, so it only accepts the message that means not found. * docs: say up front which maintenance path the Lance half takes The opening summary said the worker maintains both tables. It maintains the Iceberg one always and the Lance one only where cargo is installed, which is not what CI does. --- .github/workflows/s3-tables-tests.yml | 69 ++++ .../crates/lance/tests/common/mod.rs | 97 +++++ .../crates/lance/tests/compaction.rs | 76 +--- .../crates/lance/tests/lifecycle.rs | 123 ++++++ test/s3tables/lifecycle/Dockerfile.lance | 17 + test/s3tables/lifecycle/Dockerfile.pyiceberg | 10 + test/s3tables/lifecycle/README.md | 77 ++++ test/s3tables/lifecycle/cluster_test.go | 277 +++++++++++++ test/s3tables/lifecycle/iceberg_lifecycle.py | 163 ++++++++ .../lifecycle/iceberg_lifecycle_test.go | 389 ++++++++++++++++++ test/s3tables/lifecycle/lance_lifecycle.py | 168 ++++++++ .../lifecycle/lance_lifecycle_test.go | 130 ++++++ 12 files changed, 1525 insertions(+), 71 deletions(-) create mode 100644 seaweed-worker/crates/lance/tests/common/mod.rs create mode 100644 seaweed-worker/crates/lance/tests/lifecycle.rs create mode 100644 test/s3tables/lifecycle/Dockerfile.lance create mode 100644 test/s3tables/lifecycle/Dockerfile.pyiceberg create mode 100644 test/s3tables/lifecycle/README.md create mode 100644 test/s3tables/lifecycle/cluster_test.go create mode 100644 test/s3tables/lifecycle/iceberg_lifecycle.py create mode 100644 test/s3tables/lifecycle/iceberg_lifecycle_test.go create mode 100644 test/s3tables/lifecycle/lance_lifecycle.py create mode 100644 test/s3tables/lifecycle/lance_lifecycle_test.go 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) +}