Issue 5408 implement rust leaf partition query code

rust/Cargo.toml

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 [workspace]
-members = [ "aggregator_udfs", "filter_udfs", "compactor", "objectstore_ext", "rust_sketch", "sleeper_core", "sleeper_df" ]
+members = [ "aggregator_udfs", "filter_udfs", "apps", "objectstore_ext", "rust_sketch", "sleeper_core", "sleeper_df" ]
 resolver = "2"
 
 [workspace.package]
@@ -32,13 +32,12 @@ aws-config = { version = "1.8.6" }                  # Credential loading
 aws-credential-types = { version = "1.2.6" }        # Credential provider types
 aws-types = { version = "1.3.8" }                   # for Region
 bytes = { version = "1.10.1" }                      # Byte buffer for S3 uploading
-cargo_metadata = { version = "0.21.0" }             # Access cargo metadata programmtically
+cargo_metadata = { version = "0.22.0" }             # Access cargo metadata programmtically
 chrono = { version = "0.4.41" }                     # Log helper
 clap = { version = "4.5.47" }                       # Cmd line args processing
-tracing-subscriber = { version = "0.3.20" }
 color-eyre = { version = "0.6.5" }                  # Colourised version of `anyhow`
-cxx = { version = "1.0.170" }                       # Exception handling for Rust
-cxx-build = { version = "1.0.174" }
+cxx = { version = "1.0.175" }                       # Exception handling for Rust
+cxx-build = { version = "1.0.175" }
 datafusion = { version = "48.0.1" }
 env_logger = { version = "0.11.8" }                 # Standard logging to stderr
 futures = { version = "0.3.31" }                    # Async processing

rust/aggregator_udfs/src/nonnull.rs

@@ -28,14 +28,13 @@ use datafusion::{
         sum::sum_udaf,
     },
     logical_expr::{
-        Accumulator, AggregateUDF, AggregateUDFImpl, Documentation, EmitTo, GroupsAccumulator,
-        ReversedUDAF, SetMonotonicity, Signature, StatisticsArgs,
+        Accumulator, AggregateUDF, AggregateUDFImpl, Documentation, EmitTo, Expr,
+        GroupsAccumulator, ReversedUDAF, SetMonotonicity, Signature, StatisticsArgs,
         expr::{AggregateFunction, AggregateFunctionParams},
         function::{AccumulatorArgs, AggregateFunctionSimplification, StateFieldsArgs},
         simplify::SimplifyInfo,
         utils::AggregateOrderSensitivity,
     },
-    prelude::Expr,
     scalar::ScalarValue,
 };
 use std::{
@@ -455,16 +454,15 @@ mod tests {
             sum::{sum, sum_udaf},
         },
         logical_expr::{
-            Accumulator, AggregateUDFImpl, Documentation, EmitTo, GroupsAccumulator, ReversedUDAF,
-            SetMonotonicity, Signature, StatisticsArgs, Volatility,
+            Accumulator, AggregateUDFImpl, Documentation, EmitTo, Expr, GroupsAccumulator,
+            ReversedUDAF, SetMonotonicity, Signature, StatisticsArgs, Volatility,
             expr::{AggregateFunction, AggregateFunctionParams, WindowFunctionParams},
             function::{AccumulatorArgs, AggregateFunctionSimplification, StateFieldsArgs},
             lit,
             simplify::SimplifyContext,
             utils::AggregateOrderSensitivity,
         },
         physical_expr::LexOrdering,
-        prelude::Expr,
         scalar::ScalarValue,
     };
     use mockall::predicate::*;

rust/compactor/Cargo.toml → rust/apps/Cargo.toml

@@ -12,8 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 [package]
-name = "compactor"
-description = "Binary crate for running the data compactor from the command line."
+name = "apps"
+description = "Binary crate for running the Sleeper apps from the command line."
+default-run = "compact"
 keywords = ["sleeper", "compaction"]
 categories = ["science"]
 version = { workspace = true }
@@ -25,12 +26,13 @@ rust-version = { workspace = true }
 publish = { workspace = true }
 
 [dependencies]
+arrow = { workspace = true }
 chrono = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
-tracing-subscriber = { workspace = true }
 color-eyre = { workspace = true }
 sleeper_core = { path = "../sleeper_core" }
 env_logger = { workspace = true }
+futures = { workspace = true }
 human-panic = { workspace = true }
 log = { workspace = true, features = ["release_max_level_debug"] }
 num-format = { workspace = true }

rust/compactor/src/bin/main.rs → rust/apps/src/bin/compact.rs

