Fix Rust bugs, add audit comments, deduplicate qi(), and document CTE query

Bug fixes:
- parse_temporal_numeric now errors on unparseable strings instead of
  silently returning 0.0, preventing corrupted bounds from propagating.
- NULL valid_from/valid_until boundaries now raise pgrx::error! instead
  of silently becoming empty strings in both target and source readers.

Audit comments:
- lib.rs: Safety analysis of thread_local! per-connection caching (memory
  ownership, cross-transaction safety, SPI data extraction).
- reader.rs: Prepared statement caching via SPI_keepplan (lifecycle,
  automatic replanning, key format).
- executor_cache.rs: Executor introspection cache (invalidation via
  source_cols_hash, growth bounds, no row data cached).

Refactor:
- Extract shared qi() function into native/src/util.rs, imported by both
  reader.rs and executor_cache.rs.
- Add comprehensive doc comment to build_column_list_cte_query explaining
  each CTE and the 6 output SQL fragments.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: jhf

expected/063_temporal_merge_planner_cache.out

@@ -3332,11 +3332,11 @@ NOTICE:  Explaining:
                     target_row.stable_pk_payload as discovered_stable_pk_payload,
                     target_row.es_id AS discovered_id_1, target_row.lu_id AS discovered_id_2, target_row.type AS discovered_id_3 /* v_propagated_id_cols_list */
                 FROM source_with_eclipsed_flag source_row
-                LEFT JOIN target_rows target_row ON (source_row.type = target_row.type AND (source_row.lu_id = target_row.lu_id OR (source_row.lu_id IS NULL AND target_row.lu_id IS NULL)) AND (source_row.es_id = target_row.es_id OR (source_row.es_id IS NULL AND target_row.es_id IS NULL)) /* v_source_rows_exists_join_expr */)
+                LEFT JOIN target_rows target_row ON ((source_row.es_id = target_row.es_id OR (source_row.es_id IS NULL AND target_row.es_id IS NULL)) AND (source_row.lu_id = target_row.lu_id OR (source_row.lu_id IS NULL AND target_row.lu_id IS NULL)) AND source_row.type = target_row.type /* v_source_rows_exists_join_expr */)
 
 NOTICE:  Hash Left Join (actual rows=2.00 loops=1)
 NOTICE:    Hash Cond: (source_row.type = target_row.type)
-NOTICE:    Join Filter: (((source_row.lu_id = target_row.lu_id) OR ((source_row.lu_id IS NULL) AND (target_row.lu_id IS NULL))) AND ((source_row.es_id = target_row.es_id) OR ((source_row.es_id IS NULL) AND (target_row.es_id IS NULL))))
+NOTICE:    Join Filter: (((source_row.es_id = target_row.es_id) OR ((source_row.es_id IS NULL) AND (target_row.es_id IS NULL))) AND ((source_row.lu_id = target_row.lu_id) OR ((source_row.lu_id IS NULL) AND (target_row.lu_id IS NULL))))
 NOTICE:    Buffers: local hit=2
 NOTICE:    ->  Seq Scan on source_with_eclipsed_flag source_row (actual rows=2.00 loops=1)
 NOTICE:          Buffers: local hit=1

native/src/executor_cache.rs

@@ -5,6 +5,8 @@ use std::hash::{Hash, Hasher};
 
 use pgrx::prelude::*;
 
+use crate::util::qi;
+
 /// Cached executor introspection state.
 /// Contains all metadata and SQL fragments needed by the PL/pgSQL executor,
 /// replacing ~570 lines of per-call introspection + CTE logic.
@@ -42,6 +44,24 @@ pub struct ExecutorCachedState {
     pub source_cols_hash: u64,
 }
 
