faucet-source-oracle 1.2.0

//! Pure query planning: bind resolution, incremental filtering, discovery row
//! grouping and state-key derivation. No driver calls.

use std::collections::HashMap;

use faucet_common_oracle::{TypeFamily, quote_ident_oracle, quote_validated};
use faucet_core::replication::cursor::{CursorBookmark, CursorTracker};
use faucet_core::{DatasetDescriptor, FaucetError};
use serde_json::{Value, json};

use crate::config::{OracleReplication, OracleSourceConfig};

/// Incremental-replication context resolved for one run.
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct IncrementalCtx {
    pub column: String,
    pub start: CursorBookmark,
}

impl IncrementalCtx {
    pub fn tracker(&self) -> CursorTracker {
        CursorTracker::new(self.column.clone(), self.start.clone())
    }
}

/// `inner` filtered to `column >= :bookmark`, ordered by the cursor when
/// `ordered`; full replication passes through unchanged. The column was
/// validated with [`quote_ident_oracle`] at load time.
pub(crate) fn cursor_wrap(inner: &str, incr: Option<&IncrementalCtx>, ordered: bool) -> String {
    let Some(ctx) = incr else {
        return inner.to_string();
    };
    let c = quote_validated(&ctx.column);
    let filtered = format!("SELECT * FROM ({inner}) q WHERE q.{c} >= :bookmark");
    if ordered {
        format!("{filtered} ORDER BY q.{c}")
    } else {
        filtered
    }
}

/// A planned query: the SQL, the positional values for `:1..:n`, and the
/// bookmark value for `:bookmark` (when incremental).
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct PlannedQuery {
    pub sql: String,
    pub params: Vec<Value>,
    pub bookmark: Option<Value>,
    pub incremental: Option<IncrementalCtx>,
}

/// Resolve parent-context placeholders and the incremental cursor.
pub(crate) fn plan_query(
    config: &OracleSourceConfig,
    context: &HashMap<String, Value>,
    start_bookmark: Option<&CursorBookmark>,
) -> PlannedQuery {
    let (sql, ctx_values) = faucet_core::util::substitute_context_bind_params(
        &config.query,
        context,
        config.params.len() + 1,
        |i| format!(":{i}"),
    );
    let mut params = config.params.clone();
    params.extend(ctx_values);

    let incremental = match &config.replication {
        OracleReplication::Full => None,
        OracleReplication::Incremental {
            column,
            initial_value,
        } => Some(IncrementalCtx {
            column: column.clone(),
            start: start_bookmark
                .cloned()
                .unwrap_or_else(|| CursorBookmark::initial(initial_value.clone())),
        }),
    };
    // The cursor filter around the query always binds `:bookmark`.
    let bookmark = incremental.as_ref().map(|ctx| ctx.start.value.clone());
    PlannedQuery {
        sql,
        params,
        bookmark,
        incremental,
    }
}

/// Map the statement's bind names (as the driver reports them, upper-cased) to
/// values: `:N` → `params[N-1]`, `:BOOKMARK` → the cursor.
pub(crate) fn resolve_binds(
    names: &[String],
    params: &[Value],
    bookmark: Option<&Value>,
) -> Result<Vec<(String, Value)>, FaucetError> {
    names
        .iter()
        .map(|name| {
            if let Ok(n) = name.parse::<usize>() {
                return params
                    .get(n.wrapping_sub(1))
                    .cloned()
                    .map(|v| (name.clone(), v))
                    .ok_or_else(|| {
                        FaucetError::Config(format!(
                            "oracle query binds :{name} but only {} `params` value(s) are set",
                            params.len()
                        ))
                    });
            }
            if name.eq_ignore_ascii_case("BOOKMARK") {
                return bookmark.cloned().map(|v| (name.clone(), v)).ok_or_else(|| {
                    FaucetError::Config(
                        "oracle query uses :bookmark but replication is not incremental".into(),
                    )
                });
            }
            Err(FaucetError::Config(format!(
                "oracle query binds unknown placeholder :{name} (use :1..:n for `params` \
                 and :bookmark for the incremental cursor)"
            )))
        })
        .collect()
}

