diff --git a/.github/renovate.json b/.github/renovate.json index afdfacc..b75364a 100644 --- a/.github/renovate.json +++ b/.github/renovate.json @@ -3,5 +3,20 @@ "extends": [ "config:base" ], - "automerge": true + "automerge": false, + "packageRules": [ + { + "description": "Group Ballista and DataFusion updates together - versions must stay in sync", + "matchPackagePatterns": ["^ballista$", "^datafusion$"], + "groupName": "Ballista stack", + "automerge": false + }, + { + "description": "Auto-merge minor and patch updates for stable dependencies", + "matchUpdateTypes": ["minor", "patch"], + "matchPackagePatterns": ["^tokio$", "^anyhow$"], + "automerge": true, + "automergeType": "pr" + } + ] } diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..c005988 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,173 @@ +name: CI + +on: + push: + branches: [ main, claude/* ] + pull_request: + branches: [ main ] + +env: + CARGO_TERM_COLOR: always + RUST_BACKTRACE: 1 + +jobs: + test: + name: Test Suite + runs-on: ubuntu-latest + strategy: + matrix: + rust: + - stable + - beta + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@master + with: + toolchain: ${{ matrix.rust }} + + - name: Cache cargo registry + uses: actions/cache@v4 + with: + path: ~/.cargo/registry + key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }} + + - name: Cache cargo index + uses: actions/cache@v4 + with: + path: ~/.cargo/git + key: ${{ runner.os }}-cargo-git-${{ hashFiles('**/Cargo.lock') }} + + - name: Cache cargo build + uses: actions/cache@v4 + with: + path: target + key: ${{ runner.os }}-cargo-build-target-${{ hashFiles('**/Cargo.lock') }} + + - name: Run tests + run: cargo test --verbose + + fmt: + name: Formatting Check + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt + + - name: Check formatting + run: cargo fmt --all -- --check + + clippy: + name: Clippy (Linting) + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: clippy + + - name: Cache cargo registry + uses: actions/cache@v4 + with: + path: ~/.cargo/registry + key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }} + + - name: Cache cargo index + uses: actions/cache@v4 + with: + path: ~/.cargo/git + key: ${{ runner.os }}-cargo-git-${{ hashFiles('**/Cargo.lock') }} + + - name: Cache cargo build + uses: actions/cache@v4 + with: + path: target + key: ${{ runner.os }}-cargo-build-target-${{ hashFiles('**/Cargo.lock') }} + + - name: Run clippy + run: cargo clippy --all-targets --all-features -- -D warnings + + security-audit: + name: Security Audit + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Install cargo-audit + run: cargo install cargo-audit + + - name: Run security audit + run: cargo audit + + build: + name: Build Release + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Cache cargo registry + uses: actions/cache@v4 + with: + path: ~/.cargo/registry + key: ${{ runner.os }}-cargo-registry-${{ hashFiles('**/Cargo.lock') }} + + - name: Cache cargo index + uses: actions/cache@v4 + with: + path: ~/.cargo/git + key: ${{ runner.os }}-cargo-git-${{ hashFiles('**/Cargo.lock') }} + + - name: Cache cargo build + uses: actions/cache@v4 + with: + path: target + key: ${{ runner.os }}-cargo-build-target-${{ hashFiles('**/Cargo.lock') }} + + - name: Build release + run: cargo build --release --verbose + + - name: Upload artifacts + uses: actions/upload-artifact@v4 + with: + name: ballista-client-linux-x64 + path: target/release/ballista + + coverage: + name: Code Coverage + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Install tarpaulin + run: cargo install cargo-tarpaulin + + - name: Run coverage + run: cargo tarpaulin --verbose --all-features --workspace --timeout 120 --out xml + + - name: Upload to codecov.io + uses: codecov/codecov-action@v4 + with: + token: ${{ secrets.CODECOV_TOKEN }} + fail_ci_if_error: false diff --git a/.gitignore b/.gitignore index 088ba6b..100dde6 100644 --- a/.gitignore +++ b/.gitignore @@ -2,9 +2,9 @@ # will have compiled files and executables /target/ -# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries -# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html -Cargo.lock +# Cargo.lock is tracked for executables (as recommended by Cargo docs) +# This ensures reproducible builds across environments +# See: https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html # These are backup files generated by rustfmt **/*.rs.bk diff --git a/CLAUDE.md b/CLAUDE.md index b30d966..b958513 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,26 +4,34 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## Project Overview -Ballista is a distributed query execution client built on top of Apache Arrow Ballista and DataFusion. This repository contains a simple client application that connects to a Ballista scheduler to execute distributed queries against CSV and Parquet files. +Ballista is a production-ready distributed query execution client built on top of Apache Arrow Ballista and DataFusion. This repository demonstrates best practices for building distributed query applications with comprehensive configuration, observability, testing, and error handling. ## Architecture **Core Components:** +- **CLI Interface**: Built with `clap` for flexible command-line arguments and environment variable support +- **Configuration**: Environment-based configuration for scheduler address and file paths - **SessionContext**: The main client interface (from DataFusion) extended by Ballista to connect to remote schedulers - **Query Processing**: Uses DataFusion's DataFrame API for query construction - **Data Sources**: Supports CSV and Parquet file formats via DataFusion readers - **Extension Trait**: Ballista provides `SessionContextExt` trait that adds `remote()` method to `SessionContext` +- **Observability**: Structured logging with `tracing` for comprehensive visibility +- **Error Handling**: Rich error context using `anyhow::Context` for clear error messages **Execution Flow:** -1. Client establishes connection to Ballista scheduler via `SessionContextExt::remote()` -2. Constructs queries using DataFusion's DataFrame API (`read_csv()`, `read_parquet()`, etc.) -3. Queries are sent to scheduler for distributed execution -4. Results are displayed via `.show()` +1. Parse CLI arguments and environment variables +2. Initialize structured logging with `tracing-subscriber` +3. Client establishes connection to Ballista scheduler via `SessionContextExt::remote()` +4. Validate file existence before processing +5. Constructs queries using DataFusion's DataFrame API (`read_csv()`, `read_parquet()`, etc.) +6. Queries are sent to scheduler for distributed execution +7. Results are displayed via `.show()` with comprehensive error context **Important API Notes:** -- Ballista 48.0 uses DataFusion 48.0 (version alignment is critical) +- **Version Alignment Critical**: Ballista 49.0 requires DataFusion 49.0 (must stay in sync) - Connection requires fully qualified trait syntax: `::remote("df://localhost:50050")` - The `BallistaContext` and `BallistaConfig` types were deprecated in favor of DataFusion's native `SessionContext` +- Renovate is configured to group ballista+datafusion updates together to prevent version mismatches ## Development Commands @@ -33,11 +41,33 @@ Ballista is a distributed query execution client built on top of Apache Arrow Ba # Build the project cargo build -# Run the client (requires running Ballista cluster) +# Run the client with default settings (requires running Ballista cluster) cargo run +# Run with custom scheduler address +cargo run -- --scheduler df://remote-host:50050 + +# Run with custom files +cargo run -- --csv-file /path/to/data.csv --parquet-file /path/to/data.parquet + +# Skip CSV or Parquet processing +cargo run -- --skip-csv +cargo run -- --skip-parquet + +# Run with environment variables +BALLISTA_SCHEDULER=df://localhost:50050 CSV_FILE=testdata/test.csv cargo run + +# Run with debug logging +RUST_LOG=debug cargo run + +# Run with trace-level logging for maximum visibility +RUST_LOG=trace cargo run + # Build in release mode cargo build --release + +# Show CLI help +cargo run -- --help ``` ### Testing @@ -50,7 +80,10 @@ cargo test cargo test -- --nocapture # Run a specific test -cargo test test_name +cargo test test_args_parsing + +# Run tests with logging +RUST_LOG=debug cargo test -- --nocapture ``` ### Code Quality @@ -59,14 +92,17 @@ cargo test test_name # Check code without building cargo check -# Run clippy linter -cargo clippy +# Run clippy linter (enforced in CI) +cargo clippy --all-targets --all-features # Format code cargo fmt -# Check formatting without modifying files +# Check formatting without modifying files (enforced in CI) cargo fmt -- --check + +# Security audit +cargo audit ``` ## Cluster Setup Requirements @@ -88,30 +124,100 @@ RUST_LOG=info ballista-scheduler ```bash RUST_LOG=info ballista-executor --bind-port 50051 -c 4 RUST_LOG=info ballista-executor --bind-port 50052 -c 4 +RUST_LOG=info ballista-executor --bind-port 50053 -c 4 ``` -**Note:** The scheduler must be running on `localhost:50050` (default) for the client to connect. +**Note:** The scheduler runs on `localhost:50050` by default, which can be customized via CLI or environment variables. ## Test Data The `testdata/` directory contains sample files used by the client: -- `test.csv` - CSV format test data -- `test.parquet` - Parquet format test data +- `test.csv` - CSV format test data (100 rows, columns: c1, c2, c3, ...) +- `test.parquet` - Parquet format test data (columns: id, bool_col, timestamp_col, ...) - `alltypes_plain.parquet` - Additional Parquet test file -When modifying queries in `main.rs`, ensure the selected columns exist in these test files. +When modifying queries in `main.rs`, ensure the selected columns exist in these test files. The client validates file existence before processing. + +## Configuration + +### Environment Variables + +- `BALLISTA_SCHEDULER`: Scheduler address (default: `df://localhost:50050`) +- `CSV_FILE`: Path to CSV file (default: `testdata/test.csv`) +- `PARQUET_FILE`: Path to Parquet file (default: `testdata/test.parquet`) +- `RUST_LOG`: Logging level (`trace`, `debug`, `info`, `warn`, `error`) + +### CLI Arguments + +All environment variables can be overridden via command-line arguments: + +```bash +cargo run -- --help +``` + +Example output: +``` +Options: + -s, --scheduler Ballista scheduler address [env: BALLISTA_SCHEDULER=] [default: df://localhost:50050] + --csv-file CSV file path to query [env: CSV_FILE=] [default: testdata/test.csv] + --parquet-file Parquet file path to query [env: PARQUET_FILE=] [default: testdata/test.parquet] + --skip-csv Skip CSV processing + --skip-parquet Skip Parquet processing + -h, --help Print help + -V, --version Print version +``` ## Key Dependencies -- **ballista** (v48.0): Distributed query execution with SessionContextExt trait -- **datafusion** (v48.0): SQL query engine and DataFrame API (must match ballista version) +- **ballista** (v49.0): Distributed query execution with SessionContextExt trait +- **datafusion** (v49.0): SQL query engine and DataFrame API (MUST match ballista version) - **tokio** (v1.47): Async runtime (full features enabled) -- **anyhow** (v1.0.100): Error handling +- **anyhow** (v1.0.100): Error handling with context and backtraces +- **clap** (v4.5): CLI argument parsing with derive macros and environment variable support +- **tracing** (v0.1): Structured logging and instrumentation +- **tracing-subscriber** (v0.3): Log collection with environment-based filtering + +## CI/CD Pipeline + +The project uses GitHub Actions for continuous integration: + +### Automated Checks +- **Test Suite**: Runs on stable and beta Rust toolchains +- **Formatting**: Enforces `rustfmt` standards +- **Linting**: Clippy with warnings as errors +- **Security Audit**: `cargo-audit` checks for vulnerabilities +- **Build**: Release builds with artifact uploads +- **Coverage**: Code coverage with `tarpaulin` and Codecov integration + +### Dependency Management +- **Renovate**: Configured to group ballista+datafusion updates atomically +- **Automerge**: Disabled for critical dependencies, enabled for minor/patch updates of stable deps +- **Version Alignment**: Prevents ballista/datafusion version mismatches ## Important Notes - The client is async and requires the Tokio runtime (`#[tokio::main]`) -- All queries return `datafusion::common::Result<()>` for error propagation -- Ballista and DataFusion versions must be kept in sync (both at v48.0) -- Connection failures will occur if the scheduler is not running on port 50050 +- Error handling uses `anyhow::Result` with rich context via `.with_context()` +- **Version Alignment Critical**: Ballista 49.0 and DataFusion 49.0 must stay in sync +- Connection failures include helpful error messages indicating scheduler status +- File paths are validated before query execution - Use fully qualified trait syntax for SessionContextExt::remote() to avoid ambiguity +- Structured logging provides visibility into connection, query execution, and errors +- All functions are instrumented with `#[instrument]` for distributed tracing + +## Testing Strategy + +### Unit Tests +- CLI argument parsing validation +- Configuration defaults and overrides +- File path handling + +### Integration Tests (Future) +- Docker Compose-based cluster testing +- End-to-end query execution +- Error scenarios (missing scheduler, invalid files) + +### Test Coverage Goals +- Maintain >70% code coverage +- Cover all error paths +- Test CLI interface thoroughly diff --git a/Cargo.toml b/Cargo.toml index 5f51543..1298eea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,11 +2,26 @@ name = "ballista" version = "0.1.0" edition = "2021" - -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html +authors = ["Ballista Contributors"] +description = "A distributed query execution client built on Apache Arrow Ballista" +license = "MIT" +repository = "https://github.com/duyet/ballista" +readme = "README.md" [dependencies] -anyhow = "1.0.100" +# Core dependencies +anyhow = { version = "1.0.100", features = ["backtrace"] } ballista = { version = "49.0" } -datafusion = "48.0" +datafusion = "49.0" tokio = { version = "1.47", features = ["full"] } + +# CLI and configuration +clap = { version = "4.5", features = ["derive", "env"] } + +# Logging and observability +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } + +[dev-dependencies] +# Testing +tempfile = "3.15" diff --git a/README.md b/README.md index c45fa5e..d5c64ff 100644 --- a/README.md +++ b/README.md @@ -1,32 +1,79 @@ -# Ballista: Distributed Scheduler +# Ballista Client -[Ballista](https://github.com/apache/arrow-ballista) is a distributed compute platform primarily implemented in Rust. +A distributed query execution client built on [Apache Arrow Ballista](https://github.com/apache/arrow-ballista), demonstrating how to connect to a Ballista cluster and execute distributed queries against CSV and Parquet files. -## Starting a cluster +## Features -Local cluster for testing purposes +- **Distributed Query Processing**: Leverages Ballista's distributed execution engine +- **Multiple File Formats**: Supports CSV and Parquet data sources +- **DataFusion Integration**: Uses DataFusion's powerful DataFrame API +- **Production-Ready**: Configurable, observable, and thoroughly tested + +## Quick Start + +### 1. Install Ballista Components ```bash cargo install --locked ballista-scheduler cargo install --locked ballista-executor ``` -With these crates installed, it is now possible to start a scheduler process. +### 2. Start the Cluster +**Start the scheduler** (in terminal 1): ```bash RUST_LOG=info ballista-scheduler ``` -Next, start an executor processes in a new terminal session with the specified concurrency level. - +**Start executor(s)** (in separate terminals): ```bash +# Executor 1 RUST_LOG=info ballista-executor --bind-port 50051 -c 4 + +# Executor 2 (optional, for true distributed processing) RUST_LOG=info ballista-executor --bind-port 50052 -c 4 -RUST_LOG=info ballista-executor --bind-port 50052 -c 4 + +# Executor 3 (optional) +RUST_LOG=info ballista-executor --bind-port 50053 -c 4 ``` -## Executing a query +### 3. Run the Client ```bash +# Build and run cargo run + +# Or with custom configuration +BALLISTA_SCHEDULER=df://localhost:50050 cargo run ``` + +## Configuration + +Configure the client using environment variables: + +- `BALLISTA_SCHEDULER`: Scheduler address (default: `df://localhost:50050`) +- `RUST_LOG`: Logging level (e.g., `info`, `debug`, `trace`) + +## Development + +```bash +# Run tests +cargo test + +# Check code quality +cargo clippy + +# Format code +cargo fmt + +# Security audit +cargo audit +``` + +## Architecture + +See [CLAUDE.md](./CLAUDE.md) for detailed architecture documentation, API notes, and development guidelines. + +## License + +MIT License - see [LICENSE](./LICENSE) for details diff --git a/src/main.rs b/src/main.rs index d597b7f..b089877 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,45 +1,194 @@ +use anyhow::{Context, Result}; +use clap::Parser; use datafusion::prelude::{col, lit, CsvReadOptions, ParquetReadOptions, SessionContext}; +use std::path::PathBuf; +use tracing::{info, instrument}; + +/// Ballista distributed query client - demonstrating distributed query execution +#[derive(Parser, Debug)] +#[command(author, version, about, long_about = None)] +struct Args { + /// Ballista scheduler address + #[arg( + short, + long, + env = "BALLISTA_SCHEDULER", + default_value = "df://localhost:50050" + )] + scheduler: String, + + /// CSV file path to query + #[arg(long, env = "CSV_FILE", default_value = "testdata/test.csv")] + csv_file: PathBuf, + + /// Parquet file path to query + #[arg(long, env = "PARQUET_FILE", default_value = "testdata/test.parquet")] + parquet_file: PathBuf, + + /// Skip CSV processing + #[arg(long)] + skip_csv: bool, + + /// Skip Parquet processing + #[arg(long)] + skip_parquet: bool, +} #[tokio::main] -async fn main() -> datafusion::common::Result<()> { - // connect to Ballista scheduler - let ctx = - ::remote("df://localhost:50050") - .await?; +async fn main() -> Result<()> { + // Initialize tracing subscriber for structured logging + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_target(false) + .init(); - let csv_file = "testdata/test.csv"; - process_csv(&ctx, csv_file).await?; + let args = Args::parse(); - let parquet_file = "testdata/test.parquet"; - process_parquet(&ctx, parquet_file).await?; + info!("🚀 Starting Ballista distributed query client"); + info!("📡 Connecting to scheduler: {}", args.scheduler); + // Connect to Ballista scheduler + let ctx = connect_to_scheduler(&args.scheduler).await?; + info!("✅ Successfully connected to Ballista scheduler"); + + // Process CSV file + if !args.skip_csv { + process_csv(&ctx, &args.csv_file).await?; + } + + // Process Parquet file + if !args.skip_parquet { + process_parquet(&ctx, &args.parquet_file).await?; + } + + info!("🎉 All queries completed successfully"); Ok(()) } -async fn process_csv(ctx: &SessionContext, csv_file: &str) -> datafusion::common::Result<()> { - // define the query using the DataFrame trait +/// Establishes connection to the Ballista scheduler +#[instrument(skip_all, fields(scheduler = %scheduler_addr))] +async fn connect_to_scheduler(scheduler_addr: &str) -> Result { + info!("Establishing connection to Ballista scheduler"); + + ::remote(scheduler_addr) + .await + .with_context(|| { + format!( + "Failed to connect to Ballista scheduler at '{}'. \ + Ensure the scheduler is running and accessible.", + scheduler_addr + ) + }) +} + +/// Processes a CSV file with distributed query execution +#[instrument(skip(ctx), fields(file = %csv_file.display()))] +async fn process_csv(ctx: &SessionContext, csv_file: &PathBuf) -> Result<()> { + info!("📄 Processing CSV file"); + + // Validate file exists + if !csv_file.exists() { + anyhow::bail!( + "CSV file not found: '{}'. Please check the file path.", + csv_file.display() + ); + } + + // Execute distributed query let df = ctx - .read_csv(csv_file, CsvReadOptions::new()) - .await? - .select_columns(&["c1", "c2"])?; + .read_csv(csv_file.to_str().unwrap(), CsvReadOptions::new()) + .await + .with_context(|| format!("Failed to read CSV file: {}", csv_file.display()))? + .select_columns(&["c1", "c2"]) + .with_context(|| "Failed to select columns 'c1', 'c2'. Verify these columns exist.")?; - df.show().await?; + info!("Executing query and displaying results"); + df.show() + .await + .context("Failed to execute query or display results")?; + info!("✅ CSV query completed"); Ok(()) } -async fn process_parquet( - ctx: &SessionContext, - parquet_file: &str, -) -> datafusion::common::Result<()> { - // define the query using the DataFrame trait +/// Processes a Parquet file with distributed query execution and filtering +#[instrument(skip(ctx), fields(file = %parquet_file.display()))] +async fn process_parquet(ctx: &SessionContext, parquet_file: &PathBuf) -> Result<()> { + info!("📊 Processing Parquet file"); + + // Validate file exists + if !parquet_file.exists() { + anyhow::bail!( + "Parquet file not found: '{}'. Please check the file path.", + parquet_file.display() + ); + } + + // Execute distributed query with filter let df = ctx - .read_parquet(parquet_file, ParquetReadOptions::default()) - .await? - .select_columns(&["id", "bool_col", "timestamp_col"])? - .filter(col("id").gt(lit(1)))?; + .read_parquet( + parquet_file.to_str().unwrap(), + ParquetReadOptions::default(), + ) + .await + .with_context(|| format!("Failed to read Parquet file: {}", parquet_file.display()))? + .select_columns(&["id", "bool_col", "timestamp_col"]) + .with_context(|| { + "Failed to select columns 'id', 'bool_col', 'timestamp_col'. Verify schema." + })? + .filter(col("id").gt(lit(1))) + .context("Failed to apply filter: id > 1")?; - df.show().await?; + info!("Executing query with filter (id > 1) and displaying results"); + df.show() + .await + .context("Failed to execute query or display results")?; + info!("✅ Parquet query completed"); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_args_parsing() { + // Test default values + let args = Args::parse_from(["ballista"]); + assert_eq!(args.scheduler, "df://localhost:50050"); + assert_eq!(args.csv_file, PathBuf::from("testdata/test.csv")); + assert_eq!(args.parquet_file, PathBuf::from("testdata/test.parquet")); + assert!(!args.skip_csv); + assert!(!args.skip_parquet); + } + + #[test] + fn test_args_custom_scheduler() { + let args = Args::parse_from(["ballista", "--scheduler", "df://custom-host:9999"]); + assert_eq!(args.scheduler, "df://custom-host:9999"); + } + + #[test] + fn test_args_skip_flags() { + let args = Args::parse_from(["ballista", "--skip-csv", "--skip-parquet"]); + assert!(args.skip_csv); + assert!(args.skip_parquet); + } + + #[test] + fn test_args_custom_files() { + let args = Args::parse_from([ + "ballista", + "--csv-file", + "/custom/path.csv", + "--parquet-file", + "/custom/data.parquet", + ]); + assert_eq!(args.csv_file, PathBuf::from("/custom/path.csv")); + assert_eq!(args.parquet_file, PathBuf::from("/custom/data.parquet")); + } +}