+// Executor introspection cache — schema metadata for SQL fragment generation.
+//
+// Stores pre-computed SQL fragments (JOIN clauses, column lists, SET expressions)
+// derived from pg_attribute, sql_saga.era, and sql_saga.unique_keys catalog queries.
+// Contains NO row data, transaction state, or snapshot information — purely structural
+// metadata that changes only when DDL alters the table schema.
+//
+// Invalidation: Every call computes source_cols_hash by querying pg_attribute for
+// column names and types, then compares against the cached hash. Schema changes
+// (ALTER TABLE ADD/DROP COLUMN) produce a different hash → cache miss → full
+// re-introspection. Target table DDL is NOT independently hashed, but would cause
+// the executor's prepared statements to fail at execution time (column count/type
+// mismatch), surfacing as a clear PostgreSQL ERROR rather than silent corruption.
+//
+// Growth: One entry per distinct (target_table_oid, mode, identity_columns,
+// source_cols_hash) combination. Bounded by the number of target tables in the
+// application. Explicit reset available via temporal_merge_native_cache_reset().
+// No LRU eviction — acceptable for the batch ETL use case (small number of tables).
 thread_local! {
     /// Multi-entry cache keyed by config hash.
     pub static EXECUTOR_CACHE: RefCell<HashMap<u64, ExecutorCachedState>> = RefCell::new(HashMap::new());
@@ -90,11 +110,6 @@ fn hash_source_cols(client: &pgrx::spi::SpiClient, source_oid: u32) -> u64 {
     h.finish()
 }
 
-/// Helper: quote identifier (double-quote, escaping inner double-quotes).
-fn qi(name: &str) -> String {
-    format!("\"{}\"", name.replace('"', "\"\""))
-}
-
 /// Perform all executor introspection in a single SPI connection and return
 /// the cached state. On cache hit, returns immediately with zero SPI calls.
 #[pg_extern]
@@ -609,8 +624,30 @@ fn run_executor_introspection(
     })
 }
 