/// An owned bind value with a JSON-faithful Oracle type.
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum OwnedBind {
    Null,
    Int(i64),
    Float(f64),
    Text(String),
    Timestamp(chrono::NaiveDateTime),
    TimestampTz(chrono::DateTime<chrono::FixedOffset>),
}

/// Each resolved bind paired with its owned value, typed by its name
/// (`:bookmark` binds a timestamp-shaped bookmark as a timestamp, SQL-41).
pub(crate) fn owned_binds(binds: Vec<(String, Value)>) -> Vec<(String, OwnedBind)> {
    binds
        .into_iter()
        .map(|(n, v)| {
            let bind = OwnedBind::for_name(&n, &v);
            (n, bind)
        })
        .collect()
}

/// [`owned_binds`] with `:bookmark` bound as plain text, for a cursor column
/// that holds ISO-8601 strings rather than a date/time type (#862 M-067).
pub(crate) fn text_binds(binds: Vec<(String, Value)>) -> Vec<(String, OwnedBind)> {
    binds
        .into_iter()
        .map(|(n, v)| {
            let bind = OwnedBind::from_value(&v);
            (n, bind)
        })
        .collect()
}

/// Oracle's datetime-conversion errors: what comparing a text column holding
/// `YYYY-MM-DDTHH:MI:SS` to a timestamp bind raises under the session's NLS
/// format (ORA-01861 "literal does not match format string" and kin).
pub(crate) fn is_datetime_conversion_error(message: &str) -> bool {
    const CODES: &[&str] = &[
        "ORA-01830",
        "ORA-01839",
        "ORA-01840",
        "ORA-01841",
        "ORA-01843",
        "ORA-01847",
        "ORA-01850",
        "ORA-01851",
        "ORA-01858",
        "ORA-01861",
        "ORA-01862",
    ];
    CODES.iter().any(|c| message.contains(c))
}

impl OwnedBind {
    pub fn from_value(v: &Value) -> Self {
        match v {
            Value::Null => OwnedBind::Null,
            Value::Bool(b) => OwnedBind::Int(i64::from(*b)),
            Value::Number(n) => match n.as_i64() {
                Some(i) => OwnedBind::Int(i),
                None if n.is_u64() => OwnedBind::Text(n.to_string()),
                None => OwnedBind::Float(n.as_f64().unwrap_or_default()),
            },
            Value::String(s) => OwnedBind::Text(s.clone()),
            other => OwnedBind::Text(other.to_string()),
        }
    }

    /// The bind for placeholder `name`: [`for_bookmark`](Self::for_bookmark) for
    /// `:bookmark`, else [`from_value`](Self::from_value).
    pub fn for_name(name: &str, v: &Value) -> Self {
        if name.eq_ignore_ascii_case("BOOKMARK") {
            Self::for_bookmark(v)
        } else {
            Self::from_value(v)
        }
    }

    /// The `:bookmark` bind. A cursor value in the form this source emits for
    /// `DATE` / `TIMESTAMP` / `TIMESTAMP WITH TIME ZONE` columns
    /// (`YYYY-MM-DDTHH:MI:SS[.f][offset]`) binds as a timestamp: as text, the
    /// session's `NLS_DATE_FORMAT` conversion rejects the `T` with ORA-01861
    /// (SQL-41). An offset is kept, so the comparison is between instants.
    pub fn for_bookmark(v: &Value) -> Self {
        if let Value::String(s) = v
            && s.len() >= 19
            && s.as_bytes()[10] == b'T'
        {
            if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) {
                return OwnedBind::TimestampTz(dt);
            }
            if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S%.f") {
                return OwnedBind::Timestamp(dt);
            }
        }
        Self::from_value(v)
    }
}

