P2-T100a: full Spark 4.2 golden run, frozen baseline and ratchet

crates/krishiv-kernels/src/cast.rs

@@ -578,8 +578,9 @@ pub fn cast(
         ($o:ty, $t:ty) => {{
             let s = input.as_string::<$o>();
             let out = map_rows::<$t>(input, |i| {
-                // Out of the target's range is malformed too (Spark's `toInt`).
-                let v = text::parse_integral(s.value(i).as_bytes(), mode == CastMode::Ansi)
+                // Out of the target's range is malformed too (Spark's `toInt`);
+                // `try_cast` parses as ANSI does (`toLongExact`), so `'1.5'` is NULL.
+                let v = text::parse_integral(s.value(i).as_bytes(), mode != CastMode::Legacy)
                     .and_then(|v| match <$t as SparkIntegral>::narrow(i128::from(v)) {
                         (n, false) => Some(n),
                         (_, true) => None,

crates/krishiv-kernels/tests/casts.rs

@@ -123,7 +123,14 @@ fn string_to_numbers_match_spark() {
         one(s("it's"), int.clone(), Ansi),
         invalid("'it\\'s'", "INT")
     );
-    assert_eq!(one(s("abc"), int, Try), "NULL");
+    assert_eq!(one(s("abc"), int.clone(), Try), "NULL");
+    // `try_cast` parses as ANSI does: a fraction is malformed, not dropped
+    // (Spark 4.2 `try_cast.sql`: `TRY_CAST('1.23' AS int)` is NULL).
+    for to in [DataType::Int8, int.clone(), DataType::Int64] {
+        assert_eq!(one(s("1.23"), to.clone(), Try), "NULL");
+        assert_eq!(one(s("-4.56"), to, Try), "NULL");
+    }
+    assert_eq!(one(s(" 7 "), int, Try), "7");
     assert_eq!(
         one(s("128"), DataType::Int8, Ansi),
         invalid("'128'", "TINYINT")

crates/krishiv-sql/tests/analyzer.rs

@@ -684,6 +684,106 @@ fn repeated_correlation_keys_stay_unique() {
     assert!(a.plan.root().to_string().contains("Aggregate [a#3], "));
 }
 
+/// A foldable `LIMIT` that is negative or NULL fails analysis as Spark's
+/// `checkLimitLikeClause` does (it used to reach the planner).
+/// The `LIMIT` count of `plan`, wherever it sits.
+fn limit_of(plan: &Arc<LogicalPlan>) -> Option<Arc<Expr>> {
+    if let PlanNode::Limit { limit, .. } = &plan.node {
+        return Some(Arc::clone(limit));
+    }
+    let mut found = None;
+    plan.for_each_child(&mut |c| {
+        if found.is_none() {
+            found = limit_of(c);
+        }
+    });
+    found
+}
+
+#[test]
+fn negative_and_null_limits_fail_analysis() {
+    let g = goldens();
+    let err = |sql: &str| {
+        analyze(&g, sql, ResolutionConf::default())
+            .unwrap_err()
+            .error
+    };
+    let e = err("SELECT a FROM t LIMIT -1");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NEGATIVE"
+    );
+    assert_eq!(e.sql_state(), Some("42K0E"));
+    assert!(e.message().contains("\"-1\""), "{}", e.message());
+    assert!(e.message().contains("but got -1"), "{}", e.message());
+    let e = err("SELECT a FROM t LIMIT CAST(NULL AS INT)");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL"
+    );
+    let a = analyze(&g, "SELECT a FROM t LIMIT 0", ResolutionConf::default()).unwrap();
+    let limit = limit_of(a.plan.root()).expect("a Limit node");
+    assert!(
+        matches!(&limit.node, ExprNode::Literal(Literal::Integer(0))),
+        "{limit:?}"
+    );
+}
+
+/// `LIMIT` folds arithmetic before Spark's checks, and rejects counts that
+/// cannot be evaluated without rows.
+#[test]
+fn limit_folds_arithmetic_and_rejects_unfoldable_counts() {
+    let g = goldens();
+    let err = |sql: &str| {
+        analyze(&g, sql, ResolutionConf::default())
+            .unwrap_err()
+            .error
+    };
+    // `1 - 2` is a foldable negative count, not a passing limit.
+    let e = err("SELECT a FROM t LIMIT 1 - 2");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NEGATIVE"
+    );
+    assert_eq!(e.sql_state(), Some("42K0E"));
+    assert!(e.message().contains("but got -1"), "{}", e.message());
+    // `NULL + 1` stays NULL, and a cast of arithmetic folds through the kernels.
+    let e = err("SELECT a FROM t LIMIT CAST(NULL AS INT) + 1");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL"
+    );
+    let e = err("SELECT a FROM t LIMIT CAST(1 - 2 AS INT)");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NEGATIVE"
+    );
+    assert!(e.message().contains("but got -1"), "{}", e.message());
+    let e = err("SELECT a FROM t LIMIT CAST(CAST(NULL AS INT) + 1 AS INT)");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL"
+    );
+    // Unary minus over a constant folds as well.
+    let e = err("SELECT a FROM t LIMIT -(1)");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NEGATIVE"
+    );
+    let e = err("SELECT a FROM t LIMIT -(CAST(NULL AS INT))");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_NULL"
+    );
+    // A scalar subquery binds but is not foldable.
+    let e = err("SELECT a FROM t LIMIT (SELECT 1)");
+    assert_eq!(
+        e.error_class().name(),
+        "INVALID_LIMIT_LIKE_EXPRESSION.IS_UNFOLDABLE"
+    );
+    assert_eq!(e.sql_state(), Some("42K0E"));
+}
+
 /// The window-function goldens (`tests/fixtures/spark-4.2.0/window-goldens.json`,
 /// generated from a real Spark run) resolve to Spark's output names, types
 /// and nullability over the same schemas, and their resolved plans are a