-/// Build the SQL query that replaces the column list CTE (lines 513-640).
-/// This is executed once on cache miss; the results are cached.
+/// Build the SQL query that replaces the column list CTE (lines 513-640 of the PL/pgSQL executor).
+/// Executed once on cache miss; results are cached in ExecutorCachedState.
+///
+/// CTEs:
+///   - **target_cols**: All non-dropped columns from the target table (pg_attribute).
+///   - **common_data_cols**: Data columns, excluding range/temporal, identity, lookup, PK,
+///     GENERATED ALWAYS, and serial (nextval) columns. These are the "payload" columns
+///     that appear in UPDATE SET clauses.
+///   - **all_available_cols**: Union of common_data_cols + lookup + identity + PK columns.
+///     This is the superset from which INSERT column lists are derived.
+///   - **cols_for_insert**: Columns for normal INSERT — all_available_cols minus
+///     insert-defaulted columns (GENERATED ALWAYS, synchronized temporal cols), but
+///     always keeping identity/lookup/PK columns even if they are defaulted.
+///   - **cols_for_founding_insert**: Columns for founding INSERT — all_available_cols minus
+///     founding-defaulted columns (GENERATED ALWAYS + IDENTITY BY DEFAULT), producing a
+///     more restrictive list that lets PostgreSQL generate serial IDs for new entities.
+///
+/// The final SELECT returns 6 string fragments (any may be NULL if no columns qualify):
+///   1. **update_set_clause**: `col = CASE WHEN p.data ? 'col' THEN (p.data->>'col')::type ELSE t.col END, ...`
+///   2. **all_cols_ident**: Comma-separated quoted column names for INSERT target list.
+///   3. **all_cols_select**: Comma-separated expressions reading from `jpr_all` record with COALESCE for NOT NULL defaulted.
+///   4. **all_cols_from_jsonb**: Comma-separated `(s.full_data->>'col')::type` expressions for INSERT from JSONB.
+///   5. **founding_all_cols_ident**: Column names for founding INSERT target list.
+///   6. **founding_all_cols_from_jsonb**: `(s.full_data->>'col')::type` expressions for founding INSERT from JSONB.
 fn build_column_list_cte_query(
     target_oid: u32,
     identity_columns: &[String],

native/src/lib.rs

@@ -13,9 +13,35 @@ mod introspect;
 mod reader;
 mod sweep;
 mod types;
+mod util;
 
 use types::{CachedState, DeleteMode, MergeMode, PlanRow};
 
+// SAFETY ANALYSIS — Per-connection caching via thread_local!
+//
+// PostgreSQL uses a process-per-connection model (one OS process per backend,
+// single-threaded). Rust's thread_local! maps to per-thread storage, making it
+// effectively per-connection. Each backend gets independent cache instances with
+// no possibility of cross-connection data races.
+// See: PostgreSQL docs §20.3 "Resource Consumption" — max_connections creates
+// separate server processes; src/backend/postmaster/postmaster.c:BackendStartup().
+//
+// Memory ownership: All cached data (String, Vec, HashMap) uses Rust's system
+// allocator, NOT PostgreSQL's palloc. This means cached data is NOT subject to
+// memory context resets. Rust's ownership system guarantees deallocation on Drop.
+// pgrx 0.16.x does not route standard Rust collections through palloc.
+// See: pgrx source — palloc is only used for PgBox/PgMemoryContexts, not std types.
+//
+// Cross-transaction safety: Caches store only schema-derived metadata (column names,
+// SQL templates, type information). No row data, transaction IDs, snapshot info, or
+// visibility information is cached. Schema metadata is stable across transactions
+// unless DDL occurs, which is detected by source_cols_hash comparison on every call.
+//
+// SPI data extraction: Every Spi::connect() closure extracts data as Rust-owned
+// String/Vec/i64 values. pgrx's row.get::<String>() internally calls text_to_cstring
+// and creates a Rust String — the SPI memory context can be safely freed after the
+// closure returns without invalidating any cached data.
+// See: pgrx source — spi/client.rs get::<String>() → FromDatum → String::from().
 thread_local! {
     /// Multi-entry cache keyed by config cache_key (target_table + mode + columns).
     /// Holds one CachedState per distinct temporal_merge configuration, enabling

native/src/reader.rs

@@ -4,7 +4,30 @@ use std::collections::HashMap;
 use pgrx::prelude::*;
 
 use crate::types::{CachedState, ColCategory, ColMapping, FilterParam, PlannerContext, SourceRow, TargetRow};
-
+use crate::util::qi;
+
+// Prepared statement caching — per-connection lifetime via SPI_keepplan.
+//
+// pgrx's PreparedStatement::keep() calls SPI_keepplan() (PostgreSQL src/backend/
+// executor/spi.c), which copies the plan from SPI's transient memory context into
+// CacheMemoryContext (a long-lived, top-level context). This makes plans survive:
+//   - SPI connection close/reopen (Spi::connect scope exit)
+//   - Transaction commit/rollback boundaries
+//   - Multiple temporal_merge calls within one session
+// Plans are freed only when the backend process exits.
+//
+// Automatic replanning: When the source temp view is recreated via CREATE OR REPLACE,
+// PostgreSQL's relcache invalidation (src/utils/cache/relcache.c:RelationCacheInvalidate)
+// detects the OID change and marks cached plans as invalid. The next SPI_execute_plan()
+// call transparently replans. This is why we don't need manual invalidation for
+// source view changes.
+// See: PostgreSQL docs §54.4 "SPI_keepplan" and src/utils/cache/plancache.c.
+//
+// Key format: Plans are keyed by the full SQL string. Different target tables or
+// source views produce different SQL, so each gets its own cached plan. Stale entries
+// (from schema changes that produced different SQL) become unreachable HashMap entries.
+// The number of entries is bounded by the number of distinct (table, view) pairs used
+// in the session — typically 1-10 for batch ETL workloads.
 thread_local! {
     /// Multi-entry cache keyed by target SQL template (one per target table config).
     static TARGET_READ_STMTS: RefCell<HashMap<String, pgrx::spi::OwnedPreparedStatement>> = RefCell::new(HashMap::new());
@@ -349,12 +372,16 @@ pub fn read_target_rows_with_sql(
         for row in table {
             let valid_from: String = row
                 .get::<String>(1)
-                .unwrap_or(Some(String::new()))
-                .unwrap_or_default();
+                .unwrap_or(None)
+                .unwrap_or_else(|| {
+                    pgrx::error!("sql_saga: NULL temporal boundary (valid_from) in target row");
+                });
             let valid_until: String = row
                 .get::<String>(2)
-                .unwrap_or(Some(String::new()))
-                .unwrap_or_default();
+                .unwrap_or(None)
+                .unwrap_or_else(|| {
+                    pgrx::error!("sql_saga: NULL temporal boundary (valid_until) in target row");
+                });
 
             let (identity_keys, lookup_keys, data_payload, ephemeral_payload, pk_payload) =
                 read_target_ordinals(&row, layout);
@@ -427,12 +454,16 @@ pub fn read_source_rows_cached(
                     .unwrap_or_default();
                 let valid_from: String = row
                     .get::<String>(3)
-                    .unwrap_or(Some(String::new()))
-                    .unwrap_or_default();
+                    .unwrap_or(None)
+                    .unwrap_or_else(|| {
+                        pgrx::error!("sql_saga: NULL temporal boundary (valid_from) in source row");
+                    });
                 let valid_until: String = row
                     .get::<String>(4)
-                    .unwrap_or(Some(String::new()))
-                    .unwrap_or_default();
+                    .unwrap_or(None)
+                    .unwrap_or_else(|| {
+                        pgrx::error!("sql_saga: NULL temporal boundary (valid_until) in source row");
+                    });
 
                 // Read individual columns by ordinal — no JSON parsing
                 let (identity_keys, lookup_keys, data_payload, ephemeral_payload,
@@ -774,12 +805,16 @@ pub fn read_target_rows_parameterized(
             for row in table {
                 let valid_from: String = row
                     .get::<String>(1)
-                    .unwrap_or(Some(String::new()))
-                    .unwrap_or_default();
+                    .unwrap_or(None)
+                    .unwrap_or_else(|| {
+                        pgrx::error!("sql_saga: NULL temporal boundary (valid_from) in target row");
+                    });
                 let valid_until: String = row
                     .get::<String>(2)
-                    .unwrap_or(Some(String::new()))
-                    .unwrap_or_default();
+                    .unwrap_or(None)
+                    .unwrap_or_else(|| {
+                        pgrx::error!("sql_saga: NULL temporal boundary (valid_until) in target row");
+                    });
 
                 let (identity_keys, lookup_keys, data_payload, ephemeral_payload, pk_payload) =
                     read_target_ordinals(&row, layout);
@@ -993,10 +1028,6 @@ fn build_target_filter(
 
 // ── Helpers ──
 
-fn qi(name: &str) -> String {
-    format!("\"{}\"", name.replace('"', "\"\""))
-}
-
 fn build_until_expr(
     alias: &str,
     has_range: bool,

native/src/types.rs

@@ -257,7 +257,9 @@ fn parse_temporal_numeric(s: &str) -> f64 {
     match s {
         "infinity" => f64::INFINITY,
         "-infinity" => f64::NEG_INFINITY,
-        _ => s.parse::<f64>().unwrap_or(0.0),
+        _ => s.parse::<f64>().unwrap_or_else(|_| {
+            pgrx::error!("sql_saga: cannot parse temporal numeric bound: {:?}", s);
+        }),
     }
 }
 

native/src/util.rs

@@ -0,0 +1,4 @@
+/// Helper: quote identifier (double-quote, escaping inner double-quotes).
+pub fn qi(name: &str) -> String {
+    format!("\"{}\"", name.replace('"', "\"\""))
+}