/// Derive a stable state key from the database scope and a query fingerprint.
pub(crate) fn default_state_key(config: &OracleSourceConfig) -> String {
    key_for_scope(config, &config.connection.scope_label())
}

/// The keys [`default_state_key`] rendered before the scope named the host
/// and port, for moving a bookmark stored under one of them.
pub(crate) fn legacy_state_keys(config: &OracleSourceConfig) -> Vec<String> {
    config
        .connection
        .legacy_scope_label()
        .map(|scope| vec![key_for_scope(config, &scope)])
        .unwrap_or_default()
}

fn key_for_scope(config: &OracleSourceConfig, scope: &str) -> String {
    format!(
        "oracle:{scope}:{:016x}",
        faucet_core::shard::shard_hash(&config.query)
    )
}

/// One `ALL_TAB_COLUMNS` row joined with `ALL_TABLES.NUM_ROWS`.
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct CatalogRow {
    pub owner: String,
    pub table: String,
    pub column: String,
    pub data_type: String,
    pub scale: Option<i64>,
    pub nullable: bool,
    pub num_rows: Option<i64>,
}

/// Group catalog rows (ordered by owner, table, column id) into one descriptor
/// per table. Native `JSON` columns are serialized in the generated query and
/// listed under `json_columns`, since the driver cannot fetch them directly.
pub(crate) fn descriptors_from_catalog(
    rows: Vec<CatalogRow>,
) -> Result<Vec<DatasetDescriptor>, FaucetError> {
    let mut out = Vec::new();
    let mut i = 0;
    while i < rows.len() {
        let (owner, table) = (rows[i].owner.clone(), rows[i].table.clone());
        let mut j = i;
        while j < rows.len() && rows[j].owner == owner && rows[j].table == table {
            j += 1;
        }
        out.push(table_descriptor(&rows[i..j])?);
        i = j;
    }
    Ok(out)
}

fn table_descriptor(cols: &[CatalogRow]) -> Result<DatasetDescriptor, FaucetError> {
    let first = &cols[0];
    let from = format!(
        "{}.{}",
        quote_ident_oracle(&first.owner)?,
        quote_ident_oracle(&first.table)?
    );
    let mut select = Vec::with_capacity(cols.len());
    let mut json_columns = Vec::new();
    let mut schema_cols = Vec::with_capacity(cols.len());
    for c in cols {
        let family = TypeFamily::from_data_type(&c.data_type, c.scale);
        let q = quote_ident_oracle(&c.column)?;
        if family == TypeFamily::Json {
            select.push(format!("JSON_SERIALIZE({q} RETURNING CLOB) AS {q}"));
            json_columns.push(c.column.clone());
        } else {
            select.push(q);
        }
        let mut fragment = family.json_schema();
        if c.nullable {
            fragment = match fragment.get("type").cloned() {
                Some(Value::Array(mut types)) => {
                    types.push(json!("null"));
                    fragment["type"] = Value::Array(types);
                    fragment
                }
                _ => faucet_core::nullable_type(fragment),
            };
        }
        schema_cols.push((c.column.clone(), fragment));
    }
    let query = if json_columns.is_empty() {
        format!("SELECT * FROM {from}")
    } else {
        format!("SELECT {} FROM {from}", select.join(", "))
    };
    let patch = if json_columns.is_empty() {
        json!({ "query": query })
    } else {
        json!({ "query": query, "json_columns": json_columns })
    };
    let mut d = DatasetDescriptor::new(format!("{}.{}", first.owner, first.table), "table", patch)
        .with_schema(faucet_core::columns_to_schema(schema_cols));
    if let Some(n) = first.num_rows.filter(|n| *n >= 0) {
        d = d.with_estimated_rows(n as u64);
    }
    Ok(d)
}

#[cfg(test)]
mod tests {
    use super::*;
    use faucet_common_oracle::OracleConnectionConfig;

