Daily Tool

use oracle::{Connection, Error, ErrorKind};

#[derive(Debug)]
struct Employee {
    id: i64,
    name: String,
    salary: f64,
}

fn connect() -> Result<Connection, Error> {
    // username, password, connect string ("host:port/service_name" or a TNS alias)
    Connection::connect("username", "password", "TEST2")
}

fn create_table(conn: &Connection) -> Result<(), Error> {
    // Note: IDENTITY columns require COMPATIBLE >= 12.2, which not all
    // Oracle instances (e.g. older XE setups) have. Using a sequence +
    // trigger instead works on every Oracle version (9i and up).
    let statements = [
        "BEGIN
             EXECUTE IMMEDIATE 'CREATE TABLE employees (
                 id     NUMBER PRIMARY KEY,
                 name   VARCHAR2(100) NOT NULL,
                 salary NUMBER(10,2)
             )';
         EXCEPTION
             WHEN OTHERS THEN
                 IF SQLCODE != -955 THEN -- ORA-00955: name already used
                     RAISE;
                 END IF;
         END;",
        "BEGIN
             EXECUTE IMMEDIATE 'CREATE SEQUENCE employees_seq START WITH 1 INCREMENT BY 1';
         EXCEPTION
             WHEN OTHERS THEN
                 IF SQLCODE != -955 THEN -- ORA-00955: name already used
                     RAISE;
                 END IF;
         END;",
        "BEGIN
             EXECUTE IMMEDIATE '
                 CREATE OR REPLACE TRIGGER employees_bir
                 BEFORE INSERT ON employees
                 FOR EACH ROW
                 WHEN (NEW.id IS NULL)
                 BEGIN
                     SELECT employees_seq.NEXTVAL INTO :NEW.id FROM dual;
                 END;';
         END;",
    ];

    for sql in statements {
        conn.execute(sql, &[])?;
    }
    Ok(())
}

// ---------- CREATE ----------
fn create_employee(conn: &Connection, name: &str, salary: f64) -> Result<i64, Error> {
    let stmt = conn.execute(
        "INSERT INTO employees (name, salary) VALUES (:1, :2) RETURNING id INTO :3",
        &[&name, &salary, &None::<i64>],
    )?;
    conn.commit()?;
    let ids: Vec<i64> = stmt.returned_values(3)?;
    Ok(ids[0])
}

// ---------- READ (one) ----------
fn get_employee(conn: &Connection, id: i64) -> Result<Option<Employee>, Error> {
    let row = conn.query_row_as::<(i64, String, f64)>(
        "SELECT id, name, salary FROM employees WHERE id = :1",
        &[&id],
    );

    match row {
        Ok((id, name, salary)) => Ok(Some(Employee { id, name, salary })),
        Err(e) if e.kind() == ErrorKind::NoDataFound => Ok(None),
        Err(e) => Err(e),
    }
}

// ---------- READ (all) ----------
fn get_all_employees(conn: &Connection) -> Result<Vec<Employee>, Error> {
    let mut employees = Vec::new();
    let rows = conn.query_as::<(i64, String, f64)>(
        "SELECT id, name, salary FROM employees ORDER BY id",
        &[],
    )?;

    for row_result in rows {
        let (id, name, salary) = row_result?;
        employees.push(Employee { id, name, salary });
    }
    Ok(employees)
}

// ---------- UPDATE ----------
fn update_employee_salary(conn: &Connection, id: i64, new_salary: f64) -> Result<u64, Error> {
    let stmt = conn.execute(
        "UPDATE employees SET salary = :1 WHERE id = :2",
        &[&new_salary, &id],
    )?;
    conn.commit()?;
    Ok(stmt.row_count()?)
}

// ---------- DELETE ----------
fn delete_employee(conn: &Connection, id: i64) -> Result<u64, Error> {
    let stmt = conn.execute("DELETE FROM employees WHERE id = :1", &[&id])?;
    conn.commit()?;
    Ok(stmt.row_count()?)
}

fn main() -> Result<(), Error> {
    let conn = connect()?;
    create_table(&conn)?;

    // Create
    let id = create_employee(&conn, "Alice Johnson", 75000.0)?;
    println!("Inserted employee with id = {}", id);

    let id2 = create_employee(&conn, "Bob Smith", 62000.0)?;
    println!("Inserted employee with id = {}", id2);

    // Read one
    if let Some(emp) = get_employee(&conn, id)? {
        println!("Fetched: {:?}", emp);
    }

    // Read all
    println!("All employees:");
    for emp in get_all_employees(&conn)? {
        println!("  {:?}", emp);
    }

    // Update
    let updated = update_employee_salary(&conn, id, 80000.0)?;
    println!("Rows updated: {}", updated);

    // Delete
    let deleted = delete_employee(&conn, id2)?;
    println!("Rows deleted: {}", deleted);

    Ok(())
}

//! Read a CSV with Polars and load it into Oracle.
//!
//! Mirrors the logic of the earlier Python/pandas version:
//!   * every column is read as text (like pandas `dtype='str'`)
//!   * identifiers are truncated to 30 chars (pre-12.2 Oracle limit)
//!   * each column is sized by its max UTF-8 *byte* length:
//!       <= 4000 bytes  -> VARCHAR2(n)     (> 4000 -> CLOB)
//!   * rows are inserted with a single batched (array) INSERT
//!
//! Runtime requirement: Oracle Instant Client must be installed and findable
//! (LD_LIBRARY_PATH / PATH / DYLD_LIBRARY_PATH depending on OS).
//!
//! Usage:
//!   csv2oracle <csv_path> <table_name> [separator]
//! Connection is taken from env vars ORA_USER, ORA_PASS, ORA_DSN.

