Merge pull request #6 from OrlovEvgeny/feat/add_spinner

feat: add spinner progress indicator to all commands

Author: OrlovEvgeny

src/commands/cat.rs

@@ -1,17 +1,20 @@
 use crate::cli::CatArgs;
-use crate::input::resolve_inputs_with_config;
+use crate::input::resolve_inputs_report;
 use crate::output::OutputConfig;
 use arrow::array::{ArrayRef, RecordBatch, StringArray, UInt32Array, new_null_array};
 use arrow::compute::{SortColumn, SortOptions, concat_batches, lexsort_to_indices, take};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
-use indicatif::{ProgressBar, ProgressStyle};
 use parquet::arrow::ArrowWriter;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use std::fs::File;
 use std::sync::Arc;
 
 pub fn execute(args: &CatArgs, output: &mut OutputConfig) -> miette::Result<()> {
-    let sources = resolve_inputs_with_config(&args.files, &output.cloud_config)?;
+    let sp = output.spinner("Loading");
+    let sources = resolve_inputs_report(&args.files, &output.cloud_config, &mut |msg| {
+        sp.set_message(msg);
+    })?;
+    sp.finish_and_clear();
     let out_path = output
         .output_path()
         .ok_or_else(|| miette::miette!("cat requires an output file (-o <path>)"))?
@@ -23,7 +26,6 @@ pub fn execute(args: &CatArgs, output: &mut OutputConfig) -> miette::Result<()>
     }
 
     let output_schema: SchemaRef = if args.union_by_name {
-        // unified schema across all files, preserving column order
         let mut fields: Vec<Arc<Field>> = Vec::new();
         let mut seen_names: Vec<String> = Vec::new();
 
@@ -44,7 +46,6 @@ pub fn execute(args: &CatArgs, output: &mut OutputConfig) -> miette::Result<()>
 
         Arc::new(Schema::new(fields))
     } else {
-        // Use schema from first file
         let first_file = File::open(sources[0].path())
             .map_err(|e| miette::miette!("cannot open '{}': {}", sources[0].display_name(), e))?;
         let first_builder = ParquetRecordBatchReaderBuilder::try_new(first_file)
@@ -63,17 +64,7 @@ pub fn execute(args: &CatArgs, output: &mut OutputConfig) -> miette::Result<()>
     let mut all_batches: Vec<RecordBatch> = Vec::new();
     let mut total_rows: usize = 0;
 
-    let spinner = if output.is_tty && !output.quiet && sources.len() > 1 {
-        let pb = ProgressBar::new(sources.len() as u64);
-        pb.set_style(
-            ProgressStyle::default_bar()
-                .template("  Concatenating [{bar:20}] {pos}/{len} files")
-                .unwrap(),
-        );
-        Some(pb)
-    } else {
-        None
-    };
+    let progress = output.progress_bar("Concatenating", sources.len() as u64);
 
     for source in &sources {
         let file = File::open(source.path())
@@ -115,13 +106,9 @@ pub fn execute(args: &CatArgs, output: &mut OutputConfig) -> miette::Result<()>
 
             all_batches.push(final_batch);
         }
-        if let Some(ref pb) = spinner {
-            pb.inc(1);
-        }
-    }
-    if let Some(ref pb) = spinner {
-        pb.finish_and_clear();
+        progress.inc(1);
     }