@@ -1,32 +1,33 @@
 /*
- * Copyright 2022-2025 Crown Copyright
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+* Copyright 2022-2025 Crown Copyright
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+use apps::path_absolute;
 use chrono::Local;
 use clap::Parser;
 use color_eyre::eyre::bail;
 use human_panic::setup_panic;
 use log::info;
 use num_format::{Locale, ToFormattedString};
 use sleeper_core::{
-    ColRange, CommonConfig, OperationOutput, PartitionBound, SleeperParquetOptions,
+    ColRange, CommonConfigBuilder, OutputType, PartitionBound, SleeperParquetOptions,
     SleeperPartitionRegion, run_compaction,
 };
-use std::{collections::HashMap, io::Write, path::Path};
+use std::{collections::HashMap, io::Write};
 use url::Url;
 
-/// Implements a Sleeper compaction algorithm in Rust.
+/// Runs a Sleeper compaction algorithm.
 ///
 /// A sequence of Parquet files is read and compacted into a single output Parquet file. The input
 /// files must be individually sorted according to the row key columns and then the sort columns. A sketches file containing
@@ -63,16 +64,6 @@ struct CmdLineArgs {
     iterator_config: Option<String>,
 }
 
-/// Converts a [`Path`] reference to an absolute path (if not already absolute)
-/// and returns it as a String.
-///
-/// # Panics
-/// If the path can't be made absolute due to not being able to get the current
-/// directory or the path is not valid.
-fn path_absolute<T: ?Sized + AsRef<Path>>(path: &T) -> String {
-    std::path::absolute(path).unwrap().to_str().unwrap().into()
-}
-
 #[tokio::main(flavor = "multi_thread")]
 async fn main() -> color_eyre::Result<()> {
     // Install coloured errors
@@ -149,79 +140,26 @@ async fn main() -> color_eyre::Result<()> {
         dict_enc_values: true,
     };
 
-    let details = CommonConfig {
-        aws_config: None,
-        input_files: input_urls,
-        input_files_sorted: true,
-        row_key_cols: args.row_keys,
-        sort_key_cols: args.sort_keys,
-        region: SleeperPartitionRegion::new(map),
-        output: OperationOutput::File {
+    let details = CommonConfigBuilder::new()
+        .aws_config(None)
+        .input_files(input_urls)
+        .input_files_sorted(true)
+        .row_key_cols(args.row_keys)
+        .sort_key_cols(args.sort_keys)
+        .region(SleeperPartitionRegion::new(map))
+        .output(OutputType::File {
             output_file,
             opts: parquet_options,
-        },
-        iterator_config: args.iterator_config,
-    };
-
-    let result = run_compaction(&details).await;
-    match result {
-        Ok(r) => {
-            info!(
-                "Compaction read {} rows and wrote {} rows",
-                r.rows_read.to_formatted_string(&Locale::en),
-                r.rows_written.to_formatted_string(&Locale::en)
-            );
-        }
-        Err(e) => {
-            bail!(e);
-        }
-    }
-    Ok(())
-}
-
-#[cfg(test)]
-mod path_test {
-    use crate::path_absolute;
-
-    #[cfg(not(any(target_os = "windows", target_os = "macos")))]
-    fn cd_to_tmp() {
-        std::env::set_current_dir("/tmp").unwrap();
-    }
-
-    #[test]
-    #[cfg(not(any(target_os = "windows", target_os = "macos")))]
-    fn relative_path_converts() {
-        cd_to_tmp();
-        assert_eq!("/tmp/foo/bar/baz.txt", path_absolute("foo/bar/baz.txt"));
-    }
-
-    #[test]
-    #[cfg(not(any(target_os = "windows", target_os = "macos")))]
-    fn relative_path_converts_with_one_dot() {
-        cd_to_tmp();
-        assert_eq!("/tmp/foo/bar/baz.txt", path_absolute("./foo/bar/baz.txt"));
-    }
-
-    #[test]
-    #[cfg(not(any(target_os = "windows", target_os = "macos")))]
-    fn relative_path_converts_with_double_dot() {
-        cd_to_tmp();
-        assert_eq!(
-            "/tmp/../foo/bar/baz.txt",
-            path_absolute("../foo/bar/baz.txt")
-        );
-    }
+        })
+        .iterator_config(args.iterator_config)
+        .build()?;
 
-    #[test]
-    #[cfg(not(any(target_os = "windows", target_os = "macos")))]
-    fn absolute_path_unchanged() {
-        cd_to_tmp();
-        assert_eq!("/tmp/foo/bar", path_absolute("/tmp/foo/bar"));
-    }
+    let result = run_compaction(&details).await?;
+    info!(
+        "Compaction read {} rows and wrote {} rows",
+        result.rows_read.to_formatted_string(&Locale::en),
+        result.rows_written.to_formatted_string(&Locale::en)
+    );
 
-    #[test]
-    #[should_panic(expected = "cannot make an empty path absolute")]
-    fn empty_path_panic() {
-        let _ = path_absolute("");
-    }
+    Ok(())
 }