use std::time::Instant;

use anyhow::{Context, Result, anyhow};
use oracle::Connection;
use polars::prelude::*;

const VARCHAR2_MAX_BYTES: usize = 4000;
const BATCH_SIZE: usize = 10_000;

fn main() -> Result<()> {
    let start = Instant::now();

    let mut args = std::env::args().skip(1);
    let csv_path = args.next().ok_or_else(|| anyhow!("missing <csv_path>"))?;
    let table = args.next().ok_or_else(|| anyhow!("missing <table_name>"))?;
    let sep = match args.next().as_deref() {
        Some("tab") | Some("\\t") => b'\t',
        Some(s) if s.len() == 1 => s.as_bytes()[0],
        Some(other) => {
            return Err(anyhow!(
                "separator must be a single char or 'tab', got {other:?}"
            ));
        }
        None => b',',
    };

    let user = std::env::var("ORA_USER").context("set ORA_USER")?;
    let pass = std::env::var("ORA_PASS").context("set ORA_PASS")?;
    let dsn = std::env::var("ORA_DSN").context("set ORA_DSN (e.g. host:1521/service)")?;

    // ---- 1. read CSV, force every column to String (== pandas dtype='str')
    let df = CsvReadOptions::default()
        .with_has_header(true)
        .with_parse_options(CsvParseOptions::default().with_separator(sep))
        .try_into_reader_with_file_path(Some(csv_path.clone().into()))
        .with_context(|| format!("opening {csv_path}"))?
        .finish()
        .context("parsing CSV")?
        .lazy()
        .select([col("*").cast(DataType::String)])
        .collect()
        .context("casting all columns to String")?;

    if df.height() == 0 {
        println!("CSV has no rows; nothing to load.");
        return Ok(());
    }

    // ---- 2. truncate column names to 30 chars
    let names: Vec<String> = df
        .get_column_names()
        .iter()
        .map(|c| c.chars().take(30).collect::<String>())
        .collect();

    // pull each column as a &StringChunked once (avoid per-row lookups)
    let mut cols: Vec<&StringChunked> = Vec::with_capacity(df.width());
    for orig in df.get_column_names() {
        let sc = df
            .column(orig)?
            .as_materialized_series()
            .str()
            .with_context(|| format!("column {orig} is not String"))?;
        cols.push(sc);
    }

    // ---- 3. size each column -> VARCHAR2(n) or CLOB
    let col_defs: Vec<String> = names
        .iter()
        .zip(&cols)
        .map(|(name, sc)| {
            let max_bytes = sc
                .into_iter()
                .filter_map(|opt| opt.map(str::len)) // .len() = UTF-8 byte length
                .max()
                .unwrap_or(0)
                .max(1); // never declare VARCHAR2(0)
            let ty = if max_bytes <= VARCHAR2_MAX_BYTES {
                format!("VARCHAR2({max_bytes})")
            } else {
                "CLOB".to_string()
            };
            format!("\"{name}\" {ty}")
        })
        .collect();

    let table = table.chars().take(30).collect::<String>();
    let create_sql = format!("CREATE TABLE {table} (\n  {}\n)", col_defs.join(",\n  "));

    // ---- 4. connect
    let conn = Connection::connect(&user, &pass, &dsn).context("connecting to Oracle")?;

    // create table if it does not already exist (ORA-00955 = name already used)
    match conn.execute(&create_sql, &[]) {
        Ok(_) => println!("created table {table}"),
        Err(e) if e.to_string().contains("ORA-00955") => {
            println!("table {table} already exists; appending");
        }
        Err(e) => return Err(anyhow!("CREATE TABLE failed: {e}\nSQL was:\n{create_sql}")),
    }

    // ---- 5. batched INSERT with numbered binds :1, :2, ...
    let quoted_cols = names
        .iter()
        .map(|n| format!("\"{n}\""))
        .collect::<Vec<_>>()
        .join(", ");
    let placeholders = (1..=names.len())
        .map(|i| format!(":{i}"))
        .collect::<Vec<_>>()
        .join(", ");
    let insert_sql = format!("INSERT INTO {table} ({quoted_cols}) VALUES ({placeholders})");

    let mut batch = conn.batch(&insert_sql, BATCH_SIZE).build()?;
    let nrows = df.height();
    for i in 0..nrows {
        // Option<&str> binds as NULL when None; borrows live only this iteration
        let row: Vec<Option<&str>> = cols.iter().map(|c| c.get(i)).collect();
        let binds: Vec<&dyn oracle::sql_type::ToSql> = row
            .iter()
            .map(|v| v as &dyn oracle::sql_type::ToSql)
            .collect();
        batch.append_row(&binds)?;
    }
    batch.execute()?; // flush any partial batch
    conn.commit()?;

    let elapsed = start.elapsed();

    println!(
        "loaded {nrows} rows into {table} elapsed {:?}",
        elapsed.as_secs()
    );

    Ok(())
}

// export ORA_USER=xxx ORA_PASS=xxx ORA_DSN=host:1521/service
// cargo run --bin csv_to_oracle -- src/bin/test.csv som_edi_test ','

// $ cargo run --bin csv_to_oracle -- src/bin/test.csv som_edi_test ','
//     Finished `dev` profile [unoptimized + debuginfo] target(s) in 13.41s
//      Running `target\debug\csv_to_oracle.exe src/bin/test.csv som_edi_test ,`
// created table som_edi_test
// loaded 365480 rows into som_edi_test elapsed 15