diff --git a/seaweed-worker/Cargo.lock b/seaweed-worker/Cargo.lock index de890b6a1..b9dec6abe 100644 --- a/seaweed-worker/Cargo.lock +++ b/seaweed-worker/Cargo.lock @@ -5451,6 +5451,14 @@ dependencies = [ "tracing", ] +[[package]] +name = "seaweed-worker-sort" +version = "0.1.0" +dependencies = [ + "anyhow", + "seaweed-worker-core", +] + [[package]] name = "security-framework" version = "3.7.0" @@ -6542,11 +6550,14 @@ dependencies = [ "clap", "futures", "lance", + "lance-datafusion", "lance-index", "lance-linalg", + "lance-table", "prometheus", "reqwest 0.12.28", "seaweed-worker-core", + "seaweed-worker-sort", "serde", "serde_json", "tokio", diff --git a/seaweed-worker/Cargo.toml b/seaweed-worker/Cargo.toml index b44e9883f..c3064d660 100644 --- a/seaweed-worker/Cargo.toml +++ b/seaweed-worker/Cargo.toml @@ -2,10 +2,11 @@ # # `core` is the plugin.proto contract and nothing else; a worker crate beside it # supplies job handlers and a binary. Adding a worker means adding a member here, -# not touching the protocol. +# not touching the protocol. `sort` is neither: it is the sort specification the +# sorting jobs share, so that two of them cannot drift on what an order means. [workspace] resolver = "2" -members = ["crates/core", "crates/lance"] +members = ["crates/core", "crates/lance", "crates/sort"] [workspace.package] version = "0.1.0" diff --git a/seaweed-worker/crates/core/src/config_form.rs b/seaweed-worker/crates/core/src/config_form.rs index 09106f679..02235cd66 100644 --- a/seaweed-worker/crates/core/src/config_form.rs +++ b/seaweed-worker/crates/core/src/config_form.rs @@ -64,6 +64,17 @@ pub fn number_field(name: &str, label: &str, description: &str, min: i64, max: i } } +pub fn text_field(name: &str, label: &str, description: &str, placeholder: &str) -> ConfigField { + ConfigField { + name: name.to_string(), + label: label.to_string(), + description: description.to_string(), + field_type: ConfigFieldType::String as i32, + placeholder: placeholder.to_string(), + ..Default::default() + } +} + pub fn bool_field(name: &str, label: &str, description: &str) -> ConfigField { ConfigField { name: name.to_string(), diff --git a/seaweed-worker/crates/lance/Cargo.toml b/seaweed-worker/crates/lance/Cargo.toml index e431e7c96..4caa80bc4 100644 --- a/seaweed-worker/crates/lance/Cargo.toml +++ b/seaweed-worker/crates/lance/Cargo.toml @@ -16,11 +16,17 @@ path = "src/main.rs" [dependencies] seaweed-worker-core = { path = "../core" } +seaweed-worker-sort = { path = "../sort" } prometheus.workspace = true # Only the S3 backend: the other object stores lance enables by default are # build time this worker never spends. lance = { version = "10", default-features = false, features = ["aws"] } lance-index = "10" +# For the manifest's own fragment type, which the sort marker summarises. +lance-table = "10" +# For the execution options the sort needs; the scanner's own stream helper +# takes the default, which has spilling off. +lance-datafusion = "10" arrow-schema = "58" arrow-cast = "58" chrono = "0.4" diff --git a/seaweed-worker/crates/lance/src/jobs/mod.rs b/seaweed-worker/crates/lance/src/jobs/mod.rs index 08cbdaa7e..064a4af7d 100644 --- a/seaweed-worker/crates/lance/src/jobs/mod.rs +++ b/seaweed-worker/crates/lance/src/jobs/mod.rs @@ -9,6 +9,7 @@ pub mod cleanup; pub mod compact; pub mod indices; +pub mod sort; use std::collections::HashMap; use std::sync::Arc; @@ -54,6 +55,11 @@ pub fn handlers( .with_fallback(fallback.clone()) .with_metrics(metrics.clone()), ), + Arc::new( + sort::SortHandler::new(namespace_url.clone()) + .with_fallback(fallback.clone()) + .with_metrics(metrics.clone()), + ), Arc::new( cleanup::CleanupVersionsHandler::new(namespace_url) .with_fallback(fallback) diff --git a/seaweed-worker/crates/lance/src/jobs/sort.rs b/seaweed-worker/crates/lance/src/jobs/sort.rs new file mode 100644 index 000000000..4273b7ff2 --- /dev/null +++ b/seaweed-worker/crates/lance/src/jobs/sort.rs @@ -0,0 +1,488 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use anyhow::{anyhow, Context, Result}; +use async_trait::async_trait; +use lance::dataset::scanner::ColumnOrdering; +use lance::dataset::transaction::Operation; +use lance::dataset::write::{CommitBuilder, InsertBuilder}; +use lance::dataset::{WriteDestination, WriteMode, WriteParams}; +use lance::index::DatasetIndexExt; +use lance_datafusion::exec::{execute_plan, LanceExecutionOptions}; +use seaweed_worker_core::config_form::{form, int_or, int_value, string_or, string_value}; +use seaweed_worker_core::pb::{ + ConfigValue, DetectionComplete, DetectionProposals, ExecuteJobRequest, JobCompleted, + JobProgressUpdate, JobProposal, JobResult, JobTypeCapability, JobTypeDescriptor, + RunDetectionRequest, WorkerObservations, +}; +use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler}; +use seaweed_worker_sort::{ + config_fields, resolve, verdict, FragmentSummary, SortSpec, SortState, + CONFIG_MAX_ROWS_PER_FILE, CONFIG_MEMORY_BUDGET_MB, CONFIG_MIN_UNSORTED_ROWS, + CONFIG_SORT_FIELDS, DECLARED_FIELDS_KEY, +}; +use tracing::warn; + +use crate::catalog::{parse_id, NamespaceClient}; +use crate::dataset; +use crate::jobs::{clamp, observation, string_list, table_id, FORMAT}; + +pub const JOB_TYPE: &str = "lance_sort"; + +const DEFAULT_MIN_UNSORTED_ROWS: i64 = 1_048_576; +const DEFAULT_MEMORY_BUDGET_MB: i64 = 512; +const DEFAULT_MAX_ROWS_PER_FILE: i64 = 1_048_576; + +// The ranges the shared form offers, repeated here because a value from outside +// one is a value the UI could not have produced and every one of these is cast +// to an unsigned type on the way in. +const MIN_UNSORTED_ROWS_FLOOR: i64 = 1; +const MIN_UNSORTED_ROWS_CEILING: i64 = 1_000_000_000; +const MEMORY_BUDGET_FLOOR: i64 = 64; +const MEMORY_BUDGET_CEILING: i64 = 1_048_576; +const MAX_ROWS_PER_FILE_FLOOR: i64 = 1024; +const MAX_ROWS_PER_FILE_CEILING: i64 = 16_777_216; + +const BYTES_PER_MB: u64 = 1024 * 1024; + +/// Rewrites a table so its rows are stored in the order its fields declare, +/// which is what lets a range scan read a few files instead of all of them. +/// +/// Lance appends fragments in write order and has no notion of a sorted table, +/// so nothing but a rewrite establishes that order, and nothing but another +/// rewrite restores it once rows have been appended. +pub struct SortHandler { + namespace_url: String, + fallback: dataset::FallbackOptions, + metrics: Option, +} + +impl SortHandler { + pub fn new(namespace_url: String) -> Self { + Self { + namespace_url, + fallback: dataset::FallbackOptions::new(), + metrics: None, + } + } + + pub fn with_metrics(mut self, metrics: Option) -> Self { + self.metrics = metrics; + self + } + + /// Storage options to use where the namespace vends none. + pub fn with_fallback(mut self, fallback: dataset::FallbackOptions) -> Self { + self.fallback = fallback; + self + } + + fn client(&self) -> NamespaceClient { + NamespaceClient::new(self.namespace_url.clone()) + } +} + +/// The fragments a dataset holds, in fragment order, as the sort crate reads +/// them: the data files each one names, and the live rows it holds. +/// +/// Appends only add fragments after the existing ones and lance hands out +/// fragment ids monotonically even across an overwrite, so ordering by id makes +/// "the fragments the sort wrote are still the first ones" a prefix test. +fn fragment_summaries(dataset: &lance::Dataset) -> Vec { + let mut fragments = dataset.get_fragments(); + fragments.sort_by_key(|fragment| fragment.id()); + fragments + .iter() + .map(|fragment| summarize(fragment.metadata())) + .collect() +} + +/// The manifest's own account of one fragment. `num_rows` is physical rows less +/// deletions, and None where the manifest does not record them. +fn summarize(fragment: &lance_table::format::Fragment) -> FragmentSummary { + FragmentSummary { + files: fragment + .files + .iter() + .map(|file| file.path.clone()) + .collect(), + rows: fragment.num_rows().map(|rows| rows as u64), + } +} + +/// The scan ordering for a spec. Lance validates the column names against the +/// schema when the ordering is set, so a spec naming a column the table does +/// not have fails before anything is written. +fn orderings(spec: &SortSpec) -> Vec { + spec.fields + .iter() + .map(|field| ColumnOrdering { + ascending: !field.descending, + nulls_first: field.nulls_first, + column_name: field.path.clone(), + }) + .collect() +} + +#[async_trait] +impl JobHandler for SortHandler { + fn capability(&self) -> JobTypeCapability { + JobTypeCapability { + job_type: JOB_TYPE.to_string(), + can_detect: true, + can_execute: true, + max_detection_concurrency: 1, + max_execution_concurrency: 1, + display_name: "Lance Sort".to_string(), + description: "Rewrite Lance tables in the order their fields declare. The rewrite replaces every fragment, so indices do not survive it and want rebuilding afterwards.".to_string(), + weight: 40, + } + } + + fn descriptor(&self) -> JobTypeDescriptor { + let mut defaults: HashMap = HashMap::new(); + defaults.insert(CONFIG_SORT_FIELDS.to_string(), string_value("")); + defaults.insert( + CONFIG_MIN_UNSORTED_ROWS.to_string(), + int_value(DEFAULT_MIN_UNSORTED_ROWS), + ); + defaults.insert( + CONFIG_MEMORY_BUDGET_MB.to_string(), + int_value(DEFAULT_MEMORY_BUDGET_MB), + ); + defaults.insert( + CONFIG_MAX_ROWS_PER_FILE.to_string(), + int_value(DEFAULT_MAX_ROWS_PER_FILE), + ); + + JobTypeDescriptor { + job_type: JOB_TYPE.to_string(), + display_name: "Lance Sort".to_string(), + description: "Sort Lance tables by their declared fields".to_string(), + icon: "fas fa-sort".to_string(), + descriptor_version: 1, + worker_config_form: Some(form( + "lance-sort-worker", + "Sort", + config_fields(), + defaults.clone(), + )), + worker_default_values: defaults, + ..Default::default() + } + } + + /// Propose a job for every table whose declared order the data no longer + /// follows. Opening a dataset reads its manifest and its configuration, not + /// its data, so this stays cheap across a catalog. + async fn detect( + &self, + request: &RunDetectionRequest, + sender: &dyn DetectionSender, + ) -> Result<()> { + let configured = string_or(&request.worker_config_values, CONFIG_SORT_FIELDS, ""); + let min_unsorted_rows = clamp( + int_or( + &request.worker_config_values, + CONFIG_MIN_UNSORTED_ROWS, + DEFAULT_MIN_UNSORTED_ROWS, + ), + MIN_UNSORTED_ROWS_FLOOR, + MIN_UNSORTED_ROWS_CEILING, + ) as u64; + + let client = self.client(); + let tables = client.list_all_tables().await?; + + let mut proposals = Vec::new(); + let mut observations = Vec::new(); + for encoded in &tables { + let id = parse_id(encoded); + let table = match dataset::open(&client, &id, &self.fallback).await { + Ok(table) => table, + Err(err) => { + // One unreadable table must not end the sweep: the tables + // already read would lose their proposals. + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "open"); + } + warn!("skipping {encoded}: {err:#}"); + continue; + } + }; + if let Some(counters) = &self.metrics { + counters.worker.object_seen(JOB_TYPE); + } + let stats = match table.stats().await { + Ok(stats) => stats, + Err(err) => { + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "stats"); + } + warn!("skipping {encoded}: reading its stats failed: {err:#}"); + continue; + } + }; + + let config = table.dataset.config().clone(); + let spec = match resolve( + config.get(DECLARED_FIELDS_KEY).map(String::as_str), + &configured, + ) { + Ok(Some(spec)) => spec, + Ok(None) => { + // Not every table is one an operator wants sorted, and a + // table with no order named anywhere is not a failure. + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "no_sort_order"); + } + continue; + } + Err(err) => { + if let Some(counters) = &self.metrics { + counters.worker.object_skipped(JOB_TYPE, "sort_order"); + } + warn!("skipping {encoded}: {err:#}"); + continue; + } + }; + + let state = SortState::from_config(&config); + let decision = verdict( + &spec, + &state, + &fragment_summaries(&table.dataset), + min_unsorted_rows, + ); + + let mut attributes: HashMap = HashMap::new(); + attributes.insert("rows".to_string(), int_value(stats.rows as i64)); + attributes.insert("version".to_string(), int_value(stats.version as i64)); + attributes.insert("fragments".to_string(), int_value(stats.fragments as i64)); + attributes.insert("sort_fields".to_string(), string_value(spec.to_string())); + attributes.insert("sort_state".to_string(), string_value(decision.reason())); + if let Some(schema) = stats.schema.clone() { + attributes.insert("schema".to_string(), string_value(schema)); + } + observations.push(observation(&id, FORMAT, attributes)); + + // Logged because "detection proposed nothing" is otherwise + // indistinguishable from a table the worker could not read. + tracing::info!( + "sort detection: {encoded} has {} rows in {spec} order: {}", + stats.rows, + decision.reason() + ); + + if !decision.needs_sort() { + continue; + } + let mut parameters: HashMap = HashMap::new(); + parameters.insert("table_id".to_string(), string_list(&id)); + proposals.push(JobProposal { + proposal_id: format!("{JOB_TYPE}:{encoded}"), + dedupe_key: format!("{JOB_TYPE}:{encoded}"), + job_type: JOB_TYPE.to_string(), + summary: format!("Sort {encoded} by {spec}"), + detail: format!("{} rows, {}", stats.rows, decision.reason()), + parameters, + ..Default::default() + }); + } + + if !observations.is_empty() { + sender.send_observations(WorkerObservations { + job_type: JOB_TYPE.to_string(), + observations, + })?; + } + + let total = proposals.len() as i32; + sender.send_proposals(DetectionProposals { + request_id: request.request_id.clone(), + job_type: JOB_TYPE.to_string(), + proposals, + has_more: false, + })?; + sender.send_complete(DetectionComplete { + request_id: request.request_id.clone(), + job_type: JOB_TYPE.to_string(), + success: true, + error_message: String::new(), + total_proposals: total, + })?; + Ok(()) + } + + async fn execute( + &self, + request: &ExecuteJobRequest, + sender: &dyn ExecutionSender, + ) -> Result<()> { + let job = request + .job + .as_ref() + .ok_or_else(|| anyhow!("execute request carried no job"))?; + let id = table_id(&job.parameters) + .ok_or_else(|| anyhow!("job {} carried no table_id", job.job_id))?; + + let configured = string_or(&request.worker_config_values, CONFIG_SORT_FIELDS, ""); + let memory_budget_mb = clamp( + int_or( + &request.worker_config_values, + CONFIG_MEMORY_BUDGET_MB, + DEFAULT_MEMORY_BUDGET_MB, + ), + MEMORY_BUDGET_FLOOR, + MEMORY_BUDGET_CEILING, + ) as u64; + let max_rows_per_file = clamp( + int_or( + &request.worker_config_values, + CONFIG_MAX_ROWS_PER_FILE, + DEFAULT_MAX_ROWS_PER_FILE, + ), + MAX_ROWS_PER_FILE_FLOOR, + MAX_ROWS_PER_FILE_CEILING, + ) as usize; + + let client = self.client(); + // Re-resolve rather than trusting what detection saw: the table may have + // been repointed, its declared order changed, and the vended credentials + // have expired. + let table = dataset::open(&client, &id, &self.fallback).await?; + let before = table.stats().await?; + let config = table.dataset.config().clone(); + let spec = resolve( + config.get(DECLARED_FIELDS_KEY).map(String::as_str), + &configured, + )? + .ok_or_else(|| anyhow!("neither the table nor this worker names an order to sort by"))?; + + sender.send_progress(JobProgressUpdate { + request_id: request.request_id.clone(), + job_id: job.job_id.clone(), + job_type: JOB_TYPE.to_string(), + progress_percent: 10.0, + stage: format!("sorting {} rows by {spec}", before.rows), + ..Default::default() + })?; + + let mut scanner = table.dataset.scan(); + scanner + .order_by(Some(orderings(&spec))) + .with_context(|| format!("order the scan of {} by {spec}", table.location))?; + let plan = scanner + .create_plan() + .await + .context("plan the ordered scan")?; + + // Spilling is off in LanceExecutionOptions::default(), and the scanner's + // own stream helper takes that default, so the options are built here + // instead: a sort that cannot spill is one bounded by memory, which is + // the whole thing this job exists to avoid. + let stream = execute_plan( + plan, + LanceExecutionOptions { + use_spilling: true, + mem_pool_size: Some(memory_budget_mb * BYTES_PER_MB), + ..Default::default() + }, + ) + .context("run the ordered scan")?; + + let params = WriteParams { + mode: WriteMode::Overwrite, + max_rows_per_file, + ..Default::default() + }; + let destination = Arc::new(table.dataset.clone()); + let mut transaction = InsertBuilder::new(WriteDestination::Dataset(destination.clone())) + .with_params(¶ms) + .execute_uncommitted_stream(stream) + .await + .context("write the sorted fragments")?; + + // The marker rides in the same commit as the data it describes: + // Operation::Overwrite is the one place lance takes configuration + // values alongside fragments, and a marker written as a second commit + // would be lost if the worker died in between, re-sorting the whole + // table on the next sweep. + match &mut transaction.operation { + Operation::Overwrite { + config_upsert_values, + fragments, + .. + } => { + // The files these fragments name were written a moment ago and + // keep their names whatever version this commit becomes, which + // is what lets the marker travel inside the commit it describes. + let written: Vec = fragments.iter().map(summarize).collect(); + *config_upsert_values = + Some(SortState::record(&spec, &written, before.rows as u64)); + } + other => { + return Err(anyhow!( + "a sorted rewrite produced {other} instead of an overwrite" + )) + } + } + + let sorted = CommitBuilder::new(WriteDestination::Dataset(destination)) + .execute(transaction) + .await + .context("commit the sorted rewrite")?; + + if let Some(counters) = &self.metrics { + counters.rows_sorted.inc_by(before.rows as u64); + } + + // An overwrite carries fragments, schema and configuration — not index + // metadata — so a table that had indices no longer does. Saying so is + // the difference between a slow table and a mystery. + let dropped_indices = match table.dataset.load_indices().await { + Ok(indices) => indices.len(), + Err(err) => { + warn!("could not read the indices of {}: {err:#}", table.location); + 0 + } + }; + if dropped_indices > 0 { + warn!( + "{} lost {dropped_indices} index/indices to the sorted rewrite; run the index job to rebuild them", + table.location + ); + } + + let fragments_after = sorted.get_fragments().len(); + let mut output: HashMap = HashMap::new(); + output.insert("rows_sorted".to_string(), int_value(before.rows as i64)); + output.insert("sort_fields".to_string(), string_value(spec.to_string())); + output.insert( + "fragments_before".to_string(), + int_value(before.fragments as i64), + ); + output.insert( + "fragments_after".to_string(), + int_value(fragments_after as i64), + ); + output.insert( + "indices_dropped".to_string(), + int_value(dropped_indices as i64), + ); + + sender.send_completed(JobCompleted { + request_id: request.request_id.clone(), + job_id: job.job_id.clone(), + job_type: JOB_TYPE.to_string(), + success: true, + result: Some(JobResult { + output_values: output, + summary: format!("{} rows rewritten in {spec} order", before.rows), + ..Default::default() + }), + ..Default::default() + })?; + Ok(()) + } +} diff --git a/seaweed-worker/crates/lance/src/main.rs b/seaweed-worker/crates/lance/src/main.rs index cd7015a92..a87a4b13d 100644 --- a/seaweed-worker/crates/lance/src/main.rs +++ b/seaweed-worker/crates/lance/src/main.rs @@ -12,10 +12,7 @@ use weed_lance_worker::metrics::LanceMetrics; /// Mirrors `weed worker`'s flags, because this is the same contract from another /// language and an operator should not have to learn a second set of names. #[derive(Parser, Debug)] -#[command( - name = "weed-worker", - about = "SeaweedFS maintenance worker" -)] +#[command(name = "weed-worker", about = "SeaweedFS maintenance worker")] struct Args { /// Admin server gRPC address. #[arg(long, default_value = "localhost:23646", env = "WEED_ADMIN")] diff --git a/seaweed-worker/crates/lance/src/metrics.rs b/seaweed-worker/crates/lance/src/metrics.rs index 4eab963fb..e414a4538 100644 --- a/seaweed-worker/crates/lance/src/metrics.rs +++ b/seaweed-worker/crates/lance/src/metrics.rs @@ -18,6 +18,7 @@ pub struct LanceMetrics { pub worker: Metrics, pub fragments_removed: IntCounter, pub rows_indexed: IntCounter, + pub rows_sorted: IntCounter, pub versions_removed: IntCounter, pub bytes_reclaimed: IntCounter, } @@ -34,6 +35,10 @@ impl LanceMetrics { "lance_rows_indexed_total", "Rows brought under an index that did not cover them.", )?, + rows_sorted: metrics.counter( + "lance_rows_sorted_total", + "Rows rewritten into their table's declared order.", + )?, versions_removed: metrics.counter( "lance_versions_removed_total", "Dataset versions removed by cleanup.", diff --git a/seaweed-worker/crates/lance/tests/sort.rs b/seaweed-worker/crates/lance/tests/sort.rs new file mode 100644 index 000000000..22fe28ef9 --- /dev/null +++ b/seaweed-worker/crates/lance/tests/sort.rs @@ -0,0 +1,504 @@ +//! Drives the sort handler against a live namespace. +//! +//! Skipped unless WEED_LANCE_NAMESPACE names one, like the other integration +//! tests here: the job rewrites every fragment of a real table and commits, and +//! the commit is the half worth testing. + +use anyhow::Result; +use seaweed_worker_core::pb::{ + config_value::Kind, ConfigValue, ExecuteJobRequest, JobSpec, RunDetectionRequest, +}; +use seaweed_worker_core::JobHandler; +use weed_lance_worker::catalog::NamespaceClient; +use weed_lance_worker::jobs::sort::{SortHandler, JOB_TYPE}; + +mod common; +use common::{fallback, namespace_url, Recorder}; + +/// One live gateway and one shared catalog, and `list_all_tables` sweeps +/// everything, so these tests take a lock the way the compaction ones do. +static GATEWAY: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +fn config(values: &[(&str, ConfigValue)]) -> std::collections::HashMap { + values + .iter() + .map(|(name, value)| (name.to_string(), value.clone())) + .collect() +} + +fn int(value: i64) -> ConfigValue { + ConfigValue { + kind: Some(Kind::Int64Value(value)), + } +} + +fn text(value: &str) -> ConfigValue { + ConfigValue { + kind: Some(Kind::StringValue(value.to_string())), + } +} + +/// Declares a table and writes `batches` appends whose ids descend, so the rows +/// land in an order no sort would produce. `declared` is written into the +/// dataset's own configuration, which is where a table says how it wants to be +/// sorted. +async fn seed_unsorted_table( + url: &str, + name: &str, + batches: i64, + rows_each: i64, + declared: Option<&str>, +) -> Result { + use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator}; + use arrow_schema::{DataType, Field, Schema}; + use lance::dataset::{Dataset, WriteMode, WriteParams}; + use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; + use std::sync::Arc; + + let http = reqwest::Client::new(); + for parent in ["vec", "vec$ml"] { + http.post(format!("{url}/v1/namespace/{parent}/create")) + .json(&serde_json::json!({"mode": "EXIST_OK"})) + .send() + .await? + .error_for_status()?; + } + let encoded = format!("vec$ml${name}"); + http.post(format!("{url}/v1/table/{encoded}/declare")) + .json(&serde_json::json!({})) + .send() + .await? + .error_for_status()?; + + let client = NamespaceClient::new(url.to_string()); + let id = vec!["vec".to_string(), "ml".to_string(), name.to_string()]; + let description = client.describe_table(&id).await?; + let mut options = description.storage_options.clone(); + options.extend(fallback()); + + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])); + let total = batches * rows_each; + let mut dataset = None; + for batch_index in 0..batches { + // Ids descend across and within the batches, so the physical order is + // the reverse of the order the table declares. + let ids: Vec = (0..rows_each) + .map(|row| total - 1 - (batch_index * rows_each + row)) + .collect(); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(Int64Array::from(ids))])?; + let params = WriteParams { + mode: if batch_index == 0 { + WriteMode::Overwrite + } else { + WriteMode::Append + }, + store_params: Some(ObjectStoreParams { + storage_options_accessor: Some(Arc::new( + StorageOptionsAccessor::with_static_options(options.clone()), + )), + ..Default::default() + }), + ..Default::default() + }; + dataset = Some( + Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)], schema.clone()), + description.location.as_str(), + Some(params), + ) + .await?, + ); + } + + if let Some(declared) = declared { + let mut dataset = dataset.expect("at least one batch was written"); + dataset + .update_config([("seaweedfs.sort.fields", declared)]) + .await?; + } + Ok(encoded) +} + +/// Reads the ids back in the order they are stored, which is what a sort is +/// supposed to change. +async fn stored_ids(url: &str, name: &str) -> Result> { + use arrow_array::Int64Array; + use futures::TryStreamExt; + + let client = NamespaceClient::new(url.to_string()); + let id = vec!["vec".to_string(), "ml".to_string(), name.to_string()]; + let table = weed_lance_worker::dataset::open(&client, &id, &fallback()).await?; + let stream = table.dataset.scan().try_into_stream().await?; + let batches: Vec<_> = stream.try_collect().await?; + let mut ids = Vec::new(); + for batch in batches { + let column = batch + .column_by_name("id") + .expect("the seeded table has an id column") + .as_any() + .downcast_ref::() + .expect("id is an Int64 column") + .clone(); + ids.extend(column.values().iter().copied()); + } + Ok(ids) +} + +/// The whole loop: a table declaring an order its rows do not follow is +/// proposed, sorting it stores the rows in that order, and the marker the same +/// commit carries is what stops the next sweep proposing it again. +#[tokio::test] +async fn sorts_a_table_and_stops_proposing_it() { + let _gateway = GATEWAY.lock().await; + let Some(url) = namespace_url() else { + eprintln!("WEED_LANCE_NAMESPACE is unset, skipping"); + return; + }; + let encoded = seed_unsorted_table(&url, "sortme", 4, 250, Some("id asc")) + .await + .expect("seed an unsorted table"); + + let before = stored_ids(&url, "sortme").await.expect("read the ids back"); + assert!( + before.windows(2).any(|pair| pair[0] > pair[1]), + "the seeded table was already sorted, so the test proves nothing" + ); + + let handler = SortHandler::new(url.clone()).with_fallback(fallback()); + let recorder = Recorder::default(); + let detection = RunDetectionRequest { + request_id: "detect-sort".to_string(), + job_type: JOB_TYPE.to_string(), + worker_config_values: config(&[("min_unsorted_rows", int(1))]), + ..Default::default() + }; + handler + .detect(&detection, &recorder) + .await + .expect("detection failed"); + + let proposal = recorder + .proposals + .lock() + .unwrap() + .iter() + .find(|p| p.summary.contains(encoded.as_str())) + .cloned() + .expect("the unsorted table was not proposed"); + assert!( + proposal.summary.contains("id asc"), + "the proposal does not say what order it would use: {}", + proposal.summary + ); + + let execute = ExecuteJobRequest { + request_id: "execute-sort".to_string(), + job: Some(JobSpec { + job_id: "job-sort".to_string(), + job_type: JOB_TYPE.to_string(), + parameters: proposal.parameters.clone(), + ..Default::default() + }), + worker_config_values: config(&[ + ("memory_budget_mb", int(64)), + ("max_rows_per_file", int(1024)), + ]), + ..Default::default() + }; + handler + .execute(&execute, &recorder) + .await + .expect("sorting failed"); + + let completed = recorder.completed.lock().unwrap().clone(); + let result = completed.last().expect("no completion reported"); + assert!( + result.success, + "the sort reported failure: {}", + result.error_message + ); + + let after = stored_ids(&url, "sortme") + .await + .expect("read the sorted ids back"); + assert_eq!(after.len(), before.len(), "the sort lost or invented rows"); + assert!( + after.windows(2).all(|pair| pair[0] <= pair[1]), + "the rows are not stored in ascending id order" + ); + + // The marker rode in the same commit as the data, so a second sweep finds + // the table up to date without anything else having written it. + let second = Recorder::default(); + handler + .detect(&detection, &second) + .await + .expect("second detection failed"); + assert!( + !second + .proposals + .lock() + .unwrap() + .iter() + .any(|p| p.summary.contains(encoded.as_str())), + "a table that was just sorted must not be proposed again" + ); + eprintln!("sort result: {}", result.result.as_ref().unwrap().summary); +} + +/// A table that declares no order, with a worker configuring none either, is +/// not something to rewrite on a guess. +#[tokio::test] +async fn leaves_a_table_that_declares_no_order_alone() { + let _gateway = GATEWAY.lock().await; + let Some(url) = namespace_url() else { + eprintln!("WEED_LANCE_NAMESPACE is unset, skipping"); + return; + }; + let encoded = seed_unsorted_table(&url, "unsorted", 2, 16, None) + .await + .expect("seed a table with no declared order"); + + let handler = SortHandler::new(url.clone()).with_fallback(fallback()); + let recorder = Recorder::default(); + handler + .detect( + &RunDetectionRequest { + request_id: "detect-none".to_string(), + job_type: JOB_TYPE.to_string(), + worker_config_values: config(&[ + ("min_unsorted_rows", int(1)), + ("sort_fields", text("")), + ]), + ..Default::default() + }, + &recorder, + ) + .await + .expect("detection failed"); + + assert!( + !recorder + .proposals + .lock() + .unwrap() + .iter() + .any(|p| p.summary.contains(encoded.as_str())), + "a table with no order named anywhere must not be proposed" + ); +} + +/// The case a row-count threshold cannot see: the table replaced by a write of +/// its own, growing by far less than `min_unsorted_rows`. Every row moved, so +/// the table is unsorted, and detection has to say so however small the delta. +#[tokio::test] +async fn proposes_a_replacement_that_barely_changes_the_row_count() { + let _gateway = GATEWAY.lock().await; + let Some(url) = namespace_url() else { + eprintln!("WEED_LANCE_NAMESPACE is unset, skipping"); + return; + }; + let encoded = seed_unsorted_table(&url, "replaceme", 4, 250, Some("id asc")) + .await + .expect("seed an unsorted table"); + + let handler = SortHandler::new(url.clone()).with_fallback(fallback()); + // A threshold no realistic append would cross, so nothing but the identity + // of the data can be what proposes this table. + let detection = RunDetectionRequest { + request_id: "detect-replace".to_string(), + job_type: JOB_TYPE.to_string(), + worker_config_values: config(&[("min_unsorted_rows", int(1_000_000))]), + ..Default::default() + }; + + let first = Recorder::default(); + handler.detect(&detection, &first).await.expect("detection"); + let proposal = first + .proposals + .lock() + .unwrap() + .iter() + .find(|p| p.summary.contains(encoded.as_str())) + .cloned() + .expect("a never-sorted table was not proposed"); + + handler + .execute( + &ExecuteJobRequest { + request_id: "execute-replace".to_string(), + job: Some(JobSpec { + job_id: "job-replace".to_string(), + job_type: JOB_TYPE.to_string(), + parameters: proposal.parameters.clone(), + ..Default::default() + }), + worker_config_values: config(&[("max_rows_per_file", int(1024))]), + ..Default::default() + }, + &first, + ) + .await + .expect("sorting failed"); + + // Sorted, and left alone by the next sweep. + let after_sort = Recorder::default(); + handler + .detect(&detection, &after_sort) + .await + .expect("detection"); + assert!( + !after_sort + .proposals + .lock() + .unwrap() + .iter() + .any(|p| p.summary.contains(encoded.as_str())), + "a table that was just sorted must not be proposed again" + ); + + // Now replace the data: 1002 rows where there were 1000, shuffled again. + seed_unsorted_table(&url, "replaceme", 3, 334, Some("id asc")) + .await + .expect("replace the table's data"); + + let after_replace = Recorder::default(); + handler + .detect(&detection, &after_replace) + .await + .expect("detection"); + assert!( + after_replace + .proposals + .lock() + .unwrap() + .iter() + .any(|p| p.summary.contains(encoded.as_str())), + "a replaced table must be proposed even when the row count barely moved" + ); +} + +/// Deletions must not hide appended rows. This is the arithmetic against a real +/// manifest rather than a hand-built one: the sorted fragments keep their files +/// and lose most of their rows, a smaller batch is appended after them, and the +/// table has to be proposed *because rows were appended* — not because it looks +/// replaced, and not left alone because it now holds fewer rows than when it +/// was sorted. +#[tokio::test] +async fn deletions_do_not_hide_appended_rows() { + let _gateway = GATEWAY.lock().await; + let Some(url) = namespace_url() else { + eprintln!("WEED_LANCE_NAMESPACE is unset, skipping"); + return; + }; + let encoded = seed_unsorted_table(&url, "deleteme", 4, 250, Some("id asc")) + .await + .expect("seed an unsorted table"); + + let handler = SortHandler::new(url.clone()).with_fallback(fallback()); + let detection = RunDetectionRequest { + request_id: "detect-delete".to_string(), + job_type: JOB_TYPE.to_string(), + worker_config_values: config(&[("min_unsorted_rows", int(100))]), + ..Default::default() + }; + + let recorder = Recorder::default(); + handler + .detect(&detection, &recorder) + .await + .expect("detection"); + let proposal = recorder + .proposals + .lock() + .unwrap() + .iter() + .find(|p| p.summary.contains(encoded.as_str())) + .cloned() + .expect("a never-sorted table was not proposed"); + handler + .execute( + &ExecuteJobRequest { + request_id: "execute-delete".to_string(), + job: Some(JobSpec { + job_id: "job-delete".to_string(), + job_type: JOB_TYPE.to_string(), + parameters: proposal.parameters.clone(), + ..Default::default() + }), + worker_config_values: config(&[("max_rows_per_file", int(1024))]), + ..Default::default() + }, + &recorder, + ) + .await + .expect("sorting failed"); + + // Delete 900 of the 1000 sorted rows, then append 200 unsorted ones. The + // table is smaller than when it was sorted, so net growth reads as zero. + let client = NamespaceClient::new(url.clone()); + let id = vec!["vec".to_string(), "ml".to_string(), "deleteme".to_string()]; + let mut table = weed_lance_worker::dataset::open(&client, &id, &fallback()) + .await + .expect("reopen the sorted table"); + table + .dataset + .delete("id < 900") + .await + .expect("delete most of the sorted rows"); + append_ids(&url, "deleteme", 5_000, 200) + .await + .expect("append rows after the sorted ones"); + + let after = Recorder::default(); + handler.detect(&detection, &after).await.expect("detection"); + let proposal = after + .proposals + .lock() + .unwrap() + .iter() + .find(|p| p.summary.contains(encoded.as_str())) + .cloned() + .expect("a table with 200 unsorted rows appended was not proposed"); + assert!( + proposal.detail.contains("200 rows appended"), + "expected the appended rows to be counted, got {:?}", + proposal.detail + ); +} + +/// Appends one batch of ascending ids to an existing table. +async fn append_ids(url: &str, name: &str, first_id: i64, rows: i64) -> Result<()> { + use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator}; + use arrow_schema::{DataType, Field, Schema}; + use lance::dataset::{Dataset, WriteMode, WriteParams}; + use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; + use std::sync::Arc; + + let client = NamespaceClient::new(url.to_string()); + let id = vec!["vec".to_string(), "ml".to_string(), name.to_string()]; + let description = client.describe_table(&id).await?; + let mut options = description.storage_options.clone(); + options.extend(fallback()); + + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])); + let ids: Vec = (0..rows).map(|row| first_id + row).collect(); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(Int64Array::from(ids))])?; + let params = WriteParams { + mode: WriteMode::Append, + store_params: Some(ObjectStoreParams { + storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( + options, + ))), + ..Default::default() + }), + ..Default::default() + }; + Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)], schema.clone()), + description.location.as_str(), + Some(params), + ) + .await?; + Ok(()) +} diff --git a/seaweed-worker/crates/sort/Cargo.toml b/seaweed-worker/crates/sort/Cargo.toml new file mode 100644 index 000000000..c24a548c1 --- /dev/null +++ b/seaweed-worker/crates/sort/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "seaweed-worker-sort" +version.workspace = true +edition.workspace = true +description = "The sort specification shared by SeaweedFS sorting jobs" + +[lib] +name = "seaweed_worker_sort" + +[dependencies] +seaweed-worker-core = { path = "../core" } +anyhow.workspace = true diff --git a/seaweed-worker/crates/sort/src/lib.rs b/seaweed-worker/crates/sort/src/lib.rs new file mode 100644 index 000000000..22f4dd357 --- /dev/null +++ b/seaweed-worker/crates/sort/src/lib.rs @@ -0,0 +1,612 @@ +//! The sort specification the sorting jobs share. +//! +//! A worker sorts a table by the fields the table itself declares, and by the +//! fields its own configuration names for a table that declares none. Both +//! halves of that rule, and the spelling of a field list, live here rather than +//! in each job: a Lance job and an Iceberg job that disagreed about what +//! "id desc nulls-first" means would be two features wearing one name. +//! +//! The spelling is deliberately the one `weed/worker/tasks/iceberg` already +//! writes into an Iceberg snapshot's `sort-fields` summary, so the same order +//! reads the same way whichever format holds the table and whichever language +//! sorted it. + +use std::collections::{HashMap, HashSet}; +use std::fmt; + +use anyhow::{bail, Context, Result}; +use seaweed_worker_core::config_form::{number_field, text_field}; +use seaweed_worker_core::pb::ConfigField; + +/// Worker configuration keys. Shared so the Iceberg and Lance forms offer an +/// operator the same names for the same settings. +pub const CONFIG_SORT_FIELDS: &str = "sort_fields"; +pub const CONFIG_MIN_UNSORTED_ROWS: &str = "min_unsorted_rows"; +pub const CONFIG_MEMORY_BUDGET_MB: &str = "memory_budget_mb"; +pub const CONFIG_MAX_ROWS_PER_FILE: &str = "max_rows_per_file"; + +/// The order an operator declares on a table. The worker only ever reads this +/// one. +pub const DECLARED_FIELDS_KEY: &str = "seaweedfs.sort.fields"; + +/// What the worker recorded about the sort it last performed. Kept apart from +/// the declaration above so that "the operator asked for this order" and "the +/// worker achieved this order" stay distinguishable — comparing them is how a +/// changed order is noticed. +pub const SORTED_FIELDS_KEY: &str = "seaweedfs.sort.sorted_fields"; +pub const SORTED_ROWS_KEY: &str = "seaweedfs.sort.sorted_rows"; +/// How many fragments the sort wrote, and a digest of their data file names. +/// Together they identify the data the sort produced — see [`verdict`] for why +/// a version number cannot. +pub const SORTED_FRAGMENTS_KEY: &str = "seaweedfs.sort.sorted_fragments"; +pub const SORTED_DIGEST_KEY: &str = "seaweedfs.sort.sorted_digest"; + +/// One field of a sort order. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SortField { + pub path: String, + pub descending: bool, + pub nulls_first: bool, +} + +/// A whole sort order, in the order the fields are compared. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SortSpec { + pub fields: Vec, +} + +impl SortSpec { + /// Reads "user_id asc nulls-last, ts desc" into a spec. An empty string is + /// not an error: it is how an operator says nothing, and the caller decides + /// what that means. + /// + /// Direction defaults to ascending and null order to Iceberg's default for + /// the direction — nulls first ascending, nulls last descending — because + /// this spelling is shared with tables that already have that rule. + pub fn parse(text: &str) -> Result> { + let mut fields = Vec::new(); + let mut seen = HashSet::new(); + for entry in text.split(',') { + let entry = entry.trim(); + if entry.is_empty() { + continue; + } + let field = + parse_field(entry).with_context(|| format!("read the sort field {entry:?}"))?; + // Compared exactly, not case-folded: Arrow schemas are + // case-sensitive, so `id` and `ID` can be two real columns and + // folding them together would reject a valid order. The Iceberg + // job, whose format resolves names case-insensitively, does its own + // check against the table schema. + if !seen.insert(field.path.clone()) { + bail!("the sort field {:?} is named twice", field.path); + } + fields.push(field); + } + if fields.is_empty() { + return Ok(None); + } + Ok(Some(Self { fields })) + } +} + +/// Renders a spec back, always spelling out direction and null order so that a +/// recorded order round-trips to the same thing it was parsed from. +impl fmt::Display for SortSpec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + for (i, field) in self.fields.iter().enumerate() { + if i > 0 { + write!(f, ", ")?; + } + write!( + f, + "{} {} {}", + field.path, + if field.descending { "desc" } else { "asc" }, + if field.nulls_first { + "nulls-first" + } else { + "nulls-last" + } + )?; + } + Ok(()) + } +} + +fn parse_field(entry: &str) -> Result { + let mut tokens = entry.split_whitespace(); + let path = tokens + .next() + .expect("split_whitespace yields one token for a non-empty entry") + .to_string(); + + let mut descending = None; + let mut nulls_first = None; + for token in tokens { + // Underscores are accepted because that is how the same words are + // spelled in configuration keys, and refusing them would only teach an + // operator that the two spellings are different things. + match token.to_ascii_lowercase().replace('_', "-").as_str() { + "asc" | "ascending" if descending.is_none() => descending = Some(false), + "desc" | "descending" if descending.is_none() => descending = Some(true), + "nulls-first" if nulls_first.is_none() => nulls_first = Some(true), + "nulls-last" if nulls_first.is_none() => nulls_first = Some(false), + other => bail!("{other:?} is not a direction or a null order"), + } + } + + let descending = descending.unwrap_or(false); + Ok(SortField { + path, + descending, + nulls_first: nulls_first.unwrap_or(!descending), + }) +} + +/// The order to sort a table by: what the table declares wins, and the worker's +/// configuration is the fallback for a table that declares nothing. Neither one +/// is not an error — it means this is not a table the operator wants sorted. +/// +/// A declaration that cannot be read is an error rather than a reason to fall +/// back: sorting by the worker's default order instead of the one the table +/// asked for would silently rewrite the table the wrong way. +pub fn resolve(declared: Option<&str>, configured: &str) -> Result> { + if let Some(declared) = declared { + if let Some(spec) = SortSpec::parse(declared).context("read the table's declared order")? { + return Ok(Some(spec)); + } + } + SortSpec::parse(configured).context("read the configured sort order") +} + +/// One fragment as detection sees it. +/// +/// `rows` is the live row count the manifest records — physical rows less the +/// deletions — and is None when the manifest does not say, which older +/// fragments do not. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FragmentSummary { + pub files: Vec, + pub rows: Option, +} + +/// Every data file of these fragments, in order. +pub fn data_files(fragments: &[FragmentSummary]) -> Vec<&str> { + fragments + .iter() + .flat_map(|fragment| fragment.files.iter().map(String::as_str)) + .collect() +} + +/// A digest of data file names, in the order the fragments hold them. +/// +/// FNV-1a rather than the standard library's hasher, whose output is explicitly +/// not stable across releases: a marker that hashed differently after a +/// toolchain upgrade would re-sort every table once, silently. +pub fn digest>(files: &[S]) -> String { + let mut hash: u64 = 0xcbf2_9ce4_8422_2325; + for file in files { + for byte in file.as_ref().as_bytes().iter().chain(std::iter::once(&0u8)) { + hash ^= *byte as u64; + hash = hash.wrapping_mul(0x1000_0000_01b3); + } + } + format!("{hash:016x}") +} + +/// What the worker recorded on a table the last time it sorted it. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct SortState { + pub fields: Option, + pub rows: Option, + /// The fragments the sort wrote: how many, and a digest of their data file + /// names. Data file names are chosen before the commit, which is what makes + /// them usable in a marker the same commit carries. + pub fragments: Option, + pub digest: Option, +} + +impl SortState { + /// Reads the marker out of a table's key/value configuration. A key that + /// will not parse is treated as absent, which asks for a sort rather than + /// skipping the table: the cost of a needless sort is time, and the cost of + /// skipping is a table that silently never gets sorted again. + pub fn from_config(config: &HashMap) -> Self { + Self { + fields: config.get(SORTED_FIELDS_KEY).cloned(), + rows: config.get(SORTED_ROWS_KEY).and_then(|v| v.parse().ok()), + fragments: config + .get(SORTED_FRAGMENTS_KEY) + .and_then(|v| v.parse().ok()), + digest: config.get(SORTED_DIGEST_KEY).cloned(), + } + } + + /// The marker to write for a sort that just finished. + pub fn record( + spec: &SortSpec, + fragments: &[FragmentSummary], + rows: u64, + ) -> HashMap { + HashMap::from([ + (SORTED_FIELDS_KEY.to_string(), spec.to_string()), + (SORTED_ROWS_KEY.to_string(), rows.to_string()), + ( + SORTED_FRAGMENTS_KEY.to_string(), + fragments.len().to_string(), + ), + ( + SORTED_DIGEST_KEY.to_string(), + digest(&data_files(fragments)), + ), + ]) + } +} + +/// Why a table does or does not need sorting. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SortVerdict { + NeverSorted, + OrderChanged, + RowsAppended(u64), + /// The table was committed to since the sort, and the row count did not + /// grow — the data is not the data that was sorted. + Rewritten, + UpToDate, +} + +impl SortVerdict { + pub fn needs_sort(&self) -> bool { + !matches!(self, Self::UpToDate) + } + + pub fn reason(&self) -> String { + match self { + Self::NeverSorted => "never sorted".to_string(), + Self::OrderChanged => "the declared order changed since the last sort".to_string(), + Self::RowsAppended(rows) => format!("{rows} rows appended since the last sort"), + Self::Rewritten => "the data was replaced since the last sort".to_string(), + Self::UpToDate => "sorted".to_string(), + } + } +} + +/// Decides whether a table is worth sorting. +/// +/// The question detection actually has to answer is "is this still the data the +/// sort wrote?", and neither the row count nor the dataset version can answer +/// it. A rewrite that leaves the row count where it was is invisible to the +/// first, and the second cannot be recorded at all: the commit carrying the +/// marker is the commit that creates the version, and a commit that rebases +/// past a conflict lands on a different number again. +/// +/// Data file names can answer it. They are chosen before the commit, so the +/// marker the same commit carries can name them, and they are stable no matter +/// which version the commit ends up as. So: +/// +/// - the same files → the table is exactly what the sort left; +/// - the sorted files still there, with more after them → rows were appended, +/// and only then is `min_unsorted_rows` consulted, which is the churn that +/// threshold exists to damp; +/// - anything else → the data was replaced, whatever the row count says. +/// +/// Appends only ever add fragments after the existing ones, and lance hands out +/// fragment ids monotonically even across an overwrite, so "the sorted files +/// are still there" is a prefix test on the files in fragment order. +/// +/// Deletes leave the data files alone and write a deletion file beside them, so +/// they read as unchanged here — which is right: deleting rows does not unsort +/// the ones that remain. +pub fn verdict( + spec: &SortSpec, + state: &SortState, + fragments: &[FragmentSummary], + min_unsorted_rows: u64, +) -> SortVerdict { + let (Some(recorded_digest), Some(recorded_fragments)) = + (state.digest.as_deref(), state.fragments) + else { + return SortVerdict::NeverSorted; + }; + if state.fields.as_deref() != Some(spec.to_string().as_str()) { + return SortVerdict::OrderChanged; + } + if digest(&data_files(fragments)) == recorded_digest { + return SortVerdict::UpToDate; + } + + // Not the same files. Only a prefix match means the sorted data survived + // and the rest was appended after it. + if recorded_fragments > fragments.len() + || digest(&data_files(&fragments[..recorded_fragments])) != recorded_digest + { + return SortVerdict::Rewritten; + } + + // How many rows are in the fragments appended after the sorted ones. + // + // Not the table's net growth against the recorded row count: rows deleted + // from the sorted fragments hide appended rows one for one, and enough of + // them hide every appended row, leaving a table that reads as sorted with + // an unsorted tail. Counting the appended fragments themselves is the same + // arithmetic the threshold was always meant to do. + let mut appended = 0u64; + for fragment in &fragments[recorded_fragments..] { + // A fragment whose length the manifest does not record cannot be + // counted, and a table that cannot be judged is one to sort rather than + // one to leave alone forever. + let Some(rows) = fragment.rows else { + return SortVerdict::Rewritten; + }; + appended += rows; + } + if appended >= min_unsorted_rows.max(1) { + return SortVerdict::RowsAppended(appended); + } + SortVerdict::UpToDate +} + +/// The settings every sorting job offers, so admin renders one form for the +/// same behaviour whichever format the job sorts. +pub fn config_fields() -> Vec { + vec![ + text_field( + CONFIG_SORT_FIELDS, + "Sort fields", + "Order for tables that declare none of their own, as \"user_id asc, ts desc nulls-last\". Empty sorts only the tables that declare an order.", + "user_id asc, ts desc", + ), + number_field( + CONFIG_MIN_UNSORTED_ROWS, + "Minimum unsorted rows", + "Rows appended since the last sort before a table is worth sorting again", + 1, + 1_000_000_000, + ), + number_field( + CONFIG_MEMORY_BUDGET_MB, + "Memory budget (MB)", + "Memory the sort may hold before it spills runs to disk. A table larger than this sorts more slowly rather than failing.", + 64, + 1_048_576, + ), + number_field( + CONFIG_MAX_ROWS_PER_FILE, + "Maximum rows per file", + "Rows to write into each output file of a sorted rewrite", + 1024, + 16_777_216, + ), + ] +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_direction_and_null_order() { + let spec = SortSpec::parse("user_id, ts desc, name asc nulls-last") + .unwrap() + .unwrap(); + assert_eq!( + spec.fields, + vec![ + // Ascending defaults to nulls first, descending to nulls last, + // which is the rule the Iceberg tables sharing this spelling use. + SortField { + path: "user_id".into(), + descending: false, + nulls_first: true + }, + SortField { + path: "ts".into(), + descending: true, + nulls_first: false + }, + SortField { + path: "name".into(), + descending: false, + nulls_first: false + }, + ] + ); + } + + #[test] + fn renders_back_to_something_that_parses_the_same() { + let spec = SortSpec::parse("a, b desc nulls_first").unwrap().unwrap(); + let rendered = spec.to_string(); + assert_eq!(rendered, "a asc nulls-first, b desc nulls-first"); + assert_eq!(SortSpec::parse(&rendered).unwrap().unwrap(), spec); + } + + #[test] + fn empty_is_no_spec_rather_than_an_error() { + assert!(SortSpec::parse("").unwrap().is_none()); + assert!(SortSpec::parse(" , ").unwrap().is_none()); + } + + #[test] + fn rejects_nonsense_and_repeats() { + assert!(SortSpec::parse("a sideways").is_err()); + assert!(SortSpec::parse("a asc desc").is_err()); + assert!(SortSpec::parse("a, a desc").is_err()); + } + + #[test] + fn the_table_wins_and_configuration_fills_in() { + let configured = "b desc"; + let from_table = resolve(Some("a asc"), configured).unwrap().unwrap(); + assert_eq!(from_table.fields[0].path, "a"); + + let from_config = resolve(None, configured).unwrap().unwrap(); + assert_eq!(from_config.fields[0].path, "b"); + + let from_config_again = resolve(Some(" "), configured).unwrap().unwrap(); + assert_eq!(from_config_again.fields[0].path, "b"); + + assert!(resolve(None, "").unwrap().is_none()); + } + + // Falling back would sort the table by an order nobody asked for. + #[test] + fn a_broken_declaration_is_an_error_not_a_fallback() { + assert!(resolve(Some("a sideways"), "b desc").is_err()); + } + + /// One fragment per name, one file each, with a known row count. + fn frags(names: &[(&str, u64)]) -> Vec { + names + .iter() + .map(|(name, rows)| FragmentSummary { + files: vec![name.to_string()], + rows: Some(*rows), + }) + .collect() + } + + #[test] + fn verdicts_follow_the_marker() { + let spec = SortSpec::parse("a asc").unwrap().unwrap(); + let sorted = frags(&[("a.lance", 500), ("b.lance", 500)]); + + let fresh = SortState::default(); + assert_eq!( + verdict(&spec, &fresh, &sorted, 100), + SortVerdict::NeverSorted + ); + + let state = SortState { + fields: Some(spec.to_string()), + rows: Some(1_000), + fragments: Some(sorted.len()), + digest: Some(digest(&data_files(&sorted))), + }; + + // The same files, whatever version the commit ended up as. + assert_eq!(verdict(&spec, &state, &sorted, 100), SortVerdict::UpToDate); + + // Appended: the sorted fragments are still the first ones. + let appended_50 = frags(&[("a.lance", 500), ("b.lance", 500), ("c.lance", 50)]); + assert_eq!( + verdict(&spec, &state, &appended_50, 100), + SortVerdict::UpToDate + ); + let appended_100 = frags(&[("a.lance", 500), ("b.lance", 500), ("c.lance", 100)]); + assert_eq!( + verdict(&spec, &state, &appended_100, 100), + SortVerdict::RowsAppended(100) + ); + + // Replaced: different files, and the row count is no defence — this is + // the case a row-count threshold lets through, whatever the delta. + let replaced = frags(&[("x.lance", 500), ("y.lance", 501)]); + assert_eq!( + verdict(&spec, &state, &replaced, 100), + SortVerdict::Rewritten + ); + // Compacted into fewer files is a replacement too. + assert_eq!( + verdict(&spec, &state, &frags(&[("merged.lance", 1_000)]), 100), + SortVerdict::Rewritten + ); + + // Deleting rows rewrites no data file, so the table still reads as + // sorted — which it is. + let after_deletes = frags(&[("a.lance", 500), ("b.lance", 400)]); + assert_eq!( + verdict(&spec, &state, &after_deletes, 100), + SortVerdict::UpToDate + ); + + let other = SortSpec::parse("b desc").unwrap().unwrap(); + assert_eq!( + verdict(&other, &state, &sorted, 100), + SortVerdict::OrderChanged + ); + } + + /// Deletions must not hide appended rows. Net growth against the recorded + /// row count would say this table shrank, and it would stay unsorted with + /// an unsorted tail for as long as the deletes kept pace with the appends. + #[test] + fn deletions_do_not_mask_appended_rows() { + let spec = SortSpec::parse("a asc").unwrap().unwrap(); + let sorted = frags(&[("a.lance", 500), ("b.lance", 500)]); + let state = SortState { + fields: Some(spec.to_string()), + rows: Some(1_000), + fragments: Some(sorted.len()), + digest: Some(digest(&data_files(&sorted))), + }; + + // 800 rows deleted from the sorted fragments, 300 appended after them: + // the table is smaller than it was, and 300 rows of it are unsorted. + let mixed = frags(&[("a.lance", 100), ("b.lance", 100), ("c.lance", 300)]); + assert_eq!( + verdict(&spec, &state, &mixed, 100), + SortVerdict::RowsAppended(300) + ); + + // The threshold still damps a small append beside heavy deletion. + let small = frags(&[("a.lance", 100), ("b.lance", 100), ("c.lance", 20)]); + assert_eq!(verdict(&spec, &state, &small, 100), SortVerdict::UpToDate); + + // And net growth is still detected where there are no deletions. + let grown = frags(&[("a.lance", 500), ("b.lance", 500), ("c.lance", 700)]); + assert_eq!( + verdict(&spec, &state, &grown, 100), + SortVerdict::RowsAppended(700) + ); + } + + // An appended fragment whose length the manifest does not record cannot be + // measured against the threshold, and a table that cannot be judged is one + // to sort rather than one to leave alone forever. + #[test] + fn an_uncountable_appended_fragment_is_stale() { + let spec = SortSpec::parse("a asc").unwrap().unwrap(); + let sorted = frags(&[("a.lance", 500)]); + let state = SortState { + fields: Some(spec.to_string()), + rows: Some(500), + fragments: Some(1), + digest: Some(digest(&data_files(&sorted))), + }; + + let mut appended = sorted.clone(); + appended.push(FragmentSummary { + files: vec!["b.lance".to_string()], + rows: None, + }); + assert_eq!( + verdict(&spec, &state, &appended, 100), + SortVerdict::Rewritten + ); + // The untouched table is still up to date: the files answer that + // without needing to count anything. + assert_eq!(verdict(&spec, &state, &sorted, 100), SortVerdict::UpToDate); + } + + // Distinct columns whose names differ only in case are two columns in an + // Arrow schema, not one named twice. + #[test] + fn case_distinguishes_two_fields() { + let spec = SortSpec::parse("id asc, ID desc").unwrap().unwrap(); + assert_eq!(spec.fields.len(), 2); + assert!(SortSpec::parse("id asc, id desc").is_err()); + } + + #[test] + fn the_marker_round_trips_through_a_config_map() { + let spec = SortSpec::parse("a asc").unwrap().unwrap(); + let written = frags(&[("one.lance", 20), ("two.lance", 22)]); + let recorded = SortState::record(&spec, &written, 42); + let state = SortState::from_config(&recorded); + assert_eq!(state.rows, Some(42)); + assert_eq!(state.fragments, Some(2)); + assert_eq!(verdict(&spec, &state, &written, 1), SortVerdict::UpToDate); + } +}