    #[test]
    fn text_cursor_conversion_errors_fall_back_to_a_text_bookmark() {
        assert!(super::is_datetime_conversion_error(
            "ORA-01861: literal does not match format string"
        ));
        assert!(!super::is_datetime_conversion_error(
            "ORA-00942: table or view does not exist"
        ));
        let binds = vec![(
            "BOOKMARK".to_string(),
            serde_json::json!("2024-01-02T03:04:05Z"),
        )];
        assert!(matches!(
            super::owned_binds(binds.clone())[0].1,
            super::OwnedBind::TimestampTz(_)
        ));
        assert_eq!(
            super::text_binds(binds)[0].1,
            super::OwnedBind::Text("2024-01-02T03:04:05Z".into())
        );
    }

    #[test]
    fn temporal_bookmarks_bind_as_timestamps() {
        let naive = chrono::NaiveDate::from_ymd_opt(2024, 3, 1)
            .unwrap()
            .and_hms_opt(12, 30, 45)
            .unwrap();
        assert_eq!(
            OwnedBind::for_bookmark(&json!("2024-03-01T12:30:45")),
            OwnedBind::Timestamp(naive)
        );
        assert_eq!(
            OwnedBind::for_bookmark(&json!("1970-01-01T00:00:00")),
            OwnedBind::Timestamp(chrono::DateTime::UNIX_EPOCH.naive_utc())
        );
        assert!(matches!(
            OwnedBind::for_bookmark(&json!("2024-03-01T12:30:45.123456")),
            OwnedBind::Timestamp(t) if t.and_utc().timestamp_subsec_micros() == 123_456
        ));
        assert!(matches!(
            OwnedBind::for_bookmark(&json!("2024-03-01T12:30:45.5+05:30")),
            OwnedBind::TimestampTz(tz) if tz.offset().local_minus_utc() == 5 * 3600 + 1800
        ));
        assert!(matches!(
            OwnedBind::for_bookmark(&json!("2024-03-01T12:30:45Z")),
            OwnedBind::TimestampTz(_)
        ));
        assert_eq!(
            OwnedBind::for_bookmark(&json!("2024-03-01")),
            OwnedBind::Text("2024-03-01".into())
        );
        assert_eq!(
            OwnedBind::for_bookmark(&json!("2024-03-01Tnope-not-a-time")),
            OwnedBind::Text("2024-03-01Tnope-not-a-time".into())
        );
        assert_eq!(OwnedBind::for_bookmark(&json!(7)), OwnedBind::Int(7));
        assert!(matches!(
            OwnedBind::for_name("BOOKMARK", &json!("2024-03-01T00:00:00")),
            OwnedBind::Timestamp(_)
        ));
        assert_eq!(
            OwnedBind::for_name("1", &json!("2024-03-01T00:00:00")),
            OwnedBind::Text("2024-03-01T00:00:00".into())
        );
    }

    fn cfg(query: &str) -> OracleSourceConfig {
        OracleSourceConfig::new(
            OracleConnectionConfig::new("db", 1521, "FREEPDB1", "u", "p"),
            query,
        )
    }

    #[test]
    fn plan_full_query_keeps_params() {
        let mut c = cfg("SELECT * FROM T WHERE A = :1");
        c.params = vec![json!(5)];
        let p = plan_query(&c, &HashMap::new(), None);
        assert_eq!(p.sql, "SELECT * FROM T WHERE A = :1");
        assert_eq!(p.params, vec![json!(5)]);
        assert_eq!(p.bookmark, None);
        assert_eq!(p.incremental, None);
    }

    #[test]
    fn plan_substitutes_context_after_params() {
        let mut c = cfg("SELECT * FROM T WHERE A = :1 AND P = {parent.id}");
        c.params = vec![json!("x")];
        let ctx = HashMap::from([("parent.id".to_string(), json!(9))]);
        let p = plan_query(&c, &ctx, None);
        assert_eq!(p.sql, "SELECT * FROM T WHERE A = :1 AND P = :2");
        assert_eq!(p.params, vec![json!("x"), json!(9)]);
    }

