#![allow(deprecated)]

//! Per-thread decode-throughput benchmark for the `oracledb` thin-mode driver.
//!
//! This is the companion to `concurrent_throughput.rs`. That bench isolates the
//! *no-GIL concurrency* win: aggregate rows/sec across N parallel connections.
//! This bench isolates the orthogonal, single-threaded **language** win: how fast
//! *one* connection decodes a large result on the client, with no concurrency and
//! no GIL involved on either side.
//!
//! The point is to strip out every confound that the concurrent bench carries — a
//! shared server ceiling, thread scheduling, GIL hand-off — and measure the raw
//! thing underneath: a compiled, no-per-cell-object decoder versus an interpreted
//! one that materializes a Python object per cell. Both drivers issue the *same*
//! SQL to the *same* container over the *same* loopback socket and page the *same*
//! way; the only variable is the codec. So this is the cleanest available read on
//! "how much faster does Rust turn wire bytes into typed values," and unlike the
//! concurrent number it shows up at N = 1.
//!
//! ## The workload is a large paged fetch of mixed scalar types
//!
//! Each pass fetches [`FETCH_ROWS`] rows of a mixed [`NUMBER`, `VARCHAR2`, `DATE`]
//! projection generated by a `connect by level` row source. That generator runs
//! on the *server* once per pass and is cheap (no table, no joins); the expensive
//! work is on the *client*, decoding every cell:
//!
//!   * `NUMBER`   — parse Oracle's base-100 mantissa/exponent bytes into a
//!     lossless decimal. Rust decodes into an inline `{ i128, scale }` with no
//!     heap allocation for the common case; python-oracledb builds a Python
//!     `int`/`Decimal`/`float` object per cell.
//!   * `VARCHAR2` — UTF-8 validate + build a string. Rust builds a `String` (or,
//!     on the borrowed path, a `&str` over the fetch buffer); Python builds a
//!     `str` object per cell.
//!   * `DATE`     — decode the 7-byte Oracle date into broken-down fields. Rust
//!     keeps the fields inline in the `QueryValue`; Python builds a
//!     `datetime.datetime` object per cell.
//!
//! The row is deliberately **narrow** (a handful of columns) and paged at a
//! modest [`FETCH_ARRAYSIZE`], so no single fetch batch is wide enough to trip the
//! multi-packet wide-row reassembly defect that `concurrent_throughput.rs`
//! documents (that defect is about ~20-column rows past ~1500 per *single* batch;
//! this bench fetches a narrow row across ~[`FETCH_ROWS`]/[`FETCH_ARRAYSIZE`]
//! pages, each page well inside one packet's worth of that row). The total decode
//! volume is large precisely *because* there are many rows, not because any one
//! batch is wide.
//!
//! Throughput is `FETCH_ROWS / wall-clock`, reported as rows decoded per second on
//! the client. The Python half (`benches/compare_decode_python.py`) runs the
//! byte-identical SQL under python-oracledb thin and prints the same metric, so
//! the ratio is apples to apples.
//!
//! ## Running
//!
//! ```sh
//! eval "$(ORACLEDB_CONTAINER_NAME=rust-oracledb-lane-1525 \
//!         ORACLEDB_HOST_PORT=1525 scripts/container.sh env)"
//! CARGO_TARGET_DIR=/path/to/target \
//!   cargo bench -p oracledb --bench decode_throughput
//! ```
//!
//! When the container environment is absent the harness prints a skip notice and
//! returns cleanly, so `cargo bench` stays green offline. Like the concurrent
//! bench it is a plain `main` wired through `harness = false`: criterion's
//! sampling model is the wrong fit for a fixed, multi-second whole-result decode.

use std::time::Instant;

use oracledb::protocol::thin::{QueryResult, QueryValue};
use oracledb::{BlockingConnection, ConnectOptions, Connection};
use oracledb_protocol::ClientIdentity;

const PROGRAM: &str = "rust-oracledb-decodebench";
const MACHINE: &str = "bench-machine";
const OSUSER: &str = "bench-osuser";
const TERMINAL: &str = "bench-terminal";
const DRIVER: &str = "rust-oracledb thn : 0.0.0";

