diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 6b16886..728521a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,22 +1,62 @@ name: Release on: + push: + tags: ["v*"] workflow_dispatch: inputs: - version: - description: 'Version to release (e.g., 0.1.0)' + tag: + description: "Release tag (e.g., v0.3.0)" required: true type: string env: RUST_BACKTRACE: 1 CARGO_TERM_COLOR: always + RUSTFLAGS: -Dwarnings jobs: + # Determine the tag to use (from push or workflow_dispatch) + prepare: + name: Prepare Release + runs-on: ubuntu-latest + outputs: + tag: ${{ steps.tag.outputs.tag }} + version: ${{ steps.tag.outputs.version }} + steps: + - uses: actions/checkout@v4 + + - name: Determine tag + id: tag + run: | + if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then + TAG="${{ inputs.tag }}" + else + TAG="${{ github.ref_name }}" + fi + # Validate tag format + if [[ ! "$TAG" =~ ^v[0-9]+\.[0-9]+\.[0-9]+.*$ ]]; then + echo "::error::Invalid tag format: $TAG (expected vX.Y.Z)" + exit 1 + fi + VERSION="${TAG#v}" + echo "tag=$TAG" >> "$GITHUB_OUTPUT" + echo "version=$VERSION" >> "$GITHUB_OUTPUT" + + - name: Verify Cargo.toml version matches tag + run: | + CARGO_VERSION=$(grep '^version = ' Cargo.toml | head -1 | sed 's/version = "\(.*\)"/\1/') + if [ "$CARGO_VERSION" != "${{ steps.tag.outputs.version }}" ]; then + echo "::error::Cargo.toml version ($CARGO_VERSION) does not match tag (${{ steps.tag.outputs.version }})" + exit 1 + fi + echo "Version verified: $CARGO_VERSION" + # Validate the release before publishing validate: name: Validate Release runs-on: ubuntu-latest + needs: prepare steps: - uses: actions/checkout@v5 @@ -25,14 +65,7 @@ jobs: with: components: rustfmt, clippy - - name: Verify version matches input - run: | - CARGO_VERSION=$(grep "^version" Cargo.toml | head -1 | cut -d'"' -f2) - if [ "$CARGO_VERSION" != "${{ github.event.inputs.version }}" ]; then - echo "Error: Cargo.toml version ($CARGO_VERSION) does not match input version (${{ github.event.inputs.version }})" - exit 1 - fi - echo "Version verified: $CARGO_VERSION" + - uses: Swatinem/rust-cache@v2 - name: Check formatting run: cargo fmt -- --check @@ -54,40 +87,54 @@ jobs: - name: Check documentation run: cargo doc --no-deps --all-features - # Publish to crates.io - publish: - name: Publish to crates.io + # Cargo publish dry run + publish-check: + name: Cargo Publish (Dry Run) runs-on: ubuntu-latest - needs: validate + needs: [prepare, validate] steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v4 - name: Install Rust uses: dtolnay/rust-toolchain@stable - - name: Publish to crates.io - run: cargo publish --token ${{ secrets.CARGO_REGISTRY_TOKEN }} - env: - CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }} + - uses: Swatinem/rust-cache@v2 + + - name: Cargo publish dry run + run: cargo publish --dry-run # Create GitHub release github-release: name: Create GitHub Release runs-on: ubuntu-latest - needs: publish + needs: [prepare, publish-check] permissions: contents: write steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v4 - - name: Create Release - uses: actions/create-release@v1 + - name: Create GitHub Release env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - with: - tag_name: v${{ github.event.inputs.version }} - release_name: v${{ github.event.inputs.version }} - body: | - See [CHANGELOG.md](https://github.com/${{ github.repository }}/blob/main/CHANGELOG.md) for details. - draft: false - prerelease: false + GH_TOKEN: ${{ github.token }} + run: | + gh release create ${{ needs.prepare.outputs.tag }} \ + --title "${{ needs.prepare.outputs.tag }}" \ + --generate-notes + + # Publish to crates.io + publish: + name: Publish to crates.io + runs-on: ubuntu-latest + needs: [prepare, github-release] + steps: + - uses: actions/checkout@v4 + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + - name: Publish to crates.io + env: + CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }} + run: cargo publish diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 9b8f001..b37cada 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -13,28 +13,18 @@ repos: entry: cargo fmt -- --check language: system pass_filenames: false - stages: [pre-commit, pre-push] - id: cargo-clippy name: cargo clippy entry: cargo clippy --locked --all-targets --all-features -- -D warnings language: system pass_filenames: false - stages: [pre-commit, pre-push] - - - id: cargo-test - name: cargo test - entry: cargo test --locked --all-features - language: system - pass_filenames: false - stages: [pre-push] - id: rust-quality-thresholds name: Rust code quality thresholds entry: python3 scripts/enforce_quality.py --max-loc 550 language: system pass_filenames: false - stages: [pre-commit, pre-push] - repo: https://github.com/compilerla/conventional-pre-commit rev: v3.3.0 diff --git a/AGENTS.md b/AGENTS.md index 1cdbf8a..d535ccc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -88,6 +88,22 @@ icepick compact my_namespace.my_table --target-size 268435456 icepick snapshot list my_namespace.my_table icepick snapshot cleanup my_namespace.my_table --dry-run icepick snapshot cleanup my_namespace.my_table --older-than-days 7 --retain-last 10 + +# Commit Parquet files to a table +icepick commit /data/**/*.parquet --namespace my_ns --table events --dry-run +icepick commit /data/**/*.parquet --namespace my_ns --table events + +# Create new table from Parquet files +icepick commit /data/**/*.parquet --namespace my_ns --table events \ + --create --partition year:int,month:int + +# Specify explicit partition values (for non-Hive paths) +icepick commit /flat/*.parquet --namespace my_ns --table events \ + --partition-values year=2024,month=01 + +# Use specific file as schema exemplar +icepick commit /data/**/*.parquet --namespace my_ns --table events \ + --exemplar /data/sample.parquet --create ``` ## CORE CONCEPTS @@ -363,6 +379,29 @@ if !plan.snapshots_to_remove.is_empty() { } ``` +### Pattern 9: Committing Parquet files + +```rust +use icepick::catalog::register::{ + introspect_parquet_file, parse_hive_partition_values, convert_partition_values, + register_data_files, RegisterOptions, +}; + +// Introspect a Parquet file (without partition extraction) +let introspection = introspect_parquet_file(file_io, path, None).await?; + +// Extract partition values from Hive-style path +let hive_values = parse_hive_partition_values(path); // HashMap + +// Convert to typed values using schema +let typed_values = convert_partition_values(&hive_values, &schema)?; + +// Or provide explicit values +let mut explicit = HashMap::new(); +explicit.insert("year".to_string(), "2024".to_string()); +let typed_values = convert_partition_values(&explicit, &schema)?; +``` + ## INTEGRATION POINTS - **Async Runtime**: tokio (required for examples/tests, not enforced as dependency) diff --git a/Cargo.lock b/Cargo.lock index 04eb0c7..761ef5c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1786,6 +1786,7 @@ dependencies = [ "dotenvy", "flate2", "futures", + "glob", "gloo-timers", "http 1.3.1", "humantime", diff --git a/Cargo.toml b/Cargo.toml index 028ea2e..5ac8c78 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -45,6 +45,7 @@ chrono = { version = "0.4.42", features = ["serde"] } [target.'cfg(not(target_family = "wasm"))'.dependencies] # Note: We use default-features to ensure proxy support works in all environments reqwest = { version = "0.12", features = ["json", "rustls-tls"] } +opendal = { version = "0.54", default-features = false, features = ["services-fs"] } aws-sigv4 = { version = "1.3.6", default-features = false, features = ["sign-http"] } aws-credential-types = { version = "1.2", default-features = false } aws-config = { version = "1.8", default-features = false, features = ["rustls", "behavior-version-latest", "rt-tokio"] } @@ -56,6 +57,7 @@ comfy-table = "7" bytesize = "1" humantime = "2" tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] } +glob = "0.3" # WASM targets [target.'cfg(target_family = "wasm")'.dependencies] diff --git a/DEVELOPER.md b/DEVELOPER.md new file mode 100644 index 0000000..b899e38 --- /dev/null +++ b/DEVELOPER.md @@ -0,0 +1,288 @@ +# Developer Guide + +This guide covers using icepick as a Rust library. For CLI usage, see [README.md](README.md). + +## Installation + +Add to your `Cargo.toml`: + +```toml +[dependencies] +icepick = "0.3" +``` + +## Quick Start + +### AWS S3 Tables + +```rust +use icepick::S3TablesCatalog; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Create catalog from S3 Tables ARN + let catalog = S3TablesCatalog::from_arn( + "my-catalog", + "arn:aws:s3tables:us-west-2:123456789012:bucket/my-bucket" + ).await?; + + // Load a table + let table = catalog.load_table( + &"namespace.table_name".parse()? + ).await?; + + Ok(()) +} +``` + +### Cloudflare R2 Data Catalog + +```rust +use icepick::R2Catalog; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Create catalog for R2 + let catalog = R2Catalog::new( + "my-catalog", + "account-id", + "bucket-name", + "api-token" + ).await?; + + // Load a table + let table = catalog.load_table( + &"namespace.table_name".parse()? + ).await?; + + Ok(()) +} +``` + +### Generic Iceberg REST Catalog + +```rust +use icepick::{FileIO, RestCatalog}; +use opendal::Operator; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Configure your FileIO (S3, R2, filesystem, etc.) + let operator = Operator::via_iter(opendal::Scheme::Memory, [])?; + let file_io = FileIO::new(operator); + + // Build a catalog for any Iceberg REST endpoint (Nessie, Glue REST, custom services) + let catalog = RestCatalog::builder("nessie", "https://nessie.example.com/api/iceberg") + .with_prefix("warehouse") + .with_file_io(file_io) + .with_bearer_token(std::env::var("NESSIE_TOKEN")?) + .build()?; + + let table = catalog.load_table(&"namespace.table".parse()?).await?; + Ok(()) +} +``` + +## Authentication + +### AWS S3 Tables + +Uses the **AWS default credential provider chain** in the following order: + +1. Environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`) +2. AWS credentials file (`~/.aws/credentials`) +3. IAM instance profile (EC2) +4. ECS task role + +> **Important:** Ensure your credentials have S3 Tables permissions. + +### Cloudflare R2 Data Catalog + +Uses **Cloudflare API tokens**. To set up: + +1. Log into the Cloudflare dashboard +2. Navigate to **My Profile** → **API Tokens** +3. Create a token with **R2 read/write permissions** +4. Pass the token when constructing the catalog + +## Direct S3 Parquet Writes + +Need to write Parquet files directly to S3 for external tools (Spark, DuckDB, etc.) without Iceberg metadata? Use the `arrow_to_parquet` function: + +```rust +use icepick::{arrow_to_parquet, FileIO, io::AwsCredentials}; +use arrow::array::{Int32Array, StringArray}; +use arrow::datatypes::{DataType, Field, Schema}; +use arrow::record_batch::RecordBatch; +use parquet::basic::Compression; +use std::sync::Arc; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Setup FileIO with AWS credentials + let file_io = FileIO::from_aws_credentials( + AwsCredentials { + access_key_id: "your-key".to_string(), + secret_access_key: "your-secret".to_string(), + session_token: None, + }, + "us-west-2".to_string() + ); + + // Create Arrow data + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new("name", DataType::Utf8, false), + ])); + + let batch = RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(StringArray::from(vec!["a", "b", "c"])), + ], + )?; + + // Simple write with defaults + arrow_to_parquet(&batch, "s3://my-bucket/output.parquet", &file_io).await?; + + // With compression + arrow_to_parquet(&batch, "s3://my-bucket/compressed.parquet", &file_io) + .with_compression(Compression::ZSTD(parquet::basic::ZstdLevel::default())) + .await?; + + // Manual partitioning (Hive-style or any structure) + let date = "2025-01-15"; + let path = format!("s3://my-bucket/data/date={}/data.parquet", date); + arrow_to_parquet(&batch, &path, &file_io).await?; + + Ok(()) +} +``` + +**Note:** This writes standalone Parquet files without Iceberg metadata. For writing to Iceberg tables, use the `Transaction` API instead. + +## Registering Existing Parquet Files + +Already have Parquet files in object storage? Register them into an Iceberg table without rewriting data: + +```rust +use icepick::{R2Catalog, introspect_parquet_file, DataFileRegistrar, RegisterOptions}; +use icepick::spec::{NamespaceIdent, TableIdent}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let catalog = R2Catalog::new("my-catalog", "account-id", "bucket", "token").await?; + + let namespace = NamespaceIdent::new(vec!["my_namespace".to_string()]); + let table_ident = TableIdent::new(namespace.clone(), "my_table".to_string()); + + // Introspect existing Parquet file to get schema, row count, size + let introspection = introspect_parquet_file( + catalog.file_io(), + "s3://bucket/path/to/file.parquet", + None, // partition spec (optional) + ).await?; + + // Register the file - creates table if needed + let options = RegisterOptions::new() + .allow_create_with_schema(introspection.schema.clone()) + .allow_noop(true); // idempotent - skip already-registered files + + let result = catalog.register_data_files( + namespace, + table_ident, + vec![introspection.data_file], + options, + ).await?; + + println!("Registered {} files ({} records)", result.added_files, result.added_records); + Ok(()) +} +``` + +This is useful for: +- Migrating existing Parquet datasets to Iceberg +- Registering files written by external tools (Spark, DuckDB, etc.) +- "Write to S3, register later" workflows in serverless environments + +## Snapshot Cleanup + +Automatically expire old snapshots to reduce metadata overhead and storage costs: + +```rust +use icepick::{R2Catalog, snapshot_cleanup::{plan_snapshot_cleanup, execute_snapshot_cleanup, CleanupOptions}}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let catalog = R2Catalog::new("my-catalog", "account-id", "bucket", "token").await?; + let table = catalog.load_table(&"namespace.table".parse()?).await?; + + // Configure retention policy + let options = CleanupOptions::new() + .with_older_than_days(7) // Expire snapshots older than 7 days + .with_retain_last(10); // Always keep at least 10 most recent + + // Preview what would be removed + let plan = plan_snapshot_cleanup(&table, &options)?; + println!("Will remove {} of {} snapshots", + plan.snapshots_to_remove.len(), plan.total_snapshots); + + // Execute cleanup + if !plan.snapshots_to_remove.is_empty() { + let result = execute_snapshot_cleanup(&table, &catalog, plan).await?; + println!("Removed {} snapshots", result.snapshots_removed); + } + + Ok(()) +} +``` + +## Platform Support + +| Catalog | Linux/macOS/Windows | WASM (browser/Cloudflare Workers) | +|---------|---------------------|-----------------------------------| +| **S3 Tables** | Supported | Not supported (requires AWS SDK) | +| **R2 Data Catalog** | Supported | Supported | +| **No Catalog** (direct parquet to object storage) | Supported | Supported | + +> **Note:** R2 Data Catalog and direct Parquet writes are fully WASM-compatible, making them suitable for Cloudflare Workers, browser applications, and other WASM environments. + +## Examples + +Explore complete working examples in the [`examples/`](examples/) directory: + +| Example | Description | Command | +|---------|-------------|---------| +| [`s3_tables_basic.rs`](examples/s3_tables_basic.rs) | Complete S3 Tables workflow | `cargo run --example s3_tables_basic` | +| [`r2_basic.rs`](examples/r2_basic.rs) | Complete R2 Data Catalog workflow | `cargo run --example r2_basic` | +| [`r2_register.rs`](examples/r2_register.rs) | Register existing Parquet files | `cargo run --example r2_register` | + +## Development + +### Running Tests + +```bash +cargo test +``` + +### WASM Build + +Verify R2Catalog compiles for WASM: + +```bash +cargo build --target wasm32-unknown-unknown +``` + +### Code Quality + +```bash +# Format code +cargo fmt + +# Run linter +cargo clippy -- -D warnings + +# Check documentation +cargo doc --no-deps --all-features +``` diff --git a/README.md b/README.md index eb1eaae..06de262 100644 --- a/README.md +++ b/README.md @@ -5,282 +5,163 @@ [![License](https://img.shields.io/crates/l/icepick.svg)](LICENSE) [![Rust](https://img.shields.io/badge/rust-2021%2B-blue.svg)](https://www.rust-lang.org) -> **Experimental client for Apache Iceberg in Rust** +A CLI tool and wasm-compatible library for managing Apache Iceberg tables in AWS S3 Tables and Cloudflare R2 Data Catalog. -**icepick** provides simple access to Apache Iceberg tables in AWS S3 Tables and Cloudflare R2 Data Catalog. Built on the official [iceberg-rust](https://github.com/apache/iceberg-rust) library, icepick handles authentication, REST API details, and platform compatibility so you can focus on working with your data. +## Table of Contents ---- +- [What it does](#what-it-does) +- [Why?](#why) +- [Quickstart](#quickstart) +- [CLI Reference](#cli-reference) + - [Namespaces](#namespaces) + - [Tables](#tables) + - [Commit Files](#commit-files) + - [Compaction](#compaction) + - [Snapshots](#snapshots) +- [Cloudflare R2](#cloudflare-r2) +- [AWS S3 Tables](#aws-s3-tables) +- [Library Usage](#library-usage) -### Why icepick? +## What it does -**Why not use [iceberg-rust](https://github.com/apache/iceberg-rust)?** This project targets WASM as a compilation target (not yet supported in `iceberg-rust`) and focuses on "serverless" catalogs that implement a subset of the overall Iceberg specification. +icepick provides a simple command-line interface and wasm-friendly library for working with Apache Iceberg tables: -## Features +- **List and inspect** namespaces and tables +- **Scan tables** with partition pruning and column statistics +- **Commit Parquet files** to tables (with auto-detection of Hive-style partitions) +- **Compact small files** using bin-pack compaction +- **Clean up snapshots** based on retention policies -### Catalog Support -- **AWS S3 Tables** — Full support with SigV4 authentication (native platforms only) -- **Cloudflare R2 Data Catalog** — Full support with bearer token auth (WASM-compatible) -- **Generic REST Catalog** — Build clients for any Iceberg REST endpoint (Nessie, Glue REST, custom) -- **Direct S3 Parquet Writes** — Write Arrow data directly to S3 without Iceberg metadata +## Why? -### Table Maintenance -- **Bin-pack Compaction** — Merge small files into larger ones for better query performance -- **Snapshot Cleanup** — Automatically expire old snapshots based on retention policies -- **Partition Pruning** — Filter scans by partition values and column statistics +The official [iceberg-rust](https://github.com/apache/iceberg-rust) library doesn't yet support WASM compilation, and most Iceberg tools are built for JVM environments. icepick fills the gap for: -### Developer Experience -- **Clean API** — Simple factory methods, no complex builders -- **Type-safe errors** — Comprehensive error handling with context -- **Zero-config auth** — Uses AWS credential chain and Cloudflare API tokens -- **Production-ready** — Used in real applications with real data +- **Serverless environments** like Cloudflare Workers +- **CLI-first workflows** without spinning up Spark or Flink +- **Lightweight table maintenance** (compaction, snapshot cleanup) +- **Quick data exploration** without complex query engines -## Platform Support +## Quickstart -| Catalog | Linux/macOS/Windows | WASM (browser/Cloudflare Workers) | -|---------|---------------------|-----------------------------------| -| **S3 Tables** | ✅ | ❌ (requires AWS SDK) | -| **R2 Data Catalog** | ✅ | ✅ | -| **No Catalog** (direct parquet to object storage) | ✅ | ✅ | +### Install -> **Note:** R2 Data Catalog and direct Parquet writes are fully WASM-compatible, making them suitable for Cloudflare Workers, browser applications, and other WASM environments. - -## Installation - -Add to your `Cargo.toml`: - -```toml -[dependencies] -icepick = "0.3" +```bash +cargo install icepick --features cli ``` -## Quick Start - -### AWS S3 Tables - -```rust -use icepick::S3TablesCatalog; +### Configure -#[tokio::main] -async fn main() -> Result<(), Box> { - // Create catalog from S3 Tables ARN - let catalog = S3TablesCatalog::from_arn( - "my-catalog", - "arn:aws:s3tables:us-west-2:123456789012:bucket/my-bucket" - ).await?; +Set your catalog credentials: - // Load a table - let table = catalog.load_table( - &"namespace.table_name".parse()? - ).await?; +```bash +# For Cloudflare R2 +export ICEPICK_CATALOG_URL="https://catalog.cloudflarestorage.com//" +export ICEPICK_TOKEN="" - Ok(()) -} +# For AWS S3 Tables +export ICEPICK_CATALOG_ARN="arn:aws:s3tables:us-west-2:123456789012:bucket/my-bucket" +# Uses AWS credential chain (env vars, ~/.aws/credentials, IAM role) ``` -### Cloudflare R2 Data Catalog - -```rust -use icepick::R2Catalog; +### Verify Connection -#[tokio::main] -async fn main() -> Result<(), Box> { - // Create catalog for R2 - let catalog = R2Catalog::new( - "my-catalog", - "account-id", - "bucket-name", - "api-token" - ).await?; +```bash +# List namespaces +icepick namespace list - // Load a table - let table = catalog.load_table( - &"namespace.table_name".parse()? - ).await?; +# List tables in a namespace +icepick table list --namespace my_namespace - Ok(()) -} +# Get table info +icepick table info my_namespace.my_table ``` -### Generic Iceberg REST Catalog +## CLI Reference -```rust -use icepick::{FileIO, RestCatalog}; -use opendal::Operator; +### Namespaces -#[tokio::main] -async fn main() -> Result<(), Box> { - // Configure your FileIO (S3, R2, filesystem, etc.) - let operator = Operator::via_iter(opendal::Scheme::Memory, [])?; - let file_io = FileIO::new(operator); +```bash +# List all namespaces +icepick namespace list - // Build a catalog for any Iceberg REST endpoint (Nessie, Glue REST, custom services) - let catalog = RestCatalog::builder("nessie", "https://nessie.example.com/api/iceberg") - .with_prefix("warehouse") - .with_file_io(file_io) - .with_bearer_token(std::env::var("NESSIE_TOKEN")?) - .build()?; +# Create a namespace +icepick namespace create my_namespace - let table = catalog.load_table(&"namespace.table".parse()?).await?; - Ok(()) -} +# Delete a namespace +icepick namespace delete my_namespace ``` -## Authentication - -### AWS S3 Tables - -Uses the **AWS default credential provider chain** in the following order: +### Tables -1. Environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`) -2. AWS credentials file (`~/.aws/credentials`) -3. IAM instance profile (EC2) -4. ECS task role +```bash +# List tables in a namespace +icepick table list --namespace my_namespace -> **Important:** Ensure your credentials have S3 Tables permissions. +# Get detailed table info (schema, partitioning, snapshots) +icepick table info my_namespace.my_table -### Cloudflare R2 Data Catalog +# Scan table data (shows pruning stats with filters) +icepick table scan my_namespace.my_table -Uses **Cloudflare API tokens**. To set up: +# Scan with filter +icepick table scan my_namespace.my_table --filter "date >= '2024-01-01'" -1. Log into the Cloudflare dashboard -2. Navigate to **My Profile** → **API Tokens** -3. Create a token with **R2 read/write permissions** -4. Pass the token when constructing the catalog - -## Direct S3 Parquet Writes - -Need to write Parquet files directly to S3 for external tools (Spark, DuckDB, etc.) without Iceberg metadata? Use the `arrow_to_parquet` function: - -```rust -use icepick::{arrow_to_parquet, FileIO, io::AwsCredentials}; -use arrow::array::{Int32Array, StringArray}; -use arrow::datatypes::{DataType, Field, Schema}; -use arrow::record_batch::RecordBatch; -use parquet::basic::Compression; -use std::sync::Arc; - -#[tokio::main] -async fn main() -> Result<(), Box> { - // Setup FileIO with AWS credentials - let file_io = FileIO::from_aws_credentials( - AwsCredentials { - access_key_id: "your-key".to_string(), - secret_access_key: "your-secret".to_string(), - session_token: None, - }, - "us-west-2".to_string() - ); - - // Create Arrow data - let schema = Arc::new(Schema::new(vec![ - Field::new("id", DataType::Int32, false), - Field::new("name", DataType::Utf8, false), - ])); - - let batch = RecordBatch::try_new( - schema, - vec![ - Arc::new(Int32Array::from(vec![1, 2, 3])), - Arc::new(StringArray::from(vec!["a", "b", "c"])), - ], - )?; - - // Simple write with defaults - arrow_to_parquet(&batch, "s3://my-bucket/output.parquet", &file_io).await?; - - // With compression - arrow_to_parquet(&batch, "s3://my-bucket/compressed.parquet", &file_io) - .with_compression(Compression::ZSTD(parquet::basic::ZstdLevel::default())) - .await?; - - // Manual partitioning (Hive-style or any structure) - let date = "2025-01-15"; - let path = format!("s3://my-bucket/data/date={}/data.parquet", date); - arrow_to_parquet(&batch, &path, &file_io).await?; - - Ok(()) -} +# Limit output rows +icepick table scan my_namespace.my_table --limit 100 ``` -**Note:** This writes standalone Parquet files without Iceberg metadata. For writing to Iceberg tables, use the `Transaction` API instead. - -## Registering Existing Parquet Files - -Already have Parquet files in object storage? Register them into an Iceberg table without rewriting data: - -```rust -use icepick::{R2Catalog, introspect_parquet_file, DataFileRegistrar, RegisterOptions}; -use icepick::spec::{NamespaceIdent, TableIdent}; +### Commit Files -#[tokio::main] -async fn main() -> Result<(), Box> { - let catalog = R2Catalog::new("my-catalog", "account-id", "bucket", "token").await?; +Commit existing Parquet files to an Iceberg table: - let namespace = NamespaceIdent::new(vec!["my_namespace".to_string()]); - let table_ident = TableIdent::new(namespace.clone(), "my_table".to_string()); +```bash +# Preview what would be committed (dry run) +icepick commit /data/**/*.parquet --namespace prod --table events --dry-run - // Introspect existing Parquet file to get schema, row count, size - let introspection = introspect_parquet_file( - catalog.file_io(), - "s3://bucket/path/to/file.parquet", - None, // partition spec (optional) - ).await?; +# Commit files to existing table +icepick commit /data/**/*.parquet --namespace prod --table events - // Register the file - creates table if needed - let options = RegisterOptions::new() - .allow_create_with_schema(introspection.schema.clone()) - .allow_noop(true); // idempotent - skip already-registered files +# Create new table with partition spec +icepick commit /data/**/*.parquet --namespace prod --table events \ + --create --partition year:int,month:int - let result = catalog.register_data_files( - namespace, - table_ident, - vec![introspection.data_file], - options, - ).await?; +# For non-Hive paths, specify partition values explicitly +icepick commit /flat/*.parquet --namespace prod --table events \ + --partition-values year=2024,month=01 - println!("Registered {} files ({} records)", result.added_files, result.added_records); - Ok(()) -} +# Use specific file as schema exemplar +icepick commit /data/**/*.parquet --namespace prod --table events \ + --exemplar /data/sample.parquet --create ``` -This is useful for: -- Migrating existing Parquet datasets to Iceberg -- Registering files written by external tools (Spark, DuckDB, etc.) -- "Write to S3, register later" workflows in serverless environments - -## Snapshot Cleanup - -Automatically expire old snapshots to reduce metadata overhead and storage costs: +The commit command: +- Uses first file's schema (or `--exemplar`) as the reference +- Validates all files match the schema +- Extracts partition values from Hive-style paths automatically +- Supports `--partition-values` for flat directory structures +- Shows detailed plan with `--dry-run` before committing -```rust -use icepick::{R2Catalog, snapshot_cleanup::{plan_snapshot_cleanup, execute_snapshot_cleanup, CleanupOptions}}; +### Compaction -#[tokio::main] -async fn main() -> Result<(), Box> { - let catalog = R2Catalog::new("my-catalog", "account-id", "bucket", "token").await?; - let table = catalog.load_table(&"namespace.table".parse()?).await?; +Merge small files into larger ones for better query performance: - // Configure retention policy - let options = CleanupOptions::new() - .with_older_than_days(7) // Expire snapshots older than 7 days - .with_retain_last(10); // Always keep at least 10 most recent +```bash +# Preview compaction plan (dry run) +icepick compact my_namespace.my_table --dry-run - // Preview what would be removed - let plan = plan_snapshot_cleanup(&table, &options)?; - println!("Will remove {} of {} snapshots", - plan.snapshots_to_remove.len(), plan.total_snapshots); +# Execute compaction with default settings +icepick compact my_namespace.my_table - // Execute cleanup - if !plan.snapshots_to_remove.is_empty() { - let result = execute_snapshot_cleanup(&table, &catalog, plan).await?; - println!("Removed {} snapshots", result.snapshots_removed); - } +# Custom target file size (256 MB) +icepick compact my_namespace.my_table --target-size 268435456 - Ok(()) -} +# Only compact files smaller than 128 MB +icepick compact my_namespace.my_table --max-input-size 134217728 ``` -### CLI Usage +### Snapshots + +Manage table snapshots and clean up old versions: ```bash # List all snapshots with age and status @@ -289,7 +170,7 @@ icepick snapshot list my_namespace.my_table # Preview cleanup (dry run) icepick snapshot cleanup my_namespace.my_table --dry-run -# Execute cleanup with custom retention +# Execute cleanup with retention policy icepick snapshot cleanup my_namespace.my_table \ --older-than-days 7 \ --retain-last 10 @@ -301,44 +182,55 @@ Snapshot cleanup respects: - **Retention count** - Keeps the N most recent regardless of age - **Age threshold** - Only expires snapshots older than the threshold -## Examples - -Explore complete working examples in the [`examples/`](examples/) directory: +## Cloudflare R2 -| Example | Description | Command | -|---------|-------------|---------| -| [`s3_tables_basic.rs`](examples/s3_tables_basic.rs) | Complete S3 Tables workflow | `cargo run --example s3_tables_basic` | -| [`r2_basic.rs`](examples/r2_basic.rs) | Complete R2 Data Catalog workflow | `cargo run --example r2_basic` | -| [`r2_register.rs`](examples/r2_register.rs) | Register existing Parquet files | `cargo run --example r2_register` | +### Authentication -## Development - -### Running Tests +1. Log into the Cloudflare dashboard +2. Navigate to **My Profile** → **API Tokens** +3. Create a token with **R2 read/write permissions** +4. Set environment variables: ```bash -cargo test +export ICEPICK_CATALOG_URL="https://catalog.cloudflarestorage.com//" +export ICEPICK_TOKEN="" ``` -### WASM Build +### WASM Compatibility + +The R2 catalog is fully WASM-compatible, making it suitable for: +- Cloudflare Workers +- Browser applications (if your catalog REST API supports CORS) + +## AWS S3 Tables -Verify R2Catalog compiles for WASM: +### Authentication + +Uses the AWS default credential provider chain: + +1. Environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`) +2. AWS credentials file (`~/.aws/credentials`) +3. IAM instance profile (EC2) +4. ECS task role ```bash -cargo build --target wasm32-unknown-unknown +export ICEPICK_CATALOG_ARN="arn:aws:s3tables:us-west-2:123456789012:bucket/my-bucket" ``` -### Code Quality +> **Important:** Ensure your credentials have S3 Tables permissions. -```bash -# Format code -cargo fmt +### Platform Support -# Run linter -cargo clippy -- -D warnings +S3 Tables requires the AWS SDK and is only available on native platforms (Linux, macOS, Windows). It does not compile to WASM. -# Check documentation -cargo doc --no-deps --all-features -``` +## Library Usage + +icepick can also be used as a Rust library for programmatic access to Iceberg tables. See [DEVELOPER.md](DEVELOPER.md) for: + +- Rust API examples +- Direct Parquet writes +- Registering existing files +- WASM considerations ## Contributing diff --git a/src/bin/icepick.rs b/src/bin/icepick.rs index e6b7dd7..2d2f94f 100644 --- a/src/bin/icepick.rs +++ b/src/bin/icepick.rs @@ -2,8 +2,8 @@ use clap::{Parser, Subcommand}; use icepick::cli::commands::{ - catalog as catalog_cmd, compact as compact_cmd, namespace as namespace_cmd, - snapshot as snapshot_cmd, table as table_cmd, + catalog as catalog_cmd, commit as commit_cmd, compact as compact_cmd, + namespace as namespace_cmd, snapshot as snapshot_cmd, table as table_cmd, }; use icepick::cli::{CatalogConfig, OutputFormat}; @@ -48,6 +48,9 @@ enum Commands { /// Compact a table Compact(compact_cmd::CompactArgs), + + /// Commit Parquet files to a table + Commit(commit_cmd::CommitArgs), } #[tokio::main] @@ -73,6 +76,7 @@ async fn main() { Commands::Table(cmd) => table_cmd::execute(cmd, &config, cli.output).await, Commands::Snapshot(cmd) => snapshot_cmd::execute(cmd, &config, cli.output).await, Commands::Compact(args) => compact_cmd::execute(args, &config, cli.output).await, + Commands::Commit(args) => commit_cmd::execute(args, &config, cli.output).await, }; if let Err(e) = result { diff --git a/src/catalog/catalog_trait.rs b/src/catalog/catalog_trait.rs index b4d9f66..7a66293 100644 --- a/src/catalog/catalog_trait.rs +++ b/src/catalog/catalog_trait.rs @@ -5,6 +5,7 @@ use async_trait::async_trait; use std::collections::HashMap; use crate::error::Result; +use crate::io::FileIO; use crate::spec::{NamespaceIdent, TableCreation, TableIdent}; use crate::table::Table; @@ -12,6 +13,9 @@ use crate::table::Table; #[cfg_attr(not(target_family = "wasm"), async_trait)] #[cfg_attr(target_family = "wasm", async_trait(?Send))] pub trait Catalog: Send + Sync { + /// Get the FileIO for reading/writing data files + fn file_io(&self) -> &FileIO; + /// Create a namespace (idempotent - returns Ok if already exists) async fn create_namespace( &self, diff --git a/src/catalog/r2.rs b/src/catalog/r2.rs index 773142a..2c3b54d 100644 --- a/src/catalog/r2.rs +++ b/src/catalog/r2.rs @@ -244,6 +244,10 @@ impl R2Catalog { #[cfg(not(target_family = "wasm"))] #[async_trait] impl Catalog for R2Catalog { + fn file_io(&self) -> &crate::io::FileIO { + self.inner.file_io() + } + async fn create_namespace( &self, namespace: &NamespaceIdent, @@ -300,6 +304,10 @@ impl Catalog for R2Catalog { #[cfg(target_family = "wasm")] #[async_trait(?Send)] impl Catalog for R2Catalog { + fn file_io(&self) -> &crate::io::FileIO { + self.inner.file_io() + } + async fn create_namespace( &self, namespace: &NamespaceIdent, diff --git a/src/catalog/register/introspect.rs b/src/catalog/register/introspect.rs index 5787072..97aaee6 100644 --- a/src/catalog/register/introspect.rs +++ b/src/catalog/register/introspect.rs @@ -131,6 +131,33 @@ pub struct ParquetIntrospection { pub partition_values: Option>, } +/// Introspect a local Parquet file on the filesystem. +/// +/// This is a convenience wrapper for CLI use that creates a local FileIO +/// and calls `introspect_parquet_file`. The returned `DataFileInput` will +/// have `file_path` set to the original local path - callers should update +/// this to the remote path after uploading. +/// +/// # Arguments +/// * `path` - Absolute path to a local Parquet file +/// * `partition_spec` - Optional partition spec for extracting Hive-style partition values +#[cfg(not(target_family = "wasm"))] +pub async fn introspect_local_parquet_file( + path: &str, + partition_spec: Option<&PartitionSpec>, +) -> Result { + use crate::io::local::{create_local_file_io, get_filename}; + + let local_file_io = create_local_file_io(path)?; + let filename = get_filename(path); + let mut result = introspect_parquet_file(&local_file_io, filename, partition_spec).await?; + + // Restore the original full path (introspect_parquet_file only sees the filename) + result.data_file.file_path = path.to_string(); + + Ok(result) +} + /// Infer partition values from a path like `col1=value1/col2=5/part-000.parquet`. /// /// This is intentionally strict when a partition spec is provided: @@ -169,7 +196,18 @@ pub fn infer_partition_values_from_path( } /// Extract Hive-style `key=value` segments from a path. -fn parse_hive_partition_values(path: &str) -> HashMap { +/// +/// Returns a map of partition column names to their string values. +/// Does not validate against any schema or partition spec. +/// +/// # Example +/// ``` +/// use icepick::catalog::register::parse_hive_partition_values; +/// +/// let values = parse_hive_partition_values("s3://bucket/year=2024/month=01/file.parquet"); +/// assert_eq!(values.get("year"), Some(&"2024".to_string())); +/// ``` +pub fn parse_hive_partition_values(path: &str) -> HashMap { path.rsplit_once('/') .map(|(dirs, file)| (dirs, Some(file))) .unwrap_or((path, None)) @@ -187,6 +225,72 @@ fn parse_hive_partition_values(path: &str) -> HashMap { .collect() } +/// Convert raw string partition values to typed PartitionValue based on schema. +/// +/// Looks up each partition column in the schema to determine the correct type. +/// Unknown columns are treated as strings. +/// +/// # Example +/// ``` +/// use icepick::catalog::register::convert_partition_values; +/// use icepick::catalog::register::PartitionValue; +/// use icepick::spec::{NestedField, PrimitiveType, Schema, Type}; +/// use std::collections::HashMap; +/// +/// let schema = Schema::builder() +/// .with_fields(vec![ +/// NestedField::required_field(1, "year".to_string(), Type::Primitive(PrimitiveType::Int)), +/// ]) +/// .build() +/// .unwrap(); +/// +/// let mut raw = HashMap::new(); +/// raw.insert("year".to_string(), "2024".to_string()); +/// +/// let typed = convert_partition_values(&raw, &schema).unwrap(); +/// assert_eq!(typed.get("year"), Some(&PartitionValue::Int(2024))); +/// ``` +pub fn convert_partition_values( + raw_values: &HashMap, + schema: &Schema, +) -> Result> { + let mut typed_values = HashMap::new(); + + for (name, raw) in raw_values { + let value = match schema.fields().iter().find(|f| f.name() == name) { + Some(field) => parse_value_by_type(field.field_type(), raw)?, + None => PartitionValue::String(raw.clone()), + }; + typed_values.insert(name.clone(), value); + } + + Ok(typed_values) +} + +fn parse_value_by_type(field_type: &crate::spec::Type, raw: &str) -> Result { + use crate::spec::PrimitiveType; + + match field_type { + crate::spec::Type::Primitive(PrimitiveType::Boolean) => raw + .parse::() + .map(PartitionValue::Bool) + .map_err(|e| Error::invalid_input(format!("Invalid boolean '{}': {}", raw, e))), + crate::spec::Type::Primitive(PrimitiveType::Int) + | crate::spec::Type::Primitive(PrimitiveType::Date) => raw + .parse::() + .map(PartitionValue::Int) + .map_err(|e| Error::invalid_input(format!("Invalid int '{}': {}", raw, e))), + crate::spec::Type::Primitive(PrimitiveType::Long) + | crate::spec::Type::Primitive(PrimitiveType::Time) + | crate::spec::Type::Primitive(PrimitiveType::Timestamp) + | crate::spec::Type::Primitive(PrimitiveType::Timestamptz) => raw + .parse::() + .map(PartitionValue::Long) + .map_err(|e| Error::invalid_input(format!("Invalid long '{}': {}", raw, e))), + _ => Ok(PartitionValue::String(raw.to_string())), + } +} + fn parse_partition_value( schema: &Schema, field: &PartitionField, diff --git a/src/catalog/register/introspect/tests.rs b/src/catalog/register/introspect/tests.rs index 1d5c2ba..cd100cc 100644 --- a/src/catalog/register/introspect/tests.rs +++ b/src/catalog/register/introspect/tests.rs @@ -180,3 +180,51 @@ fn malformed_partition_value_errors() { "unexpected error: {err}" ); } + +#[test] +fn test_parse_hive_partition_values_standalone() { + let path = "s3://bucket/year=2024/month=01/data.parquet"; + let result = super::parse_hive_partition_values(path); + + assert_eq!(result.get("year"), Some(&"2024".to_string())); + assert_eq!(result.get("month"), Some(&"01".to_string())); + assert_eq!(result.len(), 2); +} + +#[test] +fn test_parse_hive_partition_values_no_partitions() { + let path = "s3://bucket/data/file.parquet"; + let result = super::parse_hive_partition_values(path); + + assert!(result.is_empty()); +} + +#[test] +fn test_convert_partition_values_to_typed() { + use crate::catalog::register::types::PartitionValue; + + // Create a simple schema with year (int) and region (string) + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required_field(1, "year".to_string(), Type::Primitive(PrimitiveType::Int)), + NestedField::required_field( + 2, + "region".to_string(), + Type::Primitive(PrimitiveType::String), + ), + ]) + .build() + .unwrap(); + + let mut raw_values = std::collections::HashMap::new(); + raw_values.insert("year".to_string(), "2024".to_string()); + raw_values.insert("region".to_string(), "us-west".to_string()); + + let result = super::convert_partition_values(&raw_values, &schema).unwrap(); + + assert_eq!(result.get("year"), Some(&PartitionValue::Int(2024))); + assert_eq!( + result.get("region"), + Some(&PartitionValue::String("us-west".to_string())) + ); +} diff --git a/src/catalog/register/mod.rs b/src/catalog/register/mod.rs index e9b0023..5ef846f 100644 --- a/src/catalog/register/mod.rs +++ b/src/catalog/register/mod.rs @@ -14,8 +14,11 @@ use crate::table::Table; use chrono::Utc; use validate::validate_schema; +#[cfg(not(target_family = "wasm"))] +pub use introspect::introspect_local_parquet_file; pub use introspect::{ - infer_partition_values_from_path, introspect_parquet_file, ParquetIntrospection, + convert_partition_values, infer_partition_values_from_path, introspect_parquet_file, + parse_hive_partition_values, ParquetIntrospection, }; pub use types::{ DataFileFormat, DataFileInput, DataFileRegistrar, EncryptionMetadata, FileMetrics, diff --git a/src/catalog/rest/catalog_trait.rs b/src/catalog/rest/catalog_trait.rs index c8b710d..57c709b 100644 --- a/src/catalog/rest/catalog_trait.rs +++ b/src/catalog/rest/catalog_trait.rs @@ -11,6 +11,10 @@ use async_trait::async_trait; #[cfg_attr(not(target_family = "wasm"), async_trait)] #[cfg_attr(target_family = "wasm", async_trait(?Send))] impl crate::catalog::Catalog for IcebergRestCatalog { + fn file_io(&self) -> &crate::io::FileIO { + IcebergRestCatalog::file_io(self) + } + async fn create_namespace( &self, namespace: &crate::spec::NamespaceIdent, diff --git a/src/catalog/rest_catalog.rs b/src/catalog/rest_catalog.rs index ab87b43..0eb5254 100644 --- a/src/catalog/rest_catalog.rs +++ b/src/catalog/rest_catalog.rs @@ -239,6 +239,10 @@ fn map_auth_error(err: Error) -> CatalogError { #[cfg(not(target_family = "wasm"))] #[async_trait] impl Catalog for RestCatalog { + fn file_io(&self) -> &crate::io::FileIO { + self.inner.file_io() + } + async fn create_namespace( &self, namespace: &NamespaceIdent, @@ -299,6 +303,10 @@ impl Catalog for RestCatalog { #[cfg(target_family = "wasm")] #[async_trait(?Send)] impl Catalog for RestCatalog { + fn file_io(&self) -> &crate::io::FileIO { + self.inner.file_io() + } + async fn create_namespace( &self, namespace: &NamespaceIdent, diff --git a/src/catalog/s3_tables.rs b/src/catalog/s3_tables.rs index aa7f77c..1dba9bf 100644 --- a/src/catalog/s3_tables.rs +++ b/src/catalog/s3_tables.rs @@ -125,6 +125,10 @@ impl S3TablesCatalog { #[cfg(not(target_family = "wasm"))] #[async_trait] impl Catalog for S3TablesCatalog { + fn file_io(&self) -> &crate::io::FileIO { + self.inner.file_io() + } + async fn create_namespace( &self, namespace: &NamespaceIdent, diff --git a/src/cli/commands/commit/helpers.rs b/src/cli/commands/commit/helpers.rs new file mode 100644 index 0000000..6f946cc --- /dev/null +++ b/src/cli/commands/commit/helpers.rs @@ -0,0 +1,267 @@ +//! Helper functions for the commit command + +use std::collections::HashMap; +use std::path::Path; +use uuid::Uuid; + +use crate::catalog::register::{convert_partition_values, PartitionValue}; +use crate::io::{create_local_file_io, get_filename}; +use crate::spec::{PartitionField, PartitionSpec, PrimitiveType, Schema, Type}; + +/// Parse a type string into a PrimitiveType +pub fn parse_type_str(type_str: &str) -> Result { + match type_str.to_lowercase().as_str() { + "boolean" | "bool" => Ok(PrimitiveType::Boolean), + "int" | "integer" => Ok(PrimitiveType::Int), + "long" | "bigint" => Ok(PrimitiveType::Long), + "float" => Ok(PrimitiveType::Float), + "double" => Ok(PrimitiveType::Double), + "date" => Ok(PrimitiveType::Date), + "time" => Ok(PrimitiveType::Time), + "timestamp" => Ok(PrimitiveType::Timestamp), + "timestamptz" => Ok(PrimitiveType::Timestamptz), + "string" => Ok(PrimitiveType::String), + "uuid" => Ok(PrimitiveType::Uuid), + "binary" => Ok(PrimitiveType::Binary), + _ => Err(format!( + "Unknown type '{}'. Valid types: boolean, int, long, float, double, date, time, timestamp, timestamptz, string, uuid, binary", + type_str + )), + } +} + +/// Parse partition spec like "year:int,month:int" into vec of (name, type) +pub fn parse_partition_spec(spec: &str) -> Result, String> { + spec.split(',') + .map(|part| { + let part = part.trim(); + let (name, type_str) = part.split_once(':').ok_or_else(|| { + format!( + "Invalid partition spec '{}'. Expected format: name:type", + part + ) + })?; + let parsed_type = parse_type_str(type_str)?; + Ok((name.to_string(), parsed_type)) + }) + .collect() +} + +/// Parse partition values like "year=2024,month=01" into HashMap +pub fn parse_partition_values_arg(values: &str) -> Result, String> { + values + .split(',') + .map(|part| { + let part = part.trim(); + let (name, value) = part.split_once('=').ok_or_else(|| { + format!( + "Invalid partition value '{}'. Expected format: name=value", + part + ) + })?; + Ok((name.to_string(), value.to_string())) + }) + .collect() +} + +/// Expand glob pattern to list of file paths +pub fn expand_glob(pattern: &str) -> Result, String> { + let paths: Result, _> = glob::glob(pattern) + .map_err(|e| format!("Invalid glob pattern '{}': {}", pattern, e))? + .collect(); + + let paths = paths.map_err(|e| format!("Error reading files matching '{}': {}", pattern, e))?; + + let parquet_files: Vec = paths + .into_iter() + .filter(|p| p.extension().map(|e| e == "parquet").unwrap_or(false)) + .map(|p| p.to_string_lossy().to_string()) + .collect(); + + if parquet_files.is_empty() { + return Err(format!( + "No Parquet files found matching pattern: {}", + pattern + )); + } + + Ok(parquet_files) +} + +/// Build a partition spec from a spec string and schema +pub fn build_partition_spec(spec_str: &str, schema: &Schema) -> Result { + let parts = parse_partition_spec(spec_str)?; + + let fields: Vec = parts + .iter() + .enumerate() + .map(|(idx, (name, expected_type))| { + let field = schema + .fields() + .iter() + .find(|f| f.name() == name) + .ok_or_else(|| format!("Partition column '{}' not found in schema", name))?; + + match field.field_type() { + Type::Primitive(actual_type) => { + if actual_type != expected_type { + return Err(format!( + "Partition column '{}' type mismatch: specified {:?} but schema has {:?}", + name, expected_type, actual_type + )); + } + } + other => { + return Err(format!( + "Partition column '{}' must be a primitive type, got {:?}", + name, other + )); + } + } + + Ok(PartitionField::new( + 1000 + idx as i32, + field.id(), + "identity", + name.clone(), + )) + }) + .collect::, String>>()?; + + Ok(PartitionSpec::new(0, fields)) +} + +/// Check if two schemas are compatible for registration. +pub fn check_schema_compatibility(expected: &Schema, actual: &Schema) -> Result<(), String> { + if expected.fields().len() != actual.fields().len() { + return Err(format!( + "field count mismatch: expected {} fields, got {}", + expected.fields().len(), + actual.fields().len() + )); + } + + for (e, a) in expected.fields().iter().zip(actual.fields().iter()) { + if e.name() != a.name() { + return Err(format!( + "field name mismatch at position: expected '{}', got '{}'", + e.name(), + a.name() + )); + } + if e.field_type() != a.field_type() { + return Err(format!( + "field '{}' type mismatch: expected {:?}, got {:?}", + e.name(), + e.field_type(), + a.field_type() + )); + } + } + + Ok(()) +} + +/// Determine partition values for a file +pub fn determine_partition_values( + file_path: &str, + explicit_values: &Option>, + partition_spec: Option<&PartitionSpec>, + schema: &Schema, +) -> Result, String> { + use crate::catalog::register::parse_hive_partition_values; + + if let Some(explicit) = explicit_values { + return convert_partition_values(explicit, schema) + .map_err(|e| format!("Invalid partition values: {}", e)); + } + + let hive_values = parse_hive_partition_values(file_path); + + if let Some(spec) = partition_spec { + for field in spec.fields() { + if !hive_values.contains_key(field.name()) { + return Err(format!( + "Missing partition value for '{}' in path '{}'. Use --partition-values to specify.", + field.name(), + file_path + )); + } + } + } + + if hive_values.is_empty() { + return Ok(HashMap::new()); + } + + convert_partition_values(&hive_values, schema) + .map_err(|e| format!("Invalid partition values from path: {}", e)) +} + +/// Format partition values as a key string for grouping +pub fn format_partition_key(values: &HashMap) -> String { + if values.is_empty() { + return String::new(); + } + + let mut parts: Vec = values + .iter() + .map(|(k, v)| format!("{}={}", k, v.to_value_string())) + .collect(); + parts.sort(); + parts.join("/") +} + +/// Generate a remote upload path for a local file +pub fn generate_upload_path(table_location: &str, local_path: &str) -> String { + let uuid = Uuid::new_v4(); + let filename = Path::new(local_path) + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or("data"); + format!( + "{}/data/{}_{}.parquet", + table_location.trim_end_matches('/'), + filename, + uuid + ) +} + +/// Upload a local file to remote storage +pub async fn upload_local_file( + local_path: &str, + remote_path: &str, + remote_file_io: &crate::io::FileIO, +) -> Result<(), String> { + let local_file_io = create_local_file_io(local_path) + .map_err(|e| format!("Failed to create local file IO: {}", e))?; + let filename = get_filename(local_path); + let data = local_file_io + .read(filename) + .await + .map_err(|e| format!("Failed to read local file {}: {}", local_path, e))?; + remote_file_io + .write(remote_path, data) + .await + .map_err(|e| format!("Failed to upload to {}: {}", remote_path, e))?; + Ok(()) +} + +/// Introspect a Parquet file (local or remote) +pub async fn introspect_file( + path: &str, + file_io: &crate::io::FileIO, +) -> Result { + use crate::catalog::register::{introspect_local_parquet_file, introspect_parquet_file}; + use crate::io::is_local_path; + + if is_local_path(path) { + introspect_local_parquet_file(path, None) + .await + .map_err(|e| format!("Failed to read {}: {}", path, e)) + } else { + introspect_parquet_file(file_io, path, None) + .await + .map_err(|e| format!("Failed to read {}: {}", path, e)) + } +} diff --git a/src/cli/commands/commit/mod.rs b/src/cli/commands/commit/mod.rs new file mode 100644 index 0000000..05d96aa --- /dev/null +++ b/src/cli/commands/commit/mod.rs @@ -0,0 +1,386 @@ +//! Commit Parquet files to an Iceberg table + +mod helpers; +mod output; + +use std::collections::HashMap; + +use clap::Args; + +use crate::catalog::register::{DataFileInput, RegisterOptions}; +use crate::cli::catalog::CatalogConfig; +use crate::cli::output::{print, OutputFormat}; +use crate::io::{get_filename, is_local_path}; +use crate::spec::{NamespaceIdent, PartitionSpec, Schema, TableIdent}; + +use helpers::{ + determine_partition_values, expand_glob, format_partition_key, generate_upload_path, + introspect_file, upload_local_file, +}; +use output::{CommitPlanOutput, CommitResultOutput, PartitionSummary, SchemaMismatch}; + +// Re-export for tests +pub use helpers::{ + build_partition_spec, check_schema_compatibility, parse_partition_spec, + parse_partition_values_arg, parse_type_str, +}; + +/// Commit Parquet files to an Iceberg table +#[derive(Debug, Args)] +pub struct CommitArgs { + /// Glob pattern for Parquet files (e.g., /data/**/*.parquet) + pub pattern: String, + + /// Target namespace + #[arg(long, short)] + pub namespace: String, + + /// Target table name + #[arg(long, short)] + pub table: String, + + /// Parquet file to use as schema exemplar (default: first file from glob) + #[arg(long)] + pub exemplar: Option, + + /// Create table if it doesn't exist + #[arg(long)] + pub create: bool, + + /// Partition columns for new table (e.g., year:int,month:int) + #[arg(long, requires = "create")] + pub partition: Option, + + /// Explicit partition values for all files (e.g., year=2024,month=01) + #[arg(long)] + pub partition_values: Option, + + /// Show plan without committing + #[arg(long)] + pub dry_run: bool, +} + +/// Result of resolving table location for local file uploads +struct TableLocationResult { + location: String, + table_was_pre_created: bool, +} + +/// Resolve table location for local file uploads. +async fn resolve_table_location( + catalog: &dyn crate::catalog::Catalog, + namespace: &NamespaceIdent, + table_ident: &TableIdent, + schema: &Schema, + partition_spec: Option<&PartitionSpec>, + table_exists: bool, + dry_run: bool, +) -> Result { + if table_exists { + let table = catalog + .load_table(table_ident) + .await + .map_err(|e| format!("Failed to load table: {}", e))?; + return Ok(TableLocationResult { + location: table.location().to_string(), + table_was_pre_created: false, + }); + } + + let ns_name = namespace.as_ref().first().map(|s| s.as_str()).unwrap_or(""); + let table_name = table_ident.name(); + + if dry_run { + return Ok(TableLocationResult { + location: format!("s3:///{}/{}", ns_name, table_name), + table_was_pre_created: false, + }); + } + + let mut creation_builder = crate::spec::TableCreation::builder() + .with_name(table_name.to_string()) + .with_schema(schema.clone()); + + if let Some(spec) = partition_spec { + creation_builder = creation_builder.with_partition_spec(spec.clone()); + } + + let creation = creation_builder + .build() + .map_err(|e| format!("Failed to build table creation: {}", e))?; + + let table = catalog + .create_table(namespace, creation) + .await + .map_err(|e| format!("Failed to create table: {}", e))?; + + println!("Created table: {}.{}", ns_name, table_name); + Ok(TableLocationResult { + location: table.location().to_string(), + table_was_pre_created: true, + }) +} + +/// Result of processing all input files +struct ProcessedFiles { + data_files: Vec, + uploads: Vec<(String, String)>, + schema_mismatches: Vec, + partition_summaries: HashMap, + total_bytes: u64, + total_rows: i64, +} + +/// Process all input files: introspect, validate schema, extract partitions +async fn process_input_files( + files: &[String], + file_io: &crate::io::FileIO, + schema: &Schema, + explicit_partition_values: &Option>, + partition_spec: Option<&PartitionSpec>, + table_location: &str, +) -> Result { + let mut data_files: Vec = Vec::new(); + let mut schema_mismatches = Vec::new(); + let mut partition_summaries: HashMap = HashMap::new(); + let mut total_bytes = 0u64; + let mut total_rows = 0i64; + let mut uploads: Vec<(String, String)> = Vec::new(); + + for file_path in files { + let introspection = introspect_file(file_path, file_io).await?; + + if let Err(mismatch_reason) = check_schema_compatibility(schema, &introspection.schema) { + schema_mismatches.push(SchemaMismatch { + file_path: file_path.clone(), + reason: mismatch_reason, + }); + continue; + } + + let partition_values = determine_partition_values( + file_path, + explicit_partition_values, + partition_spec, + schema, + )?; + + let partition_key = format_partition_key(&partition_values); + let entry = partition_summaries.entry(partition_key).or_insert((0, 0)); + entry.0 += 1; + entry.1 += introspection.data_file.record_count; + + total_bytes += introspection.data_file.file_size_in_bytes as u64; + total_rows += introspection.data_file.record_count; + + let mut data_file = introspection.data_file; + data_file.partition_values = partition_values; + + if is_local_path(file_path) { + let remote_path = generate_upload_path(table_location, file_path); + uploads.push((file_path.clone(), remote_path.clone())); + data_file.file_path = remote_path; + } + + data_files.push(data_file); + } + + Ok(ProcessedFiles { + data_files, + uploads, + schema_mismatches, + partition_summaries, + total_bytes, + total_rows, + }) +} + +/// Upload local files to remote storage +async fn execute_uploads( + uploads: &[(String, String)], + file_io: &crate::io::FileIO, +) -> Result<(), String> { + if uploads.is_empty() { + return Ok(()); + } + + println!("Uploading {} local files...", uploads.len()); + for (local_path, remote_path) in uploads { + println!(" {} -> {}", get_filename(local_path), remote_path); + upload_local_file(local_path, remote_path, file_io).await?; + } + println!("Upload complete"); + Ok(()) +} + +/// Build partition summaries for output +fn build_partition_summaries( + partition_summaries: HashMap, +) -> Vec { + partition_summaries + .into_iter() + .map(|(k, (count, rows))| PartitionSummary { + partition_value: if k.is_empty() { + "(unpartitioned)".to_string() + } else { + k + }, + file_count: count, + row_count: rows, + }) + .collect() +} + +/// Execute the commit command +pub async fn execute( + args: CommitArgs, + config: &CatalogConfig, + format: OutputFormat, +) -> Result<(), String> { + let files = expand_glob(&args.pattern)?; + println!("Found {} Parquet files", files.len()); + + let has_local_files = files.iter().any(|f| is_local_path(f)); + if has_local_files { + println!("Detected local files - will upload to table storage"); + } + + let catalog = config.create_catalog().await?; + let file_io = catalog.file_io(); + + let exemplar_path = args.exemplar.as_ref().unwrap_or(&files[0]); + let exemplar = introspect_file(exemplar_path, file_io).await?; + let schema = exemplar.schema.clone(); + println!("Schema from: {}", exemplar_path); + + let partition_spec = args + .partition + .as_ref() + .map(|s| build_partition_spec(s, &schema)) + .transpose()?; + let explicit_partition_values = args + .partition_values + .as_ref() + .map(|pv| parse_partition_values_arg(pv)) + .transpose()?; + + if args.namespace.is_empty() { + return Err("Namespace cannot be empty".to_string()); + } + if args.table.is_empty() { + return Err("Table name cannot be empty".to_string()); + } + let namespace = NamespaceIdent::from_strs(&[args.namespace.as_str()]); + let table_ident = TableIdent::from_strs(&[args.namespace.as_str()], &args.table); + + let table_exists = catalog + .table_exists(&table_ident) + .await + .map_err(|e| format!("Failed to check if table exists: {}", e))?; + + if !table_exists && !args.create { + return Err(format!( + "Table {}.{} does not exist. Use --create to create it.", + args.namespace, args.table + )); + } + + let (table_location, table_was_pre_created) = if has_local_files { + let result = resolve_table_location( + catalog.as_ref(), + &namespace, + &table_ident, + &schema, + partition_spec.as_ref(), + table_exists, + args.dry_run, + ) + .await?; + (result.location, result.table_was_pre_created) + } else { + (String::new(), false) + }; + + let processed = process_input_files( + &files, + file_io, + &schema, + &explicit_partition_values, + partition_spec.as_ref(), + &table_location, + ) + .await?; + + if !processed.schema_mismatches.is_empty() && !args.dry_run { + return Err(format!( + "{} files have schema mismatches. Run with --dry-run to see details.", + processed.schema_mismatches.len() + )); + } + + let partitions = build_partition_summaries(processed.partition_summaries); + + if args.dry_run { + let plan = CommitPlanOutput { + schema_source: exemplar_path.clone(), + target_table: format!("{}.{}", args.namespace, args.table), + will_create_table: !table_exists, + partition_columns: partition_spec + .as_ref() + .map(|s| s.fields().iter().map(|f| f.name().to_string()).collect()) + .unwrap_or_default(), + files_to_commit: processed.data_files.len(), + files_to_upload: processed.uploads.len(), + total_rows: processed.total_rows, + total_bytes: processed.total_bytes, + partitions, + schema_mismatches: processed.schema_mismatches, + }; + print(&plan, format); + return Ok(()); + } + + execute_uploads(&processed.uploads, file_io).await?; + + let options = if args.create && !table_exists && !table_was_pre_created { + let mut opts = RegisterOptions::new().allow_create_with_schema(schema.clone()); + if let Some(spec) = partition_spec { + opts = opts.with_partition_spec(spec); + } + opts.allow_noop(true) + } else { + RegisterOptions::new().allow_noop(true) + }; + + // Clear source_schema to skip validation (catalog may assign different field IDs) + let data_files: Vec = processed + .data_files + .into_iter() + .map(|mut f| { + f.source_schema = None; + f + }) + .collect(); + + let result = crate::catalog::register::register_data_files( + catalog.as_ref(), + namespace, + table_ident, + data_files, + options, + ) + .await + .map_err(|e| format!("Commit failed: {}", e))?; + + let output = CommitResultOutput { + target_table: format!("{}.{}", args.namespace, args.table), + table_created: result.table_was_created || table_was_pre_created, + files_committed: result.added_files, + rows_committed: result.added_records, + files_skipped: result.skipped_files.len(), + snapshot_id: result.snapshot_id, + }; + + print(&output, format); + Ok(()) +} diff --git a/src/cli/commands/commit/output.rs b/src/cli/commands/commit/output.rs new file mode 100644 index 0000000..c10aabf --- /dev/null +++ b/src/cli/commands/commit/output.rs @@ -0,0 +1,130 @@ +//! Output types for the commit command + +use serde::Serialize; + +use crate::cli::output::{format_bytes, format_number, Outputable}; + +/// Commit plan output (dry run) +#[derive(Debug, Serialize)] +pub struct CommitPlanOutput { + pub schema_source: String, + pub target_table: String, + pub will_create_table: bool, + pub partition_columns: Vec, + pub files_to_commit: usize, + pub files_to_upload: usize, + pub total_rows: i64, + pub total_bytes: u64, + pub partitions: Vec, + pub schema_mismatches: Vec, +} + +#[derive(Debug, Serialize)] +pub struct PartitionSummary { + pub partition_value: String, + pub file_count: usize, + pub row_count: i64, +} + +#[derive(Debug, Serialize)] +pub struct SchemaMismatch { + pub file_path: String, + pub reason: String, +} + +impl Outputable for CommitPlanOutput { + fn to_text(&self) -> String { + let mut lines = vec![]; + + lines.push(format!("Schema source: {}", self.schema_source)); + lines.push(String::new()); + + if self.will_create_table { + lines.push(format!("Target: {} (will be created)", self.target_table)); + } else { + lines.push(format!("Target: {} (existing)", self.target_table)); + } + + if !self.partition_columns.is_empty() { + lines.push(format!( + " Partitioned by: {}", + self.partition_columns.join(", ") + )); + } + lines.push(String::new()); + + if self.files_to_upload > 0 { + lines.push(format!( + "Files to upload: {} local files", + self.files_to_upload + )); + } + + lines.push(format!( + "Files to commit: {} ({} rows, {})", + self.files_to_commit, + format_number(self.total_rows as u64), + format_bytes(self.total_bytes) + )); + + for part in &self.partitions { + lines.push(format!( + " {}: {} files, {} rows", + part.partition_value, + part.file_count, + format_number(part.row_count as u64) + )); + } + + if !self.schema_mismatches.is_empty() { + lines.push(String::new()); + lines.push(format!( + "Schema mismatches: {}", + self.schema_mismatches.len() + )); + for mismatch in &self.schema_mismatches { + lines.push(format!(" {}: {}", mismatch.file_path, mismatch.reason)); + } + } + + lines.push(String::new()); + lines.push("Run without --dry-run to commit.".to_string()); + + lines.join("\n") + } +} + +/// Commit result output +#[derive(Debug, Serialize)] +pub struct CommitResultOutput { + pub target_table: String, + pub table_created: bool, + pub files_committed: usize, + pub rows_committed: i64, + pub files_skipped: usize, + pub snapshot_id: i64, +} + +impl Outputable for CommitResultOutput { + fn to_text(&self) -> String { + let mut lines = vec![]; + + if self.table_created { + lines.push(format!("Created table: {}", self.target_table)); + } else { + lines.push(format!("Committed to: {}", self.target_table)); + } + + lines.push(format!( + " Files: {} committed, {} skipped", + self.files_committed, self.files_skipped + )); + lines.push(format!( + " Rows: {}", + format_number(self.rows_committed as u64) + )); + lines.push(format!(" Snapshot: {}", self.snapshot_id)); + + lines.join("\n") + } +} diff --git a/src/cli/commands/mod.rs b/src/cli/commands/mod.rs index 7d653e7..4ff1f47 100644 --- a/src/cli/commands/mod.rs +++ b/src/cli/commands/mod.rs @@ -1,6 +1,7 @@ //! CLI commands pub mod catalog; +pub mod commit; pub mod compact; pub mod namespace; pub mod snapshot; diff --git a/src/io/local.rs b/src/io/local.rs new file mode 100644 index 0000000..71f3d07 --- /dev/null +++ b/src/io/local.rs @@ -0,0 +1,77 @@ +//! Local filesystem utilities for the CLI +//! +//! This module provides helpers for working with local Parquet files, +//! including detection of local paths and creating FileIO instances +//! for local filesystem operations. + +use crate::error::{Error, Result}; +use crate::io::FileIO; +use opendal::Operator; + +/// Check if a path is a local filesystem path (has no URI scheme) +pub fn is_local_path(path: &str) -> bool { + !path.contains("://") +} + +/// Create a FileIO for local filesystem operations +/// +/// Creates an OpenDAL Fs operator rooted at the parent directory of the given path. +/// This allows reading the file using just its filename. +/// +/// # Arguments +/// * `path` - Absolute path to a local file +/// +/// # Returns +/// A FileIO configured for local filesystem access +pub fn create_local_file_io(path: &str) -> Result { + use opendal::services::Fs; + use std::path::Path; + + let file_path = Path::new(path); + let root = file_path.parent().and_then(|p| p.to_str()).unwrap_or("/"); + + let builder = Fs::default().root(root); + let operator = Operator::new(builder) + .map_err(|e| Error::IoError(format!("Failed to create local operator: {}", e)))? + .finish(); + + Ok(FileIO::new(operator)) +} + +/// Get the filename portion of a path +/// +/// # Arguments +/// * `path` - A file path (local or remote) +/// +/// # Returns +/// The filename component of the path +pub fn get_filename(path: &str) -> &str { + std::path::Path::new(path) + .file_name() + .and_then(|f| f.to_str()) + .unwrap_or(path) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_is_local_path() { + assert!(is_local_path("/path/to/file.parquet")); + assert!(is_local_path("./relative/file.parquet")); + assert!(is_local_path("file.parquet")); + assert!(!is_local_path("s3://bucket/file.parquet")); + assert!(!is_local_path("https://example.com/file.parquet")); + } + + #[test] + fn test_get_filename() { + assert_eq!(get_filename("/path/to/file.parquet"), "file.parquet"); + assert_eq!(get_filename("file.parquet"), "file.parquet"); + assert_eq!( + get_filename("s3://bucket/path/file.parquet"), + "file.parquet" + ); + } +} diff --git a/src/io/mod.rs b/src/io/mod.rs index e4e4aff..1c2e839 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -2,5 +2,9 @@ //! WASM-compatible via OpenDAL mod file_io; +#[cfg(not(target_family = "wasm"))] +pub mod local; pub use file_io::{AwsCredentials, FileIO, VendedCredentialProvider, VendedCredentials}; +#[cfg(not(target_family = "wasm"))] +pub use local::{create_local_file_io, get_filename, is_local_path}; diff --git a/tests/commit_command.rs b/tests/commit_command.rs new file mode 100644 index 0000000..2be5089 --- /dev/null +++ b/tests/commit_command.rs @@ -0,0 +1,141 @@ +//! Tests for the commit command +//! +//! Integration tests require a running catalog and are marked #[ignore]. +//! Run with: cargo test --test commit_command -- --ignored + +use icepick::cli::commands::commit::{ + build_partition_spec, check_schema_compatibility, parse_partition_spec, + parse_partition_values_arg, parse_type_str, +}; +use icepick::spec::{NestedField, PrimitiveType, Schema, Type}; +use std::process::Command; + +fn schema(fields: &[(&str, PrimitiveType)]) -> Schema { + Schema::builder() + .with_fields( + fields + .iter() + .enumerate() + .map(|(i, (n, t))| { + NestedField::required_field( + (i + 1) as i32, + n.to_string(), + Type::Primitive(t.clone()), + ) + }) + .collect(), + ) + .build() + .unwrap() +} + +#[test] +fn test_parse_partition_spec() { + let result = parse_partition_spec("year:int,month:int").unwrap(); + assert_eq!( + result, + vec![ + ("year".into(), PrimitiveType::Int), + ("month".into(), PrimitiveType::Int) + ] + ); + // Test type aliases + assert_eq!(parse_type_str("bool").unwrap(), PrimitiveType::Boolean); + assert_eq!(parse_type_str("integer").unwrap(), PrimitiveType::Int); + assert_eq!(parse_type_str("bigint").unwrap(), PrimitiveType::Long); + // Error cases + assert!(parse_partition_spec("year:invalid") + .unwrap_err() + .contains("Unknown type")); + assert!(parse_partition_spec("year") + .unwrap_err() + .contains("Expected format")); +} + +#[test] +fn test_parse_partition_values() { + let result = parse_partition_values_arg("year=2024,month=01").unwrap(); + assert_eq!(result.get("year"), Some(&"2024".to_string())); + assert_eq!(result.get("month"), Some(&"01".to_string())); +} + +#[test] +fn test_check_schema_compatibility() { + let s1 = schema(&[("id", PrimitiveType::Long), ("name", PrimitiveType::String)]); + let s2 = schema(&[("id", PrimitiveType::Long), ("name", PrimitiveType::String)]); + assert!(check_schema_compatibility(&s1, &s2).is_ok()); + // Field count mismatch + let s3 = schema(&[("id", PrimitiveType::Long)]); + assert!(check_schema_compatibility(&s1, &s3) + .unwrap_err() + .contains("field count")); + // Field name mismatch + let s4 = schema(&[ + ("user_id", PrimitiveType::Long), + ("name", PrimitiveType::String), + ]); + assert!(check_schema_compatibility(&s1, &s4) + .unwrap_err() + .contains("field name")); + // Field type mismatch + let s5 = schema(&[("id", PrimitiveType::Int), ("name", PrimitiveType::String)]); + assert!(check_schema_compatibility(&s1, &s5) + .unwrap_err() + .contains("type mismatch")); +} + +#[test] +fn test_build_partition_spec() { + let s = schema(&[ + ("id", PrimitiveType::Long), + ("year", PrimitiveType::Int), + ("month", PrimitiveType::Int), + ]); + let result = build_partition_spec("year:int,month:int", &s).unwrap(); + assert_eq!(result.fields().len(), 2); + assert_eq!(result.fields()[0].name(), "year"); + // Column not found + let s2 = schema(&[("id", PrimitiveType::Long)]); + assert!(build_partition_spec("year:int", &s2) + .unwrap_err() + .contains("not found")); + // Type mismatch + let s3 = schema(&[("year", PrimitiveType::String)]); + assert!(build_partition_spec("year:int", &s3) + .unwrap_err() + .contains("type mismatch")); +} + +#[test] +#[ignore] +fn test_commit_dry_run() { + let output = Command::new("cargo") + .args([ + "run", + "--features", + "cli", + "--", + "commit", + "/tmp/test-data/**/*.parquet", + "--namespace", + "test", + "--table", + "events", + "--dry-run", + ]) + .output() + .expect("Failed to execute command"); + + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + + println!("stdout: {}", stdout); + println!("stderr: {}", stderr); + + // Should either succeed with a plan or fail with "no files found" + assert!( + output.status.success() || stderr.contains("No Parquet files found"), + "Command failed unexpectedly: {}", + stderr + ); +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs index b77c2b5..5703f61 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -4,6 +4,7 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use async_trait::async_trait; use icepick::catalog::Catalog; use icepick::error::{Error, Result}; +use icepick::io::FileIO; use icepick::spec::{NamespaceIdent, TableCreation, TableIdent}; use icepick::table::Table; use tokio::sync::{Mutex, RwLock}; @@ -11,6 +12,7 @@ use tokio::sync::{Mutex, RwLock}; /// Simple in-memory catalog used by integration tests pub struct TestCatalog { table: RwLock, + file_io: FileIO, updates: Mutex>, fail_next_update: AtomicBool, load_calls: AtomicUsize, @@ -18,8 +20,11 @@ pub struct TestCatalog { impl TestCatalog { pub fn new(table: Table) -> Self { + // Use the table's FileIO for this catalog + let file_io = table.file_io().clone(); Self { table: RwLock::new(table), + file_io, updates: Mutex::new(Vec::new()), fail_next_update: AtomicBool::new(false), load_calls: AtomicUsize::new(0), @@ -46,6 +51,10 @@ impl TestCatalog { #[async_trait] impl Catalog for TestCatalog { + fn file_io(&self) -> &FileIO { + &self.file_io + } + async fn create_namespace( &self, _namespace: &NamespaceIdent, diff --git a/tests/integration_commit_test.rs b/tests/integration_commit_test.rs index 9b7d967..467eb21 100644 --- a/tests/integration_commit_test.rs +++ b/tests/integration_commit_test.rs @@ -9,10 +9,22 @@ use opendal::Operator; use std::collections::HashMap; // Simple in-memory catalog for testing -struct TestCatalog; +struct TestCatalog { + file_io: FileIO, +} + +impl TestCatalog { + fn new(file_io: FileIO) -> Self { + Self { file_io } + } +} #[async_trait::async_trait] impl Catalog for TestCatalog { + fn file_io(&self) -> &FileIO { + &self.file_io + } + async fn create_namespace( &self, _namespace: &NamespaceIdent, @@ -118,7 +130,7 @@ async fn test_end_to_end_commit_with_stats() { .unwrap(); // Commit - let catalog = TestCatalog; + let catalog = TestCatalog::new(file_io.clone()); let timestamp_ms = 1234567890; table .transaction() @@ -213,7 +225,7 @@ async fn test_multiple_sequential_commits() { ); // First commit - let catalog = TestCatalog; + let catalog = TestCatalog::new(file_io.clone()); let data_file1 = DataFile::builder() .with_file_path("memory://warehouse/test/multi/data/file1.parquet") .with_file_format("PARQUET") diff --git a/tests/register_test.rs b/tests/register_test.rs index a234cc1..7453364 100644 --- a/tests/register_test.rs +++ b/tests/register_test.rs @@ -9,18 +9,25 @@ use tokio::sync::RwLock; struct RefreshingCatalog { table: RwLock, + file_io: FileIO, } impl RefreshingCatalog { fn new(table: icepick::table::Table) -> Self { + let file_io = table.file_io().clone(); Self { table: RwLock::new(table), + file_io, } } } #[async_trait::async_trait] impl icepick::catalog::Catalog for RefreshingCatalog { + fn file_io(&self) -> &FileIO { + &self.file_io + } + async fn create_namespace( &self, _namespace: &NamespaceIdent, diff --git a/tests/test_table_writer.rs b/tests/test_table_writer.rs index 055c9c7..dc4d194 100644 --- a/tests/test_table_writer.rs +++ b/tests/test_table_writer.rs @@ -35,6 +35,10 @@ impl SimpleCatalog { #[async_trait::async_trait] impl Catalog for SimpleCatalog { + fn file_io(&self) -> &icepick::io::FileIO { + &self.file_io + } + async fn create_namespace( &self, _namespace: &NamespaceIdent,