    #[test]
    fn plan_incremental_uses_bookmark_or_initial() {
        let mut c = cfg("SELECT * FROM T WHERE U >= :bookmark");
        c.replication = OracleReplication::Incremental {
            column: "U".into(),
            initial_value: json!(0),
        };
        let p = plan_query(&c, &HashMap::new(), None);
        assert_eq!(p.bookmark, Some(json!(0)));
        let stored = CursorBookmark::initial(json!(42));
        let p = plan_query(&c, &HashMap::new(), Some(&stored));
        assert_eq!(p.bookmark, Some(json!(42)));
        assert_eq!(p.incremental.unwrap().start.value, json!(42));

        // Without the token the outer cursor filter still binds `:bookmark`.
        c.query = "SELECT * FROM T".into();
        let p = plan_query(&c, &HashMap::new(), None);
        assert_eq!(p.bookmark, Some(json!(0)));
        assert!(p.incremental.is_some());
    }

    #[test]
    fn cursor_wrap_filters_with_ge_and_orders() {
        let ctx = IncrementalCtx {
            column: "UPDATED_AT".into(),
            start: CursorBookmark::initial(json!(0)),
        };
        assert_eq!(
            cursor_wrap("SELECT * FROM T", Some(&ctx), true),
            "SELECT * FROM (SELECT * FROM T) q WHERE q.\"UPDATED_AT\" >= :bookmark \
             ORDER BY q.\"UPDATED_AT\""
        );
        assert!(!cursor_wrap("SELECT 1 FROM DUAL", Some(&ctx), false).contains("ORDER BY"));
        assert_eq!(
            cursor_wrap("SELECT 1 FROM DUAL", None, true),
            "SELECT 1 FROM DUAL"
        );
    }

    #[test]
    fn tracker_keeps_rows_tied_at_the_bookmark() {
        let mut first = IncrementalCtx {
            column: "U".into(),
            start: CursorBookmark::initial(json!(0)),
        }
        .tracker();
        let a = json!({"ID": 1, "U": "2024-01-01T00:00:00"});
        assert!(first.admit(&a));
        let mut second = IncrementalCtx {
            column: "U".into(),
            start: CursorBookmark::from_state(first.bookmark().to_value()).unwrap(),
        }
        .tracker();
        assert!(!second.admit(&a));
        assert!(second.admit(&json!({"ID": 2, "U": "2024-01-01T00:00:00"})));
    }

    #[test]
    fn resolve_binds_by_name() {
        let names = vec!["1".to_string(), "BOOKMARK".to_string(), "2".to_string()];
        let out = resolve_binds(&names, &[json!("a"), json!(2)], Some(&json!(7))).unwrap();
        assert_eq!(
            out,
            vec![
                ("1".to_string(), json!("a")),
                ("BOOKMARK".to_string(), json!(7)),
                ("2".to_string(), json!(2)),
            ]
        );
        assert!(resolve_binds(&["3".into()], &[json!(1)], None).is_err());
        assert!(resolve_binds(&["0".into()], &[json!(1)], None).is_err());
        assert!(resolve_binds(&["BOOKMARK".into()], &[], None).is_err());
        let err = resolve_binds(&["FOO".into()], &[], None).unwrap_err();
        assert!(err.to_string().contains(":FOO"));
    }

    #[test]
    fn owned_bind_types() {
        assert_eq!(OwnedBind::from_value(&Value::Null), OwnedBind::Null);
        assert_eq!(OwnedBind::from_value(&json!(true)), OwnedBind::Int(1));
        assert_eq!(OwnedBind::from_value(&json!(-3)), OwnedBind::Int(-3));
        assert_eq!(
            OwnedBind::from_value(&json!(u64::MAX)),
            OwnedBind::Text(u64::MAX.to_string())
        );
        assert_eq!(OwnedBind::from_value(&json!(1.5)), OwnedBind::Float(1.5));
        assert_eq!(
            OwnedBind::from_value(&json!("s")),
            OwnedBind::Text("s".into())
        );
        assert_eq!(
            OwnedBind::from_value(&json!({"a": 1})),
            OwnedBind::Text("{\"a\":1}".into())
        );
    }