/// Total rows decoded per pass. Large enough that the per-pass cost is dominated
/// by per-cell decode (not statement setup or the single round-trip latency), and
/// large enough to average out host jitter over a multi-second window. Kept in
/// lockstep with `benches/compare_decode_python.py`.
const FETCH_ROWS: u64 = 300_000;

/// Arraysize (prefetch / page size) for the paged fetch. Narrow row × this page
/// size keeps every batch comfortably inside the single-packet envelope, so the
/// multi-packet wide-row reassembly path documented in `concurrent_throughput.rs`
/// is never exercised. ~[`FETCH_ROWS`]/[`FETCH_ARRAYSIZE`] fetch round-trips page
/// the result.
const FETCH_ARRAYSIZE: u32 = 1_000;

/// Number of measured passes; the reported throughput is the median, which is
/// robust to the occasional slow pass on a shared host.
const PASSES: usize = 5;

/// One unmeasured warmup pass primes the statement cache and the decode path.
const WARMUP_PASSES: usize = 1;

/// The decode-heavy SELECT: a mixed `NUMBER` + `VARCHAR2` + `DATE` projection over
/// a `connect by level` generator. Byte-identical to the Python harness's SQL so
/// the two decoders see the same wire bytes.
///
///   * two `NUMBER`  columns — distinct multi-digit integers (several mantissa
///     bytes each), so the base-100 decoder does real work per cell;
///   * two `VARCHAR2` columns — distinct ~32-char strings, so each cell forces a
///     UTF-8 validate + string build;
///   * one `DATE`    column — a distinct date per row, so each cell forces a
///     7-byte Oracle-date decode.
fn decode_sql() -> String {
    format!(
        "select \
           level * 1000003 + 7                          as n0, \
           level * 31 + 100000                           as n1, \
           rpad('row' || to_char(level) || '-a', 32, 'x') as v0, \
           rpad('row' || to_char(level) || '-b', 32, 'y') as v1, \
           date '2000-01-01' + numtodsinterval(mod(level, 9000), 'day') as d0 \
         from dual connect by level <= {FETCH_ROWS}"
    )
}

/// Number of columns the projection returns; used to bound the per-row cell loop.
const DECODE_COLS: usize = 5;

/// Build connect options from the harness container environment, or `None` so the
/// bench can self-skip when the container is not configured.
fn connect_options() -> Option<ConnectOptions> {
    let connect_string = std::env::var("PYO_TEST_CONNECT_STRING").ok()?;
    let user = std::env::var("PYO_TEST_MAIN_USER").ok()?;
    let password = std::env::var("PYO_TEST_MAIN_PASSWORD").ok()?;
    let identity = ClientIdentity::new(PROGRAM, MACHINE, OSUSER, TERMINAL, DRIVER).ok()?;
    Some(ConnectOptions::new(
        connect_string,
        user,
        password,
        identity,
    ))
}

/// Decode the whole result once, touching every cell so the optimizer cannot
/// elide the decode the benchmark exists to measure, and return the row count.
/// `release_cursor` returns the server cursor to the statement cache so a
/// later pass reuses it — matching how python-oracledb reuses a cursor object.
fn decode_once(conn: &mut Connection, sql: &str) -> u64 {
    let first = BlockingConnection::execute_raw(
        conn,
        sql,
        FETCH_ARRAYSIZE,
        &[],
        oracledb::protocol::thin::ExecuteOptions::default(),
        None,
    )
    .expect("decode-heavy execute");
    let cursor_id = first.cursor_id;
    let mut total = touch_all_cells(&first);
    let mut more_rows = first.more_rows;
    let mut previous_row: Option<Vec<Option<QueryValue>>> = first.rows.last().cloned();
    while more_rows && cursor_id != 0 {
        let batch = BlockingConnection::fetch_rows(
            conn,
            cursor_id,
            FETCH_ARRAYSIZE,
            previous_row.as_deref(),
        )
        .expect("decode-heavy fetch page");
        total += touch_all_cells(&batch);
        more_rows = batch.more_rows;
        if let Some(last) = batch.rows.last().cloned() {
            previous_row = Some(last);
        }
    }
    conn.release_cursor(cursor_id);
    total
}

