added dedup function

Author: jsai28

README.md

@@ -28,6 +28,7 @@ SUBCOMMANDS:
     cat         Concatenate multiple files or all files in a directory
     convert     Convert file format (CSV, Parquet, JSON)
     count       Count the number of rows in a file
+    dedup       Remove duplicate rows
     describe    Show summary statistics for a file
     help        Prints this message or the help of the given subcommand(s)
     query       Run a SQL query on a file

src/bin/main.rs

@@ -1,5 +1,5 @@
 use datafusion::prelude::*;
-use dfkit::commands::{cat, convert, count, describe, dfsplit, query, reverse, schema, sort, view};
+use dfkit::commands::{cat, convert, count, describe, dfsplit, query, reverse, schema, sort, view, dedup};
 use dfkit::utils::{DfKitError, parse_file_list};
 use std::env;
 use std::path::PathBuf;
@@ -101,6 +101,14 @@ pub enum Commands {
         #[structopt(short, long, parse(from_os_str))]
         output: PathBuf,
     },
+
+    #[structopt(about = "Remove duplicate rows")]
+    Dedup {
+        #[structopt(short, long, parse(from_os_str))]
+        filename: PathBuf,
+        #[structopt(short, long, parse(from_os_str))]
+        output: Option<PathBuf>,
+    }
 }
 
 #[tokio::main]
@@ -158,6 +166,9 @@ async fn main() -> Result<(), DfKitError> {
             let file_list = parse_file_list(files, dir)?;
             cat(&ctx, file_list, &output).await?;
         }
+        Commands::Dedup { filename, output } => {
+            dedup(&ctx, &filename, output).await?;
+        }
     }
 
     Ok(())

src/commands.rs

@@ -218,3 +218,22 @@ pub async fn cat(
 
     Ok(())
 }
+
+pub async fn dedup(
+    ctx: &SessionContext,
+    filename: &Path,
+    output: Option<PathBuf>,
+) -> Result<(), DfKitError> {
+    let _ = register_table(&ctx, "t", &filename).await?;
+    let df = ctx.sql("SELECT DISTINCT * FROM t").await?;
+
+    if let Some(out_path) = output {
+        let file_type = file_type(&filename)?;
+        write_output(df, &out_path, &file_type).await?;
+        println!("Deduplicated file written to: {}", out_path.display());
+    } else {
+        df.show().await?;
+    }
+
+    Ok(())
+}

tests/test.rs

@@ -376,3 +376,30 @@ fn test_url_file_split_to_local() {
 
     assert_eq!(files.len(), 5);
 }
+
+
+#[test]
+fn test_dedup_csv_file() {
+    let temp = tempdir().unwrap();
+    let input_path = create_extended_csv(temp.path());
+    fs::write(&input_path, "name,age\nalice,30\nbob,40\ncharlie,25\nalice,30").unwrap();
+
+    let mut cmd = Command::cargo_bin("dfkit").unwrap();
+    let output = cmd
+        .args(["dedup", "--filename", input_path.to_str().unwrap()])
+        .assert()
+        .success()
+        .get_output()
+        .stdout
+        .clone();
+
+    assert_snapshot!(String::from_utf8(output).unwrap(), @r"
+    +---------+-----+
+    | name    | age |
+    +---------+-----+
+    | bob     | 40  |
+    | charlie | 25  |
+    | alice   | 30  |
+    +---------+-----+
+    ");
+}