From 6feb1493001f70fc4d72490716b7ee487b2913f7 Mon Sep 17 00:00:00 2001 From: Jimmy Kirk Date: Sat, 15 Aug 2026 20:24:02 -0500 Subject: [PATCH] fix(schema): emit one coslice witness per ordinal instead of a cross product 'cosliceClosure' emitted every shortest leg once for every target reachable past it, on the stated assumption that this was "bounded 2x". It is not. The row count is |legs on shortest paths| * |targets reachable past them| which grows with graph density, not with path length. On a 300-object corpus it reached 29,786,641 rows for ~106k (seed, target) pairs -- 281 rows per pair. Nothing consumed that fan-out. 'materializeDecompositionCoslice' ranks with ROW_NUMBER() PARTITION BY (seed, target, direction, ordinal) ORDER BY (leg_from, leg_to) and keeps rn = 1, so all but one row per ordinal was built only to be discarded -- after a window function had sorted all 29.8M of them and joined them five ways. That is what exhausted memory. This picks the same witness at emission time, in Haskell, using the same total order. Output is unchanged; the intermediate no longer exists. Measured on a fixed 150-object input, old build vs new: path_leg_fwd 41,332 -> 10,961 decomposition_coslice 10,939 == 10,939 content md5 identical reaches, column_risk, live_proc, dead_code, taint_paths, schema_objects all identical At 300 objects, where the old build could not finish at all: path_leg_fwd 29,786,641 -> 617,296 (48x) wall clock OOM / SIGSEGV -> 49s The existing diamond regression test bounded the number of distinct *targets*, which never grew -- the blow-up was in rows *per* target. The added test asserts the invariant the materializer actually relies on: at most one row per (seed, target, ordinal). --- compiler/src/PB/Analysis/SchemaClosure.hs | 72 ++++++++++++++++------- compiler/test/SchemaClosureTest.hs | 30 ++++++++++ 2 files changed, 82 insertions(+), 20 deletions(-) diff --git a/compiler/src/PB/Analysis/SchemaClosure.hs b/compiler/src/PB/Analysis/SchemaClosure.hs index 141d73ac..bd892b49 100644 --- a/compiler/src/PB/Analysis/SchemaClosure.hs +++ b/compiler/src/PB/Analysis/SchemaClosure.hs @@ -13,10 +13,15 @@ -- graph with a small seed set). Includes self-pairs via cycles (a node in -- a cycle reaches itself). -- * 'cosliceClosure' is the forward + backward shortest-path witness --- reconstruction: for every seed it emits EVERY shortest leg on a path --- to a target (set semantics through a diamond, bounded 2x), so the --- downstream 'PB.Pipeline.DuckDb.materializeDecompositionCoslice' --- ROW_NUMBER tie-break picks one witness per ordinal. +-- reconstruction: for every (seed, target, ordinal) it emits the ONE +-- witness leg that 'PB.Pipeline.DuckDb.materializeDecompositionCoslice' +-- keeps — least (leg_from, leg_to), matching that materializer's +-- ROW_NUMBER tie-break exactly. +-- +-- It used to emit every shortest leg and leave the choice to SQL, on the +-- stated assumption that this was "bounded 2x". It is not: the row count +-- is |legs on shortest paths| * |targets reachable past them|, which +-- grows with graph density. See 'emitOneWitnessPerOrdinal'. -- -- All three closures are built on the single 'PB.Algebra.Closure.reachFrom' -- primitive (Boolean for reachability, min-plus for hop distances) — there is @@ -214,37 +219,64 @@ fwdForSeed :: Text -> Map Text [(Text, Text)] -> Map Text (Set Text) -> Map Text Int -> [[Text]] fwdForSeed s adjFwd reach dist = let legs = [ (x, y, k) | (x, ns) <- Map.toList adjFwd, (y, k) <- ns ] - emit (lf, lt, k) = + contribs (lf, lt, k) = case (Map.lookup lf dist, Map.lookup lt dist) of (Just o, Just o') | o' == o + 1 -> - let finalHop = [s, lt, T.pack (show o), lf, lt, k] - inter = [ [s, t, T.pack (show o), lf, lt, k] - | t <- Set.toList (Map.findWithDefault Set.empty lt reach) - , Just dt <- [Map.lookup t dist] - , dt > o + 1 ] - in finalHop : inter + [ ((t, o), (lf, lt, k)) + | t <- lt : [ t' | t' <- Set.toList (Map.findWithDefault Set.empty lt reach) + , Just dt <- [Map.lookup t' dist] + , dt > o + 1 ] + ] _ -> [] - in concatMap emit legs + in emitOneWitnessPerOrdinal s (concatMap contribs legs) backForSeed :: Text -> Map Text [(Text, Text)] -> Map Text (Set Text) -> Map Text Int -> [[Text]] backForSeed s adjFwd revReach dist = let legs = [ (x, y, k) | (x, ns) <- Map.toList adjFwd, (y, k) <- ns ] - emit (lf, lt, k) = + contribs (lf, lt, k) = case (Map.lookup lt dist, Map.lookup lf dist) of (Just o, Just o') | o' == o + 1 -> -- lf is the FAR node (dist o+1), lt the NEAR node (dist o); the -- backward head binds target = lf (path_leg_back(s, t, o, t, lt, kind) -- with leg(t, lt)). - let finalHop = [s, lf, T.pack (show o), lf, lt, k] - inter = [ [s, t, T.pack (show o), lf, lt, k] - | t <- Set.toList (Map.findWithDefault Set.empty lf revReach) - , Just dt <- [Map.lookup t dist] - , dt > o + 1 ] - in finalHop : inter + [ ((t, o), (lf, lt, k)) + | t <- lf : [ t' | t' <- Set.toList (Map.findWithDefault Set.empty lf revReach) + , Just dt <- [Map.lookup t' dist] + , dt > o + 1 ] + ] _ -> [] - in concatMap emit legs + in emitOneWitnessPerOrdinal s (concatMap contribs legs) + +-- | Collapse witness candidates to the single one the consumer keeps. +-- +-- 'materializeDecompositionCoslice' ranks with +-- +-- ROW_NUMBER() OVER (PARTITION BY seed_key, target_key, direction, leg_ordinal +-- ORDER BY leg_from, leg_to) +-- +-- and takes @rn = 1@ — so per (seed, target, ordinal) exactly one leg +-- survives, the minimum by (leg_from, leg_to). Choosing it here instead of +-- emitting every candidate and discarding the rest in SQL is what keeps this +-- linear in the output rather than in the cross product. +-- +-- The old code emitted one row per (shortest leg x reachable target), which +-- the module header described as "bounded 2x". That bound does not hold: the +-- count is |legs on shortest paths| * |targets reachable past them|, so it +-- grows with graph density, not with path length. On a real 300-object corpus +-- it reached 29.8 million rows for ~106k (seed, target) pairs — 281 per pair — +-- and the downstream window function over that is what exhausted memory. +emitOneWitnessPerOrdinal :: Text -> [((Text, Int), (Text, Text, Text))] -> [[Text]] +emitOneWitnessPerOrdinal s cands = + [ [s, t, T.pack (show o), lf, lt, k] + | ((t, o), (lf, lt, k)) <- Map.toList (Map.fromListWith minWitness cands) + ] + where + -- Matches the SQL tie-break exactly: least (leg_from, leg_to). + minWitness a@(lf1, lt1, _) b@(lf2, lt2, _) + | (lf1, lt1) <= (lf2, lt2) = a + | otherwise = b -- | Materialize @reaches@, @path_leg_fwd@, @path_leg_back@ as real DuckDB -- tables, computed by 'legPriority' / 'reachClosure' / diff --git a/compiler/test/SchemaClosureTest.hs b/compiler/test/SchemaClosureTest.hs index 378225a2..bc451ba4 100644 --- a/compiler/test/SchemaClosureTest.hs +++ b/compiler/test/SchemaClosureTest.hs @@ -128,5 +128,35 @@ tests = testGroup "SchemaClosure (closures, production)" targetCount = Set.size (Set.fromList [ t | [_, t, _, _, _, _] <- fwd ]) in assertBool ("target count " <> show targetCount <> " should stay <= object count " <> show objCount) (targetCount <= objCount) + + , testCase "one witness per (seed, target, ordinal) -- no cross-product fan-out" $ + -- The assertion the diamond case above was missing. It bounded the + -- number of distinct *targets*, which never grew; the blow-up was in + -- rows *per* target, because every shortest leg was emitted once for + -- every target reachable past it. On a real corpus that reached 281 + -- rows per (seed, target) pair and 29.8M rows overall. + -- + -- 'materializeDecompositionCoslice' keeps exactly one row per + -- (seed, target, direction, ordinal) via ROW_NUMBER, so anything + -- beyond one here is built only to be discarded. + let n = 15 :: Int + tbl (i :: Int) s = "t" <> T.pack (show i) <> s + colName = "id" + seedKey = "col:" <> tbl 0 "" <> "." <> colName + layerFks i = + [ CatFkRow Nothing (tbl i "") colName Nothing (tbl i "a") colName + , CatFkRow Nothing (tbl i "") colName Nothing (tbl i "b") colName + , CatFkRow Nothing (tbl i "a") colName Nothing (tbl (i+1) "") colName + , CatFkRow Nothing (tbl i "b") colName Nothing (tbl (i+1) "") colName + ] + inp = emptyInputs { inCatalogFks = concatMap layerFks [0 .. n-1] } + leg = legRowsOf (sgLegs (buildSchema inp)) + (fwd, back) = cosliceClosure [seedKey] leg + keysOf rows = [ (t, o) | [_, t, o, _, _, _] <- rows ] + dupes rows = length (keysOf rows) - Set.size (Set.fromList (keysOf rows)) + in do assertBool ("forward emitted " <> show (dupes fwd) <> " duplicate (target, ordinal) rows") + (dupes fwd == 0) + assertBool ("backward emitted " <> show (dupes back) <> " duplicate (target, ordinal) rows") + (dupes back == 0) ] ]