/// Read every cell of every row through the typed accessors so the decoded
/// `NUMBER` text, `VARCHAR2` string, and `DATE` fields are actually consumed.
/// Returns the row count. `black_box` on the folded sink prevents the loop from
/// being optimized away.
fn touch_all_cells(result: &QueryResult) -> u64 {
    let mut sink: u64 = 0;
    for row in 0..result.rows.len() {
        for col in 0..DECODE_COLS {
            if let Some(value) = result.cell(row, col) {
                if let Some(n) = value.as_i64() {
                    sink = sink.wrapping_add(n as u64);
                } else if let Some(text) = value.as_text() {
                    sink = sink.wrapping_add(text.len() as u64);
                } else if let QueryValue::DateTime {
                    year,
                    month,
                    day,
                    hour,
                    minute,
                    second,
                    ..
                } = value
                {
                    sink = sink.wrapping_add(
                        *year as u64
                            + *month as u64
                            + *day as u64
                            + *hour as u64
                            + *minute as u64
                            + *second as u64,
                    );
                }
            }
        }
    }
    std::hint::black_box(sink);
    result.rows.len() as u64
}

/// Median of a slice of throughputs (rows/sec). Robust to the occasional slow
/// pass on a shared host; with [`PASSES`] = 5 the middle element is the median.
fn median(mut values: Vec<f64>) -> f64 {
    values.sort_by(|a, b| a.partial_cmp(b).expect("throughputs are finite"));
    let mid = values.len() / 2;
    if values.len() % 2 == 1 {
        values[mid]
    } else {
        (values[mid - 1] + values[mid]) / 2.0
    }
}

fn main() {
    let Some(options) = connect_options() else {
        eprintln!(
            "skipped decode_throughput bench: PYO_TEST_* environment not configured \
             (source scripts/container.sh env to run against the container)"
        );
        return;
    };

    let mut conn = match BlockingConnection::connect(options) {
        Ok(conn) => conn,
        Err(err) => {
            eprintln!("skipped decode_throughput bench: connect failed: {err}");
            return;
        }
    };

    let sql = decode_sql();

    println!("rust-oracledb per-thread decode throughput (thin mode)");
    println!(
        "workload: paged fetch of {FETCH_ROWS} rows x {DECODE_COLS} cols \
         (2 NUMBER, 2 VARCHAR2, 1 DATE), arraysize {FETCH_ARRAYSIZE}, single connection"
    );
    println!("passes: {PASSES} measured (after {WARMUP_PASSES} warmup), median reported\n");

    for _ in 0..WARMUP_PASSES {
        let rows = decode_once(&mut conn, &sql);
        assert_eq!(rows, FETCH_ROWS, "warmup decoded the full result");
    }

    let mut throughputs = Vec::with_capacity(PASSES);
    println!("{:>5}  {:>16}  {:>14}", "pass", "rows/sec", "wall (ms)");
    for pass in 1..=PASSES {
        let start = Instant::now();
        let rows = decode_once(&mut conn, &sql);
        let elapsed = start.elapsed().as_secs_f64();
        assert_eq!(rows, FETCH_ROWS, "decoded the full result");
        let throughput = rows as f64 / elapsed.max(f64::MIN_POSITIVE);
        throughputs.push(throughput);
        println!("{pass:>5}  {throughput:>16.0}  {:>14.1}", elapsed * 1e3);
    }

    let median_throughput = median(throughputs.clone());
    let best = throughputs.iter().copied().fold(f64::MIN, f64::max);
    println!(
        "\nmedian: {median_throughput:>.0} rows/sec decoded (best pass {best:.0}); \
         single connection, no GIL involved on either side."
    );

    let _ = BlockingConnection::close(conn);
}