    #[test]
    fn state_key_is_stable_and_valid() {
        let c = cfg("SELECT * FROM T");
        let k = default_state_key(&c);
        assert_eq!(k, default_state_key(&c));
        let scope = c.connection.scope_label();
        assert!(k.starts_with(&format!("oracle:{scope}:")), "{k}");
        assert!(scope.contains("FREEPDB1") && scope != "FREEPDB1", "{scope}");
        let legacy = legacy_state_keys(&c);
        assert_eq!(legacy.len(), 1);
        assert!(legacy[0].starts_with("oracle:FREEPDB1:"), "{legacy:?}");
        assert_eq!(
            &legacy[0]["oracle:FREEPDB1".len()..],
            &k[k.rfind(':').unwrap()..]
        );
        assert!(k.ends_with(&format!(
            "{:016x}",
            faucet_core::shard::shard_hash(&c.query)
        )));
        faucet_core::state::validate_state_key(&k).unwrap();
    }

    fn row(owner: &str, table: &str, col: &str, ty: &str, scale: Option<i64>) -> CatalogRow {
        CatalogRow {
            owner: owner.into(),
            table: table.into(),
            column: col.into(),
            data_type: ty.into(),
            scale,
            nullable: col != "ID",
            num_rows: Some(12),
        }
    }

    #[test]
    fn descriptors_group_and_type() {
        let rows = vec![
            row("APP", "ORDERS", "ID", "NUMBER", Some(0)),
            row("APP", "ORDERS", "TOTAL", "NUMBER", Some(2)),
            row("APP", "ORDERS", "DOC", "JSON", None),
            row("HR", "EMP", "ID", "NUMBER", Some(0)),
        ];
        let ds = descriptors_from_catalog(rows).unwrap();
        assert_eq!(ds.len(), 2);
        assert_eq!(ds[0].name, "APP.ORDERS");
        assert_eq!(ds[0].estimated_rows, Some(12));
        assert_eq!(
            ds[0].config_patch["query"],
            "SELECT \"ID\", \"TOTAL\", JSON_SERIALIZE(\"DOC\" RETURNING CLOB) AS \"DOC\" \
             FROM \"APP\".\"ORDERS\""
        );
        assert_eq!(ds[0].config_patch["json_columns"], json!(["DOC"]));
        let schema = ds[0].schema.as_ref().unwrap();
        assert_eq!(
            schema["properties"]["ID"]["type"],
            json!(["integer", "string"])
        );
        assert_eq!(
            schema["properties"]["TOTAL"]["type"],
            json!(["number", "string", "null"])
        );
        assert_eq!(
            ds[1].config_patch,
            json!({"query": "SELECT * FROM \"HR\".\"EMP\""})
        );
    }

    #[test]
    fn descriptors_handle_missing_stats_and_bad_names() {
        let mut r = row("APP", "T", "ID", "NUMBER", Some(0));
        r.num_rows = None;
        let ds = descriptors_from_catalog(vec![r]).unwrap();
        assert_eq!(ds[0].estimated_rows, None);
        assert!(descriptors_from_catalog(vec![row("A\"", "T", "ID", "NUMBER", None)]).is_err());
        assert!(descriptors_from_catalog(Vec::new()).unwrap().is_empty());
    }
    #[test]
    fn owned_binds_keep_names_and_type_each_value() {
        let got = owned_binds(vec![
            ("bookmark".into(), json!("2024-03-01T00:00:00")),
            ("n".into(), json!(1)),
        ]);
        assert_eq!(got[0].0, "bookmark");
        assert!(matches!(got[0].1, OwnedBind::Timestamp(_)));
        assert_eq!(got[1].1, OwnedBind::Int(1));
    }
}