crates/krishiv-testkit/CLAUDE.md

@@ -8,7 +8,12 @@ Conformance harness: Spark `hash()`/`xxhash64()` golden fixtures (`tests/fixture
 the `SQLQueryTestSuite` golden runner over `tests/spark-golden/<tag>/` (`.sql.out` parser, `Engine` trait,
 three-number report, exclusion manifest `tests/spark-golden/exclusions.json`; `cargo xtask golden` writes
 `target/golden-report.json`; `--engine sql-parse` measures the `krishiv-sql` parser on the DQL statements,
-`golden::sql_parse`; no CI ratchet yet), the batch == bounded-stream invariant
+`golden::sql_parse`; `--engine session` runs every statement through `krishiv-session`, `golden::session_engine`),
+the frozen statement-level baseline `tests/spark-golden/known-failures.json` (`golden::known_failures`: every failing
+statement with a category — unsupported feature with its plan task, divergence with ADR 0023, or bug with its owner;
+classification rules in the same file, `golden::classify`) and its CI ratchet (test `golden_ratchet`: a newly failing
+statement fails, a listed statement that passes must be removed; refresh with
+`cargo run -p krishiv-testkit --bin golden-report -- --engine session --ratchet update`), the batch == bounded-stream invariant
 harness, a sqllogictest runner over `tests/sqllogictest/`, and the Spark function matrix (P2-T106a):
 `tests/fixtures/spark-<ver>/function-matrix/` (generated by `tests/scripts/gen_function_matrix.py` against the
 Spark container), loaded by `fn_matrix`, with the coverage dashboard from

crates/krishiv-testkit/Cargo.toml

@@ -29,6 +29,7 @@ krishiv-memory = { path = "../krishiv-memory" }
 krishiv-plan = { path = "../krishiv-plan" }
 object_store = { workspace = true, features = ["fs"] }
 parquet = { workspace = true, features = ["arrow"] }
+krishiv-session = { path = "../krishiv-session", features = ["unstable-ir"] }
 krishiv-sql = { path = "../krishiv-sql" }
 krishiv-types = { path = "../krishiv-types" }
 regex-lite.workspace = true

crates/krishiv-testkit/src/bin/golden-report.rs

@@ -2,35 +2,71 @@
 //! Runs the Spark `SQLQueryTestSuite` golden files against an engine and writes the report
 //! (three numbers, per-file denominators, exclusion manifest) as JSON.
 //!
-//! `cargo run -p krishiv-testkit --bin golden-report -- [--tag TAG] [--jdk 17|21] [--manifest FILE] [--out FILE]`
-//! (or `cargo xtask golden`). The default engine is the reject-all stub (0/0/0 with the real
-//! denominators); `--engine sql-parse` measures the `krishiv-sql` parser on the DQL statements
-//! ([`krishiv_testkit::golden::sql_parse`]).
+//! `cargo run -p krishiv-testkit --bin golden-report -- [--tag TAG] [--jdk 17|21] [--manifest FILE]
+//! [--out FILE] [--engine reject-all|sql-parse|session] [--cases FILE] [--ratchet check|update]
+//! [--known-failures FILE]` (or `cargo xtask golden`).
+//!
+//! * The default engine is the reject-all stub (0/0/0 with the real denominators);
+//!   `--engine sql-parse` measures the `krishiv-sql` parser on the DQL statements
+//!   ([`krishiv_testkit::golden::sql_parse`]); `--engine session` runs every statement through
+//!   `krishiv-session` ([`krishiv_testkit::golden::session_engine`]).
+//! * `--cases FILE` writes every statement that did not pass, with why.
+//! * `--ratchet check` compares the run with the frozen baseline
+//!   (`tests/spark-golden/known-failures.json`) and fails on any difference;
+//!   `--ratchet update` rewrites the baseline: passing statements leave it, new failures are
+//!   classified by its rules (the command fails, writing nothing, if a failure matches no rule).
 #![forbid(unsafe_code)]
 
 use std::io::Write as _;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::process::ExitCode;
 
+use krishiv_testkit::golden::known_failures::KnownFailures;
 use krishiv_testkit::golden::manifest::ExclusionManifest;
-use krishiv_testkit::golden::run::{RejectAllEngine, run_with};
+use krishiv_testkit::golden::run::{Engine, RejectAllEngine, RunResult, run_with};
+use krishiv_testkit::golden::session_engine::SessionEngine;
 use krishiv_testkit::golden::sql_parse::{ParseOnlyEngine, in_dql_scope};
 use krishiv_testkit::golden::{DEFAULT_TAG, GoldenSet, Jdk, report};
 
-const USAGE: &str = "usage: golden-report [--tag TAG] [--jdk 17|21] [--manifest FILE] [--out FILE] [--engine reject-all|sql-parse]";
+const USAGE: &str = "usage: golden-report [--tag TAG] [--jdk 17|21] [--manifest FILE] [--out FILE] \
+[--engine reject-all|sql-parse|session] [--cases FILE] [--ratchet check|update] [--known-failures FILE]";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Ratchet {
+    Off,
+    Check,
+    Update,
+}
+
+fn ratchet(
+    mode: Ratchet,
+    path: &Path,
+    set: &GoldenSet,
+    result: &RunResult,
+) -> Result<String, String> {
+    match mode {
+        Ratchet::Check => KnownFailures::check_file(path, result),
+        Ratchet::Update => KnownFailures::update_file(path, result, set),
+        Ratchet::Off => Ok(String::new()),
+    }
+}
 
 fn run_cli() -> Result<(), String> {
     let mut tag = DEFAULT_TAG.to_owned();
     let mut manifest_path = ExclusionManifest::default_path();
+    let mut known_path = KnownFailures::default_path();
     let mut out = None;
     let mut jdk = Jdk::default();
-    let mut sql_parse = false;
+    let mut engine_name = "reject-all".to_owned();
+    let mut cases_out = None;
+    let mut mode = Ratchet::Off;
     let mut args = std::env::args().skip(1);
     while let Some(flag) = args.next() {
         let value = args.next().ok_or_else(|| USAGE.to_owned())?;
         match flag.as_str() {
             "--tag" => tag = value,
             "--manifest" => manifest_path = PathBuf::from(value),
+            "--known-failures" => known_path = PathBuf::from(value),
             "--jdk" => {
                 jdk = match value.as_str() {
                     "17" => Jdk::Java17,
@@ -39,43 +75,63 @@ fn run_cli() -> Result<(), String> {
                 }
             }
             "--out" => out = Some(PathBuf::from(value)),
-            "--engine" => {
-                sql_parse = match value.as_str() {
-                    "reject-all" => false,
-                    "sql-parse" => true,
+            "--engine" => match value.as_str() {
+                "reject-all" | "sql-parse" | "session" => engine_name = value,
+                _ => return Err(USAGE.to_owned()),
+            },
+            "--cases" => cases_out = Some(PathBuf::from(value)),
+            "--ratchet" => {
+                mode = match value.as_str() {
+                    "check" => Ratchet::Check,
+                    "update" => Ratchet::Update,
                     _ => return Err(USAGE.to_owned()),
                 }
             }
             _ => return Err(USAGE.to_owned()),
         }
     }
+    if mode != Ratchet::Off && engine_name != "session" {
+        return Err("--ratchet needs --engine session".to_owned());
+    }
     let manifest = ExclusionManifest::load(&manifest_path).map_err(|e| e.to_string())?;
     let dir = GoldenSet::tag_dir(&tag);
     let set = GoldenSet::scan_for(&dir, jdk).map_err(|e| format!("{}: {e}", dir.display()))?;
-    let result = if sql_parse {
-        run_with(
+    let result = match engine_name.as_str() {
+        "sql-parse" => run_with(
             &mut ParseOnlyEngine::default(),
             &set,
             &manifest,
             &in_dql_scope,
-        )
-    } else {
-        run_with(&mut RejectAllEngine, &set, &manifest, &|_| true)
+        ),
+        "session" => {
+            let mut engine = SessionEngine::new().map_err(|e| e.message)?;
+            run_with(&mut engine as &mut dyn Engine, &set, &manifest, &|_| true)
+        }
+        _ => run_with(&mut RejectAllEngine, &set, &manifest, &|_| true),
     }
     .map_err(|e| e.to_string())?;
-    let json = report::to_json_string(&report::to_json(&result, &tag, &manifest))
-        .map_err(|e| e.to_string())?;
     let mut stdout = std::io::stdout().lock();
     stdout
         .write_all(report::summary(&result).as_bytes())
         .map_err(|e| e.to_string())?;
+    if let Some(path) = &cases_out {
+        let text =
+            report::to_json_string(&report::cases_json(&result)).map_err(|e| e.to_string())?;
+        std::fs::write(path, text).map_err(|e| format!("{}: {e}", path.display()))?;
+    }
     if let Some(path) = out {
+        let json = report::to_json_string(&report::to_json(&result, &tag, &manifest))
+            .map_err(|e| e.to_string())?;
         if let Some(parent) = path.parent() {
             std::fs::create_dir_all(parent).map_err(|e| format!("{}: {e}", parent.display()))?;
         }
         std::fs::write(&path, json).map_err(|e| format!("{}: {e}", path.display()))?;
         writeln!(stdout, "report written to {}", path.display()).map_err(|e| e.to_string())?;
     }
+    if mode != Ratchet::Off {
+        let msg = ratchet(mode, &known_path, &set, &result)?;
+        writeln!(stdout, "{msg}").map_err(|e| e.to_string())?;
+    }
     Ok(())
 }