faucet-source-oracle 1.2.2

//! 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;