lance: a maintenance job that sorts a table by its declared fields (#11113)

* lance: a maintenance job that sorts a table by its declared fields

Lance appends fragments in write order and has no notion of a sorted table, so
nothing but a rewrite establishes one, and nothing but another rewrite restores
it once rows have been appended. lance_sort reads the order from the dataset's
own configuration, falls back to the worker's, and rewrites the table in it.

The spec and the marker live in crates/sort rather than in the job, because
weed/worker/tasks/iceberg sorts too: two jobs that disagreed about what
"id desc nulls-first" means would be two features wearing one name.

The sort spills. lance builds its DataFusion runtime with a FairSpillPool and a
disk manager, but only when LanceExecutionOptions::use_spilling is set, and that
struct derives Default over a plain bool — so Scanner::try_into_stream, which
fills its options with ..Default::default(), is precisely the path that does not
spill. The job builds the plan with create_plan and executes it with spilling on
and the operator's memory budget.

The marker rides in the same commit as the data: Operation::Overwrite is the one
operation carrying config values alongside fragments, so a sorted table and the
record of its sorting cannot disagree. It records the version the sort read, not
the one it wrote, which is not knowable while the marker is being assembled.
Detection treats anything committed after the sort's own commit as data the sort
did not produce — row counts alone cannot see a rewrite that leaves the count
where it was, and such a table would look sorted forever.

Claude-Session: https://claude.ai/code/session_015SZkLTUvd1svDu4xdr6Q3y

* lance: identify a sorted table by the files it wrote, not its version

Review found three ways the version-based marker misjudges a table, and they
share a cause: the version a sort produces is not knowable while the marker is
being assembled, so the marker recorded the version it read and detection
inferred the rest. A commit that rebases past a conflict lands on a different
number, and the inference then reads a rewrite into an ordinary table — a full
re-sort, and its indices, for nothing.

Data file names do not have that problem. They are chosen before the commit, so
the commit can carry them, and they do not change with the version it lands on.
The marker now records how many files the sort wrote and a digest of their
names, and detection asks whether the table still holds them: the same files
means untouched, the same files followed by more means appended, anything else
means the data was replaced.

That also closes the hole the row threshold left. A replacement that grew the
table by fewer rows than min_unsorted_rows read as sorted, however many rows had
actually moved; the threshold now applies only where the sorted files are still
in place, which is what it was for. A marker without a row count is stale rather
than a zero to compare against, and deletes stop forcing a re-sort — they write
a deletion file beside the data rather than rewriting it, and removing rows does
not unsort the ones that remain.

Sort fields are also compared exactly rather than case-folded. Arrow schemas are
case-sensitive, so `id` and `ID` are two columns, and folding them together
rejected a valid order.

Claude-Session: https://claude.ai/code/session_015SZkLTUvd1svDu4xdr6Q3y

* lance: count the rows appended after a sort, not the table's net growth

Review found that rows deleted from the sorted fragments hide appended rows one
for one: the threshold compared the live row count against the count recorded at
sort time, so 800 deletions and 300 appends read as a table that shrank, and a
table where deletions keep pace with appends stays "sorted" with an unsorted
tail forever.

The fragments say it directly. The marker already records how many fragments the
sort wrote, so the ones after that prefix are exactly what arrived since, and
the manifest carries each fragment's live row count — physical rows less its
deletions. Counting those is the arithmetic the threshold was always meant to
do, and it needs no row count from the marker at all.

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, so an uncountable appended fragment reads as stale.

Claude-Session: https://claude.ai/code/session_015SZkLTUvd1svDu4xdr6Q3y

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

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"

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

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"

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)

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<crate::metrics::LanceMetrics>,
+}
+
+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<crate::metrics::LanceMetrics>) -> 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<FragmentSummary> {
+    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<ColumnOrdering> {
+    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<String, ConfigValue> = 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<String, ConfigValue> = 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<String, ConfigValue> = 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(&params)
+            .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<FragmentSummary> = 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<String, ConfigValue> = 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(())
+    }
+}

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

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.",

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<String, ConfigValue> {
+    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<String> {
+    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<i64> = (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<Vec<i64>> {
+    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::<Int64Array>()
+            .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<i64> = (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(())
+}

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

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<SortField>,
+}
+
+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<Option<Self>> {
+        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<SortField> {
+    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<Option<SortSpec>> {
+    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<String>,
+    pub rows: Option<u64>,
+}
+
+/// 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<S: AsRef<str>>(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<String>,
+    pub rows: Option<u64>,
+    /// 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<usize>,
+    pub digest: Option<String>,
+}
+
+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<String, String>) -> 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<String, String> {
+        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<ConfigField> {
+    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<FragmentSummary> {
+        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);
+    }
+}