faucet_source_oracle/
config.rs

1//! Configuration for the Oracle query source.
2
3use faucet_common_oracle::OracleConnectionConfig;
4use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError, validate_batch_size};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9fn default_max_connections() -> u32 {
10    10
11}
12fn default_batch_size() -> usize {
13    DEFAULT_BATCH_SIZE
14}
15fn default_statement_timeout_secs() -> u64 {
16    300
17}
18
19/// How the source replicates rows across runs.
20///
21/// Serializes as `{ type: full }` or
22/// `{ type: incremental, column: "UPDATED_AT", initial_value: ... }`.
23#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, Default, PartialEq)]
24#[serde(tag = "type", rename_all = "snake_case")]
25pub enum OracleReplication {
26    /// Every run fetches the full result set (default).
27    #[default]
28    Full,
29    /// Only rows not yet read, in `column` order, resuming from the stored
30    /// bookmark (or `initial_value` on the first run).
31    ///
32    /// The source wraps the query as
33    /// `SELECT * FROM (<query>) q WHERE q."COLUMN" >= :bookmark ORDER BY q."COLUMN"`
34    /// and drops the rows it already emitted at the bookmark value, so rows
35    /// that share a cursor value (ties, late commits, `DATE`'s one-second
36    /// resolution) are never skipped. Every page carries a bookmark, so a
37    /// crash replays at most one page. Write `:bookmark` in the query (with
38    /// `>=`) to apply the cursor further inside it; a strict `>` is refused.
39    /// `column` names the key exactly as it appears in the output — Oracle
40    /// upper-cases unquoted names. Rows whose `column` is NULL are never read.
41    Incremental {
42        /// Output column holding the replication cursor (e.g. `UPDATED_AT`).
43        column: String,
44        /// Lower bound (inclusive) used on the first run, before any bookmark
45        /// is stored.
46        initial_value: Value,
47    },
48}
49
50/// Primary-key range sharding for clustered (Mode B) execution: each shard runs
51/// `SELECT * FROM (<query>) WHERE "<key>" >= lo AND "<key>" < hi`. The key must
52/// be an integer-valued output column, named exactly as Oracle reports it.
53#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
54pub struct ShardConfig {
55    /// Integer output column to range-partition on (quoted as an identifier).
56    pub key: String,
57}
58
59/// Configuration for [`OracleSource`](crate::OracleSource).
60#[derive(Clone, Serialize, Deserialize, JsonSchema)]
61#[schemars(extend("x-faucet-follows-servers" = true))]
62pub struct OracleSourceConfig {
63    /// Connection settings (`connect_string` or `host` + `service_name`/`sid`).
64    #[serde(flatten)]
65    pub connection: OracleConnectionConfig,
66    /// SQL query. Bind [`params`](Self::params) with `:1`, `:2`, …, and the
67    /// incremental cursor with `:bookmark`.
68    #[schemars(extend("x-faucet-sql" = true))]
69    pub query: String,
70    /// Positional bind values for `:1`…`:n`.
71    #[serde(default)]
72    pub params: Vec<Value>,
73    /// Maximum pooled sessions. Defaults to 10.
74    #[serde(default = "default_max_connections")]
75    pub max_connections: u32,
76    /// Records per emitted page, also used as the fetch array size. `0` emits
77    /// the whole result set as one page. Defaults to [`DEFAULT_BATCH_SIZE`].
78    #[serde(default = "default_batch_size")]
79    pub batch_size: usize,
80    /// Per-round-trip call timeout in seconds (`0` disables). Defaults to 300.
81    #[serde(default = "default_statement_timeout_secs")]
82    pub statement_timeout_secs: u64,
83    /// Replication mode. Defaults to [`OracleReplication::Full`].
84    #[serde(default)]
85    pub replication: OracleReplication,
86    /// Explicit state-store key for the bookmark. Derived from the database and
87    /// a query fingerprint when unset.
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub state_key: Option<String>,
90    /// Optional PK-range sharding for clustered execution. Inert outside the
91    /// cluster coordinator.
92    #[serde(default, skip_serializing_if = "Option::is_none")]
93    pub shard: Option<ShardConfig>,
94    /// Output columns whose text holds JSON (a `JSON_SERIALIZE(...)` or an
95    /// `IS JSON` `VARCHAR2`/`CLOB`); their values are emitted as parsed JSON.
96    #[serde(default, skip_serializing_if = "Vec::is_empty")]
97    pub json_columns: Vec<String>,
98}
99
100impl std::fmt::Debug for OracleSourceConfig {
101    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102        f.debug_struct("OracleSourceConfig")
103            .field("connection", &self.connection)
104            .field("query", &self.query)
105            .field("params", &self.params)
106            .field("max_connections", &self.max_connections)
107            .field("batch_size", &self.batch_size)
108            .field("statement_timeout_secs", &self.statement_timeout_secs)
109            .field("replication", &self.replication)
110            .field("state_key", &self.state_key)
111            .field("shard", &self.shard)
112            .field("json_columns", &self.json_columns)
113            .finish()
114    }
115}
116
117impl OracleSourceConfig {
118    /// A full-replication config with defaults elsewhere.
119    pub fn new(connection: OracleConnectionConfig, query: impl Into<String>) -> Self {
120        Self {
121            connection,
122            query: query.into(),
123            params: Vec::new(),
124            max_connections: default_max_connections(),
125            batch_size: default_batch_size(),
126            statement_timeout_secs: default_statement_timeout_secs(),
127            replication: OracleReplication::Full,
128            state_key: None,
129            shard: None,
130            json_columns: Vec::new(),
131        }
132    }
133
134    /// Validate the fail-fast invariants.
135    pub fn validate(&self) -> Result<(), FaucetError> {
136        self.connection.validate()?;
137        validate_batch_size(self.batch_size)?;
138        if self.query.trim().is_empty() {
139            return Err(FaucetError::Config(
140                "oracle source requires a `query`".into(),
141            ));
142        }
143        if let OracleReplication::Incremental {
144            column,
145            initial_value,
146        } = &self.replication
147        {
148            if column.trim().is_empty() {
149                return Err(FaucetError::Config(
150                    "oracle incremental replication requires a non-empty `column`".into(),
151                ));
152            }
153            faucet_common_oracle::quote_ident_oracle(column)?;
154            if initial_value.is_null() {
155                return Err(FaucetError::Config(
156                    "oracle incremental replication requires a non-null `initial_value`".into(),
157                ));
158            }
159            if faucet_core::replication::cursor::strict_comparison(
160                &self.query.to_ascii_lowercase(),
161                BOOKMARK_TOKEN,
162            ) {
163                return Err(FaucetError::Config(format!(
164                    "oracle: compare `{BOOKMARK_TOKEN}` with `>=`, not `>`: the source re-reads \
165                     the bookmark value and drops the rows it already emitted, and a strict \
166                     comparison skips rows that share it"
167                )));
168            }
169        }
170        if let Some(shard) = &self.shard {
171            faucet_common_oracle::quote_ident_oracle(&shard.key)?;
172        }
173        if let Some(key) = &self.state_key {
174            faucet_core::state::validate_state_key(key)?;
175        }
176        Ok(())
177    }
178}
179
180/// The placeholder a query uses to read the incremental bookmark
181/// (`WHERE UPDATED_AT >= :bookmark`, any case).
182pub const BOOKMARK_TOKEN: &str = ":bookmark";
183
184#[cfg(test)]
185mod tests;
faucet_source_oracle/config.rs

faucet_source_oracle/
config.rs