+    drop(progress);
 
     let write_batches: Vec<RecordBatch> = if let Some(ref sort_by) = args.sort_by {
         if all_batches.is_empty() {
@@ -189,7 +176,6 @@ pub fn execute(args: &CatArgs, output: &mut OutputConfig) -> miette::Result<()>
     Ok(())
 }
 
-/// Align a batch to a target schema by reordering columns and filling missing ones with nulls.
 fn align_batch_to_schema(
     batch: &RecordBatch,
     file_schema: &SchemaRef,
@@ -207,7 +193,6 @@ fn align_batch_to_schema(
         {
             columns.push(batch.column(idx).clone());
         } else {
-            // Column not present in this file — fill with nulls
             columns.push(new_null_array(target_field.data_type(), num_rows));
         }
     }
@@ -216,7 +201,6 @@ fn align_batch_to_schema(
         .map_err(|e| miette::miette!("failed to build aligned batch: {}", e))
 }
 
-/// Add a `_filename` column to a batch.
 fn add_filename_column(
     batch: &RecordBatch,
     filename: &str,

src/commands/check.rs

@@ -1,7 +1,7 @@
 use crate::cli::CheckArgs;
 use crate::error::PqError;
 use crate::input::cloud::is_cloud_url;
-use crate::input::{resolve_inputs, resolve_inputs_with_config};
+use crate::input::{resolve_inputs, resolve_inputs_report};
 use crate::output::{OutputConfig, OutputFormat};
 use crate::parquet_ext::metadata;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
@@ -75,7 +75,12 @@ fn parse_size_bytes(s: &str) -> Option<u64> {
 
 pub fn execute(args: &CheckArgs, output: &mut OutputConfig) -> miette::Result<()> {
     let sources = if args.files.iter().any(|f| is_cloud_url(f)) {
-        resolve_inputs_with_config(&args.files, &output.cloud_config)?
+        let sp = output.spinner("Loading");
+        let s = resolve_inputs_report(&args.files, &output.cloud_config, &mut |msg| {
+            sp.set_message(msg);
+        })?;
+        sp.finish_and_clear();
+        s
     } else {
         resolve_inputs(&args.files)?
     };
@@ -214,7 +219,10 @@ pub fn execute(args: &CheckArgs, output: &mut OutputConfig) -> miette::Result<()
 
             if args.read_data {
                 let expected_rows = metadata::total_rows(meta);
-                match validate_data(source.path()) {
+                let sp = output.spinner("Validating data pages");
+                let validate_result = validate_data(source.path());
+                sp.finish_and_clear();
+                match validate_result {
                     Ok(actual_rows) => {
                         checks.push(CheckItem {
                             name: "All pages decompressible".to_string(),

src/commands/compact.rs

@@ -1,10 +1,9 @@
 use crate::cli::CompactArgs;
 use crate::error::PqError;
-use crate::input::resolve_inputs_with_config;
+use crate::input::resolve_inputs_report;
 use crate::output::{OutputConfig, OutputFormat};
 use crate::parquet_ext::metadata;
 use arrow::record_batch::RecordBatch;
-use indicatif::{ProgressBar, ProgressStyle};
 use parquet::arrow::ArrowWriter;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use parquet::basic::Compression;
@@ -60,25 +59,12 @@ fn resolve_compression(args: &CompactArgs) -> Compression {
     }
 }
 
-/// Create an optional progress bar. Returns `None` when stdout is not a TTY
-/// (piped output) so that machine-readable output stays clean.
-fn make_progress_bar(output: &OutputConfig, total: u64) -> Option<ProgressBar> {
-    if !output.is_tty || output.quiet {
-        return None;
-    }
-
-    let pb = ProgressBar::new(total);
-    pb.set_style(
-        ProgressStyle::default_bar()
-            .template("{spinner:.green} [{bar:40.cyan/blue}] {pos}/{len} groups ({eta})")
-            .unwrap_or_else(|_| ProgressStyle::default_bar())
-            .progress_chars("##-"),
-    );
-    Some(pb)
-}
-
 pub fn execute(args: &CompactArgs, output: &mut OutputConfig) -> miette::Result<()> {
-    let sources = resolve_inputs_with_config(&args.files, &output.cloud_config)?;
+    let sp = output.spinner("Loading");
+    let sources = resolve_inputs_report(&args.files, &output.cloud_config, &mut |msg| {
+        sp.set_message(msg);
+    })?;
+    sp.finish_and_clear();
     if sources.is_empty() {
         return Err(miette::miette!("no files to compact"));
     }
@@ -253,8 +239,7 @@ pub fn execute(args: &CompactArgs, output: &mut OutputConfig) -> miette::Result<
             .map_err(|e| miette::miette!("cannot create output directory '{}': {}", out_dir, e))?;
     }
 
-    // ── Progress bar ──────────────────────────────────────────────────
-    let progress = make_progress_bar(output, groups.len() as u64);
+    let progress = output.progress_bar("Compacting", groups.len() as u64);
 
     // ── Execute compaction ────────────────────────────────────────────
     let mut succeeded: usize = 0;
@@ -285,9 +270,7 @@ pub fn execute(args: &CompactArgs, output: &mut OutputConfig) -> miette::Result<
                     }
                 }
                 succeeded += 1;
-                if let Some(ref pb) = progress {
-                    pb.inc(1);
-                }
+                progress.inc(1);
                 writeln!(
                     output.writer,
                     "  Group {}: {} files -> {}",
@@ -299,18 +282,14 @@ pub fn execute(args: &CompactArgs, output: &mut OutputConfig) -> miette::Result<
             }
             Err(e) => {
                 failed += 1;
-                if let Some(ref pb) = progress {
-                    pb.inc(1);
-                }
+                progress.inc(1);
                 writeln!(output.writer, "  Group {}: FAILED — {}", group_idx + 1, e)
                     .map_err(|e| miette::miette!("{}", e))?;
             }
         }
     }
 
-    if let Some(ref pb) = progress {
-        pb.finish_and_clear();
-    }
+    drop(progress);
 
     writeln!(
         output.writer,

src/commands/convert.rs

@@ -4,7 +4,6 @@ use arrow::csv as arrow_csv;
 use arrow::datatypes::Schema;
 use arrow::ipc;
 use arrow::json as arrow_json;
-use indicatif::{ProgressBar, ProgressStyle};
 use parquet::arrow::ArrowWriter;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use parquet::basic::Compression;
@@ -66,24 +65,10 @@ pub fn execute(args: &ConvertArgs, output: &mut OutputConfig) -> miette::Result<
     } else {
         let target = crate::input::cloud::resolve_output_path(&out_path)?;
         let input = &args.files[0];
-        let spinner = if output.is_tty && !output.quiet {
-            let pb = ProgressBar::new_spinner();
-            pb.set_style(
-                ProgressStyle::default_spinner()
-                    .template("{spinner} Converting {msg}...")
-                    .unwrap(),
-            );
-            pb.set_message(input.clone());
-            pb.enable_steady_tick(std::time::Duration::from_millis(100));
-            Some(pb)
-        } else {
-            None
-        };
+        let sp = output.spinner("Converting");
         convert_single(input, target.local_path(), args)?;
         target.finalize(&output.cloud_config)?;
-        if let Some(pb) = spinner {
-            pb.finish_and_clear();
-        }
+        sp.finish_and_clear();
         eprintln!("Converted {} → {}", input, out_path);
     }
 

src/commands/count.rs

@@ -1,5 +1,5 @@
 use crate::cli::CountArgs;
-use crate::input::resolve_inputs_with_config;
+use crate::input::resolve_inputs_report;
 use crate::output::{OutputConfig, OutputFormat};
 use crate::parquet_ext::metadata;
 use num_format::{Locale, ToFormattedString};
@@ -12,7 +12,11 @@ struct CountResult {
 }
 
 pub fn execute(args: &CountArgs, output: &mut OutputConfig) -> miette::Result<()> {
-    let sources = resolve_inputs_with_config(&args.files, &output.cloud_config)?;
+    let sp = output.spinner("Loading");
+    let sources = resolve_inputs_report(&args.files, &output.cloud_config, &mut |msg| {
+        sp.set_message(msg);
+    })?;
+    sp.finish_and_clear();
     let single_file = sources.len() == 1;
 
     let mut results: Vec<CountResult> = Vec::new();

src/commands/diff.rs

@@ -1,5 +1,5 @@
 use crate::cli::DiffArgs;
-use crate::input::resolve_inputs_with_config;
+use crate::input::resolve_inputs_report;
 use crate::output::table;
 use crate::output::{OutputConfig, OutputFormat};
 use crate::parquet_ext::metadata;
@@ -56,7 +56,11 @@ struct DataChange {
 }
 
 pub fn execute(args: &DiffArgs, output: &mut OutputConfig) -> miette::Result<()> {
-    let sources = resolve_inputs_with_config(&args.files, &output.cloud_config)?;
+    let sp = output.spinner("Loading");
+    let sources = resolve_inputs_report(&args.files, &output.cloud_config, &mut |msg| {
+        sp.set_message(msg);
+    })?;
+    sp.finish_and_clear();
     if sources.len() < 2 {
         return Err(miette::miette!(
             "diff requires exactly 2 files (got {})",
@@ -149,7 +153,10 @@ pub fn execute(args: &DiffArgs, output: &mut OutputConfig) -> miette::Result<()>
     };
 
     let data_diff = if args.data {
-        Some(compute_data_diff(args, &sources)?)
+        let sp = output.spinner("Comparing data");
+        let result = compute_data_diff(args, &sources)?;
+        sp.finish_and_clear();
+        Some(result)
     } else {
         None
     };

src/commands/head.rs

@@ -1,6 +1,6 @@
 use crate::cli::HeadArgs;
 use crate::input::column_selector::resolve_projection;
-use crate::input::resolve_inputs_with_config;
+use crate::input::resolve_inputs_report;
 use crate::output::{OutputConfig, OutputFormat};
 use arrow::array::{Array, ArrayRef, AsArray, StringArray};
 use arrow::datatypes::DataType;
@@ -14,7 +14,11 @@ use std::path::Path;
 use std::sync::Arc;
 
 pub fn execute(args: &HeadArgs, output: &mut OutputConfig) -> miette::Result<()> {
-    let sources = resolve_inputs_with_config(&args.files, &output.cloud_config)?;
+    let sp = output.spinner("Loading");
+    let sources = resolve_inputs_report(&args.files, &output.cloud_config, &mut |msg| {
+        sp.set_message(msg);
+    })?;
+    sp.finish_and_clear();
 
     for source in &sources {
         let file = File::open(source.path())

src/commands/inspect.rs

@@ -1,5 +1,5 @@
 use crate::cli::InspectArgs;
-use crate::input::resolve_inputs_with_config;
+use crate::input::resolve_inputs_report;
 use crate::output::table;
 use crate::output::{OutputConfig, OutputFormat};
 use crate::parquet_ext::metadata;
@@ -35,7 +35,11 @@ struct SchemaColumn {
 }
 
 pub fn execute(args: &InspectArgs, output: &mut OutputConfig) -> miette::Result<()> {
-    let sources = resolve_inputs_with_config(&args.files, &output.cloud_config)?;
+    let sp = output.spinner("Loading");
+    let sources = resolve_inputs_report(&args.files, &output.cloud_config, &mut |msg| {
+        sp.set_message(msg);
+    })?;
+    sp.finish_and_clear();
 
     if args.raw {
         for source in &sources {
@@ -302,6 +306,7 @@ fn execute_multi_table(
     let mut mismatch_count: usize = 0;
     let mut first_num_columns: Option<usize> = None;
 
+    let progress = output.progress_bar("Inspecting", sources.len() as u64);
     for source in sources {
         let meta = metadata::read_metadata(source.path())?;
         let file_size = source.file_size();
@@ -341,7 +346,9 @@ fn execute_multi_table(
             compression,
             created_by,
         ]);
+        progress.inc(1);
     }
+    drop(progress);
 
     writeln!(
         output.writer,