[validation-only] isolate explicit SQLite URI from unrelated DATABASE_URL (#912)

.claude/skills/dataprof/reference/api.md

@@ -37,7 +37,7 @@ Keyword arguments: `engine`, `chunk_size`, `memory_limit_mb`, `format`,
 `max_rows`, `name`, `csv_delimiter`, `csv_flexible`, `jsonl_on_error`,
 `sampling`, `stop_condition`, `on_progress`, `progress_interval_ms`,
 `quality_dimensions`, `metrics`, `locale`, `positive_columns`,
-`identifier_columns`, `temporal_columns`.
+`identifier_columns`, `temporal_columns`, `temporal_pairs`.
 
 ## Profiler builder (when keyword arguments are not enough)
 
@@ -56,7 +56,8 @@ report = (
 Every builder method returns the profiler, so calls chain. Other methods:
 `chunk_size`, `memory_limit_mb`, `format`, `name`, `csv_delimiter`,
 `csv_flexible`, `sampling`, `stop_condition`, `stop_when`, `on_progress`,
-`progress_interval_ms`, `locale`, `positive_columns`, `temporal_columns`.
+`progress_interval_ms`, `locale`, `positive_columns`, `temporal_columns`,
+`temporal_pairs`.
 
 ## ProfileReport: execution metadata
 

.claude/skills/dataprof/reference/interpretation.md

@@ -181,7 +181,8 @@ with every locale-specific pattern suppressed.
 `before.compare(after)` returns deltas. Three caveats:
 
 - Comparing a sampled profile to a full one measures your sampling, not drift.
-  Check `sampling_applied` on both sides first.
+  Each score's `evidence` and `reports.a`/`reports.b` `evidence` say whether a
+  side was truncated, sampled or scored from a sample; read them first.
 - Comparing profiles with different `metrics` selections produces absence, not
   change. Narrow both sides the same way.
 - A stored report can predate a change in how something is measured: text

crates/dataprof-core/src/progress.rs

@@ -52,6 +52,17 @@ impl ProgressSink {
         !matches!(self, Self::None)
     }
 
+    /// Reserve channel capacity for a terminal event without changing the
+    /// public shape of the Channel variant.
+    #[doc(hidden)]
+    #[cfg(feature = "async-streaming")]
+    pub fn reserve_terminal(&self) -> Option<tokio::sync::mpsc::OwnedPermit<ProgressEvent>> {
+        match self {
+            Self::Channel(tx) => tx.clone().try_reserve_owned().ok(),
+            _ => None,
+        }
+    }
+
     /// Deliver one event. A full channel drops it rather than block profiling.
     pub fn emit(&self, event: ProgressEvent) {
         match self {

crates/dataprof-core/src/semantic.rs

@@ -12,6 +12,8 @@ pub enum SemanticHintKind {
     Identifier,
     /// A [`SemanticHints::temporal_columns`] entry.
     Temporal,
+    /// A column named by a [`SemanticHints::temporal_pairs`] entry.
+    TemporalPair,
 }
 
 impl SemanticHintKind {
@@ -21,6 +23,7 @@ impl SemanticHintKind {
             Self::Positive => "positive_columns",
             Self::Identifier => "identifier_columns",
             Self::Temporal => "temporal_columns",
+            Self::TemporalPair => "temporal_pairs",
         }
     }
 }
@@ -66,6 +69,8 @@ pub struct SemanticHints {
     pub positive_columns: Vec<String>,
     pub identifier_columns: Vec<String>,
     pub temporal_columns: Vec<String>,
+    /// Start/end column pairs declared by the caller, compared as given.
+    pub temporal_pairs: Vec<(String, String)>,
 }
 
 impl SemanticHints {
@@ -74,6 +79,7 @@ impl SemanticHints {
             positive_columns,
             identifier_columns,
             temporal_columns: Vec::new(),
+            temporal_pairs: Vec::new(),
         }
     }
 
@@ -82,10 +88,18 @@ impl SemanticHints {
         self
     }
 
+    /// Declare start/end column pairs for timeliness to compare as given,
+    /// such as `valid_from`/`valid_until`, which no role word pairs by name.
+    pub fn with_temporal_pairs(mut self, temporal_pairs: Vec<(String, String)>) -> Self {
+        self.temporal_pairs = temporal_pairs;
+        self
+    }
+
     pub fn is_empty(&self) -> bool {
         self.positive_columns.is_empty()
             && self.identifier_columns.is_empty()
             && self.temporal_columns.is_empty()
+            && self.temporal_pairs.is_empty()
     }
 
     pub fn is_identifier_column(&self, column_name: &str) -> bool {
@@ -106,7 +120,21 @@ impl SemanticHints {
             .any(|candidate| candidate == column_name)
     }
 
-    /// Every `(kind, column_name)` pair across all three hint lists.
+    /// Each column named by a declared pair, once, in the order first named.
+    pub fn temporal_pair_columns(&self) -> Vec<&str> {
+        let mut columns: Vec<&str> = Vec::new();
+        for (start, end) in &self.temporal_pairs {
+            for name in [start.as_str(), end.as_str()] {
+                if !columns.contains(&name) {
+                    columns.push(name);
+                }
+            }
+        }
+        columns
+    }
+
+    /// Every `(kind, column_name)` pair across the hint lists, with each column
+    /// of a declared pair listed once.
     pub fn iter(&self) -> impl Iterator<Item = (SemanticHintKind, &str)> {
         let positive = self
             .positive_columns
@@ -120,7 +148,11 @@ impl SemanticHints {
             .temporal_columns
             .iter()
             .map(|c| (SemanticHintKind::Temporal, c.as_str()));
-        positive.chain(identifier).chain(temporal)
+        let pairs = self
+            .temporal_pair_columns()
+            .into_iter()
+            .map(|c| (SemanticHintKind::TemporalPair, c));
+        positive.chain(identifier).chain(temporal).chain(pairs)
     }
 
     /// Fail loudly when a hint names a column that is not in the schema.
@@ -129,6 +161,14 @@ impl SemanticHints {
     /// hint that cannot bind must not vanish: a typo'd or stale column name is a
     /// mistake the profiler should surface, not silently discard.
     pub fn validate_names(&self, column_names: &[&str]) -> Result<(), DataProfilerError> {
+        // A column compared with itself can never be out of order, so such a
+        // pair would report a clean check that never happened.
+        if let Some((name, _)) = self.temporal_pairs.iter().find(|(start, end)| start == end) {
+            return Err(DataProfilerError::InvalidSemanticHint {
+                message: format!("temporal_pairs names '{name}' as both start and end"),
+                suggestion: "Pair a start column with a different end column.".to_string(),
+            });
+        }
         let mut unknown: Vec<(SemanticHintKind, &str)> = self
             .iter()
             .filter(|(_, name)| !column_names.contains(name))
@@ -167,14 +207,17 @@ impl SemanticHints {
     /// effective while producing neither metrics nor binding evidence.
     pub fn validate_quality_usage(&self, quality_computed: bool) -> Result<(), DataProfilerError> {
         if quality_computed
-            || (self.positive_columns.is_empty() && self.temporal_columns.is_empty())
+            || (self.positive_columns.is_empty()
+                && self.temporal_columns.is_empty()
+                && self.temporal_pairs.is_empty())
         {
             return Ok(());
         }
 
         Err(DataProfilerError::InvalidSemanticHint {
-            message: "positive_columns and temporal_columns require the Quality metric pack"
-                .to_string(),
+            message:
+                "positive_columns, temporal_columns and temporal_pairs require the Quality metric pack"
+                    .to_string(),
             suggestion: "Include MetricPack::Quality (\"quality\" in Python), or remove the value-driven hint. Identifier hints remain usable without quality because they affect column typing."
                 .to_string(),
         })
@@ -210,9 +253,11 @@ impl SemanticHints {
 
         Err(DataProfilerError::InvalidSemanticHint {
             message: format!("semantic hint(s) bound to no matching value: {details}"),
-            suggestion: "positive_columns need numeric values; temporal_columns need date values. \
+            suggestion:
+                "positive_columns need numeric values; temporal_columns and temporal_pairs \
+                 need date values. \
                  Fix the column choice or remove the hint."
-                .to_string(),
+                    .to_string(),
         })
     }
 }

crates/dataprof-db/src/connectors/sqlite.rs

@@ -69,11 +69,20 @@ impl DatabaseConnector for SqliteConnector {
     async fn connect(&mut self) -> Result<(), DataProfilerError> {
         #[cfg(feature = "sqlite")]
         {
-            let db_path = self.get_db_path()?;
-            let connection_string = if db_path == ":memory:" {
-                "sqlite::memory:".to_string()
+            // sqlx parses SQLite-specific URL options (mode, cache, immutable,
+            // vfs). Passing the original sqlite: URI through unchanged also
+            // preserves its absolute/relative path semantics (#912 / #839).
+            // Previously rebuilding the URL from get_db_path() silently
+            // dropped every query parameter, including mode=ro and mode=rwc.
+            let connection_string = if self.config.connection_string.starts_with("sqlite:") {
+                self.config.connection_string.clone()
             } else {
-                format!("sqlite://{}", db_path)
+                let db_path = self.get_db_path()?;
+                if db_path == ":memory:" {
+                    "sqlite::memory:".to_string()
+                } else {
+                    format!("sqlite://{}", db_path)
+                }
             };
 
             let pool = SqlitePoolOptions::new()

crates/dataprof-db/src/lib.rs

@@ -144,7 +144,11 @@ fn apply_environment_configuration(
                 .apply_to_connection_string(config.connection_string.clone(), database_type);
         }
 
-        if config.load_credentials_from_env {
+        if config.load_credentials_from_env && database_type != "sqlite" {
+            // An explicit SQLite URI is a local file target, not a consumer of
+            // generic DATABASE_URL / DB_* credentials from other engines.
+            // Rebuilding it with foreign settings may replace its path or
+            // introduce unsupported user-info/host components (#912).
             let credentials = DatabaseCredentials::from_environment(database_type);
             config.connection_string =
                 credentials.apply_to_connection_string(&config.connection_string);
@@ -191,8 +195,8 @@ pub async fn analyze_database_with_options(
     // ignored. Checked before connecting so nothing is scanned first.
     if !options.semantic_hints().is_empty() {
         return Err(DataProfilerError::UnsupportedDataSource {
-            message: "positive_columns, identifier_columns, and temporal_columns are not \
-                      supported for database profiling yet"
+            message: "positive_columns, identifier_columns, temporal_columns, and temporal_pairs \
+                      are not supported for database profiling yet"
                 .to_string(),
         });
     }
@@ -280,10 +284,16 @@ pub async fn analyze_database_with_options(
     // from value-level checks, and the columns no longer line up.
     let mut uniqueness = RowUniquenessTracker::default();
     if !aligned.is_empty() {
+        let container_rows: Vec<&[usize]> = columns.container_rows().collect();
         for row_index in 0..actual_rows_processed {
             let mut signature = RowSignature::default();
-            for cells in &aligned {
-                signature.push_field(&cells[row_index]);
+            for (cells, containers) in aligned.iter().zip(&container_rows) {
+                // Ascending by construction: rows are pushed in order.
+                if containers.binary_search(&row_index).is_ok() {
+                    signature.push_container_field(&cells[row_index]);
+                } else {
+                    signature.push_field(&cells[row_index]);
+                }
             }
             uniqueness.observe(signature.finish());
         }

crates/dataprof-db/src/query_columns.rs

@@ -18,9 +18,9 @@ use std::ops::Index;
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct QueryColumns {
     columns: Vec<(String, Vec<String>)>,
-    /// Per column, by position, how many values were a JSON object or array or
-    /// a SQL array. Always as long as `columns`.
-    containers: Vec<usize>,
+    /// Per column, by position, the rows whose value was a JSON object or array
+    /// or a SQL array, ascending. Always as long as `columns`.
+    containers: Vec<Vec<usize>>,
 }
 
 impl QueryColumns {
@@ -58,8 +58,10 @@ impl QueryColumns {
     /// container: a JSON object or array, or a SQL array.
     pub fn push_decoded(&mut self, index: usize, value: String, container: bool) {
         if let Some((_, data)) = self.columns.get_mut(index) {
+            if container {
+                self.containers[index].push(data.len());
+            }
             data.push(value);
-            self.containers[index] += usize::from(container);
         }
     }
 
@@ -69,11 +71,18 @@ impl QueryColumns {
         self.columns
             .iter()
             .zip(&self.containers)
-            .filter(|(_, count)| **count > 0)
-            .map(|((name, _), count)| (name.clone(), *count))
+            .filter(|(_, rows)| !rows.is_empty())
+            .map(|((name, _), rows)| (name.clone(), rows.len()))
             .collect()
     }
 
+    /// Per column, in query order, the rows whose value was a container,
+    /// ascending. Row signatures tag those values apart from a string that
+    /// spells the same text (#945).
+    pub fn container_rows(&self) -> impl Iterator<Item = &[usize]> {
+        self.containers.iter().map(Vec::as_slice)
+    }
+
     /// Number of columns.
     pub fn len(&self) -> usize {
         self.columns.len()
@@ -155,7 +164,7 @@ impl Index<&str> for QueryColumns {
 impl FromIterator<(String, Vec<String>)> for QueryColumns {
     fn from_iter<I: IntoIterator<Item = (String, Vec<String>)>>(iter: I) -> Self {
         let columns: Vec<_> = iter.into_iter().collect();
-        let containers = vec![0; columns.len()];
+        let containers = vec![Vec::new(); columns.len()];
         Self {
             columns,
             containers,