Repository navigation
Conversation
PythonEvalType gains Spark's relation-kind codes (201-217, 251-254, 300-302; pyspark/util.py at 4.2.0) with a kind classifier and is_scalar. The logical IR gains MapPartitions, GroupMap, CoGroupMap, PythonTableFunction (with its TABLE argument) and GeneratorKind::PythonUdtf; each references its UDF by PythonUdfRef, never the callable. Walkers, printer, canonical encoding and cache fingerprint cover them. An aggregate mixing a grouped-aggregate Python UDF with another aggregate is refused with Spark's INVALID_PANDAS_UDF_PLACEMENT (0A000). The plan codec refuses the Python UDTF generator until its feature lands. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
No filter moves through MapPartitions, GroupMap or CoGroupMap (the pushdown's default arm keeps it above), and column pruning reads every column of their inputs: the UDF sees them all (group_ops.py L305 at 4.2.0; ADR 0022 §5). Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
The relation kinds get a second runtime seam beside PreparedUdf: RelationCallSpec, PreparedRelationUdf::open, a UdfTask fed by Frames (Batch with a side, EndGroup, End) and read through next. A task's send and next may be awaited at once, so a runtime can wait for its results to be taken. UdfRuntime::prepare_relation defaults to KRISHIV_UDF_NOT_AVAILABLE, so the existing runtimes keep compiling. RelationSink runs map, grouped, cogrouped and aggregate shapes in PythonEval's exchange shape: one task per driver opened at its first frame; rows held until a frame is full (maxRecordsPerBatch, and maxBytesPerBatch when set) or the group or side ends; groups cut where the sorted key changes; Blocked while a lane's window of frames is full; results stored in the shared plan's spillable list. Held rows and queued frames are reserved; a refused batch changes nothing; cancel and failure drop the tasks. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…P3-T014a, T085 §3.1) Physical nodes PythonMap, PythonGroupMap, PythonCoGroupMap and PythonAggregate (pure data: the UDF by name and version, frame columns, declared result) run on RelationSink. The planner gives a grouped kind a hash exchange on its keys (one partition without keys) and the spillable sort; a cogroup's two sides go through one union stream sorted by key then side, so a key's left rows reach the UDF before its right rows. An aggregate whose aggregates are Python UDAFs plans onto PythonAggregate; mixing them with built-ins is INVALID_PANDAS_UDF_PLACEMENT. Barrier mode, Python window aggregates and UDTFs are KRISHIV_UNSUPPORTED_FEATURE (PR-26c); a scalar PythonEval refuses a relation eval type. A worker without a runtime refuses with KRISHIV_UDF_NOT_AVAILABLE. Plan codec feature python_udf_relations (semantic): the four nodes, GENERATOR_KIND_PYTHON_UDTF and every eval type past the scalar ones. Proto messages, encode and decode, inventory cases, goldens, and a reader without the feature refuses each plan naming it. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
The structural cache key of each relation node, rewrites of their children and expressions, frames cut by maxBytesPerBatch, and a task result of another width failing the phase. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…(P3-T014a review W1) The Python aggregate feeds every row of a group to the UDF, so these clauses would be ignored silently. Refused at planning with KRISHIV_UNSUPPORTED_FEATURE until Spark's class is verified. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…s an empty frame (P3-T014a review M1, M2) Spark's layout (PandasGroupUtils.resolveArgOffsets): a grouped or cogrouped kind's frames lead with its key columns, computed keys included, so a runtime can build `key` for f(key, pdf). RelationCallSpec carries each side's column names; arity counts the data columns only. A cogroup key without rows on a side sends one zero-row Batch for that side (left before the right rows, right before EndGroup), so the Python side builds Spark's empty DataFrame from the frame's schema. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…eamed aggregate (P3-T014a review Md1-Md3, m3) - Each task result is reserved as the pump reads it; a refused one stops the task's reads until the driver reserves it (sink returns NeedMemory, finalize waits). finalize stores results as they arrive, writing a part to a spill file when the pool has no room. - The frame window is checked per frame: rows cut but not framed wait in a reserved backlog. A chunk's reservation is shared by the lanes' frames and released by the last taker; client-form copies are reserved. - A grouped aggregate keeps one key batch per input batch and joins the oldest pending groups with the values as every lane yields them, so the rows reach the spillable store; a task yielding another number of values than groups fails. - Result mismatches raise KRISHIV_PYTHON_UDF_RESULT_MISMATCH (39000), not INTERNAL_ERROR. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…-m6, m8-m10) - Test helper: Blocked took the batch, so the next batch follows (m1); a driver blocked on every batch sends each row once. - A slow fake task blocks the real driver and loses nothing (m2). - A weighted (change-stream) batch is refused (m4). - Only a Python aggregate UDF over a window gets the PR-26c refusal; a scalar one keeps the window planner's own error (m5). - INVALID_PANDAS_UDF_PLACEMENT is raised at physical planning only, as Spark does; analysis accepts the aggregate (m6). - Seam docs: one task per driver; UdfTask takes &self and must not call back into the operator (m8, m10); cache.rs doc comment back on plan (m9). Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…P3-T014a re-review N1) A right-only cogroup key whose rows spanned two input batches got a second zero-row left frame after its first right frames. The empty side now counts as seen when its frame is queued. Regression test for a left-only and a right-only key split over two batches. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…spill refused results, hand stored bytes over (P3-T014a re-review N2-N5) - N2: `wait` counts an aggregate driver ready only when every lane has values (what `pair::joined` needs), so `finalize` no longer loops inside one poll while one lane is ahead. - N3: in `finalize` a refused map or grouped result is stored with the others (spill file when the pool is full) instead of waiting for memory; an aggregate's refused value is reserved with `try_grow` after each step and otherwise waits for its memory while the other lanes keep moving. A finished aggregate with values left over is `check_complete`'s error rather than another loop. - N4: a stored result's reservation passes to the store's reservation through the consumer's grant credit (`lend`), never counted twice. - N5: tests for two aggregate lanes at different speeds (unit, bounded by a poll count; planner, values paired per key) and for a result refused in `finalize` under a small memory limit. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
A column key is also a data column, so `groupBy("a")` frames are `a, a, b`:
the `RelationCallSpec::input_names` doc says a runtime splits by key_count,
never by name.
Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…3-T014a re-review R1) wait() counted a lane with any queued result as joinable, but joined() needs rows; a lane holding only 0-row batches made finalize spin. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…s-ir Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com> # Conflicts: # crates/krishiv-common/krishiv-error-classes.json # crates/krishiv-exec/tests/planner/main.rs
|
Important
This repository does not receive automatic reviews because it has fewer than 10 stars.
 |
Codex Review Summary: ❌ FailedThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ec31271d09
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
Merging this PR will not alter performance1 warning
Comparing 
Footnotes
|
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1a573cab2c
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…CI docs) The module doc linked UdfRuntime::prepare_relation, which is not in scope there; rustdoc -D warnings failed the docs job. A reference link to crate::python_eval::UdfRuntime::prepare_relation resolves both uses. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…view) A Python aggregate's keys and call arguments went to the native compiler with their scalar Python calls in them, and so did a scalar call over a call's result: GROUP BY f(v), udaf(f(v)) and f(udaf(v)) failed at planning. As the ordinary aggregate path does, and as Spark does (ExtractGroupingPythonUDFFromAggregate, ExtractPythonUDFs and ExtractPythonUDFFromAggregate at v4.2.0), the calls in keys and arguments are extracted into PythonEval nodes below the node, which runs each call over their columns, and the calls over its results into nodes above it. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…'s collation (#746 review) GroupMap and CoGroupMap passed collated string keys as they are to the exchange, the sort and the group boundaries, which compare bytes: under UTF8_LCASE, 'A' and 'a' were two groups (two one-sided cogroup calls). Spark groups and cogroups by the collation (GroupedIterator and CoGroupedIterator orderings at v4.2.0). The planner now appends each collated key's comparison key, the one the collation rewrite gives aggregates, sorts and exchanges (krishiv_optimizer::collation_key, now public), and clusters, sorts and cuts groups on it; the frames keep each row's own key values. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…e memory (#746 review) In finalize a frame whose copies the pool refused was awaited first, while the lanes' results held reservations nothing could reclaim (the lane queues are not spillable, and the store's parts are not until it finalizes): the wait could only time out. A driver's results are now stored before it waits, written out while a frame waits for memory, and the frame tries again; the wait for what is still refused comes after every driver stored its results. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
… rows (#746 review) take_input reserves a whole batch, but cut_held split that reservation over the held rows only: the last cut of a batch's first group took the whole batch's bytes, so its later groups' slices, which keep the batch's buffers, held none once the first group's frames were taken. The driver now counts the rows the reservation covers (taken, not cut yet) and each cut takes its share of all of them. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…746 review) Spark's ArrowAggregatePythonExec (L156-157 at v4.2.0) returns an empty partition as it is, without calling the UDF, so a keyless grouped-agg Python UDF over empty input yields zero rows, unlike a built-in global aggregate. The test pins that the relation sink does the same. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b0ce59723f
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
@codex review |
|
Codex Review: Didn't find any major issues. Can't wait for the next one! Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
|
Note on commit b7332a4: its message ("reviews: T063 N1b re-review ...") is wrong; a shared message file was overwritten before the commit ran. The commit is the fix for the two memory threads in |
|
@codex review |
|
Codex Review: Didn't find any major issues. Hooray! Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
…s-ir Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com> # Conflicts: # crates/krishiv-common/krishiv-error-classes.json
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 91f01669f9
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…as Spark's writer With spark.sql.execution.arrow.maxBytesPerBatch set, the relation sink cut frames by the held input batch's average bytes per row. Uneven rows (two 30 KiB strings, then empty ones) and computed arguments that grow the payload made frames the average did not see. The cut now sizes the frame columns the tasks get (computed arguments, client forms) as Spark's writer does: rows are written while the frame's Arrow size is under the cap, so a frame ends at the row that brings it to the cap or past (BatchedPythonArrowInput.writeSizedBatch, PythonArrowInput.scala:226-233 at v4.2.0; the size is Arrow Java's getBufferSizeFor per field, ArrowWriter.scala:135-143 and 183-187). The size of rows start..start+k comes from validity, offsets and value spans (no copy), and grows with k, so the cut is a binary search. A map's rows travel as one struct column (MapInBatchEvaluatorFactory.scala:56-71), whose validity counts. Spark's default, Int.MaxValue, reads as no limit (PythonArrowInput.scala:227). Tests: the uneven-strings case (frames 2, 100), rows that each fill most of the cap (two per frame), and a row over the cap by itself (it ends the frame it joins; alone it is a frame of its own). Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 53e4e3fbc5
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
@codex review |
|
Codex Review: Didn't find any major issues. Keep it up! Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
…s-ir Conflicts with TABLESAMPLE (#761, `sample`), resolved keeping both sides: - plan.proto: main's `SampleNode sample = 27` keeps its number; the python_udf_relations nodes move to 28-31 (python_map, python_group_map, python_co_group_map, python_aggregate). Generated code regenerated with plan-protogen; the four python_udf_relations node goldens re-blessed (eval types and the UDTF generator are byte-identical). - krishiv-plan encode.rs: `Sample` keeps tag 36; MapPartitions, GroupMap, CoGroupMap and PythonTableFunction take 37-40 (ENCODING_VERSION unchanged: neither side had released them). - plan.rs, walk.rs, lib.rs, physical/mod.rs, features.rs, codec tests: both sides' arms and variants. - LogicalPlan's PartialEq compares the children pair by pair before the derived PlanNode comparison, which then finds them in the walk's memo: with both sides' new variants that derived frame (1,880 bytes unoptimised) nested once per level overflowed the 2 MiB stack at MAX_DEPTH (tests::depth::plans_at_the_cap_survive_every_traversal). Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
… of filling the pool In an aggregate with several UDF lanes a group is stored only once every lane has its value, and lane results are not spillable. A fast lane could reserve results until the pool was full before a slower lane's first value was reserved; nothing could be joined or released, and the slow lane's value waited for memory until arbitration failed (UNABLE_TO_ACQUIRE_MEMORY). A lane now reads no result while it holds more rows than a peer whose task still runs (`lane.rs` `ahead`): it waits on its own task, holding at most one result past the others, and is woken when a peer yields or ends. A lane of a map or grouped shape has no peers. Memory accounting is unchanged: every result is still reserved as it arrives. Test: two lanes, the slow one gated, a 1 MiB pool with spill on: the fast lane reads one value then waits, and finalize completes (before: the slow lane's 104-byte value was refused). Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
…es to take Round-5 review M1: the hold of 7da68d8 stopped a fast lane's reads while any running peer had fewer rows. A runtime that takes no frame while its results are unread then filled the fast lane's window, which stops the frames to every lane; a peer runtime that needs later frames before it yields (one batching values across groups) never got them, and exec has no call deadline: a hang. The hold now also needs the lagging peer to have frames it has not taken (`lane.rs` `held_back`: `in_flight() > 0`). The peer wakes the waiting pump when it takes its last frame, yields or stops (ended or failed). A peer starved of input never holds anyone back, so progress depends only on a lane taking frames it already has; past that the pool decides (a spill or a classed error), never a hang. The lane with the fewest rows is never held and a lane is held only by a peer with strictly fewer rows, so the holds form no cycle. With the production window the lead stays bounded by the window plus what the peer runtime buffers. Also from the review: - m1: only an aggregate's lanes are peers (`Lane::group(.., peered)`); a map or grouped shape's lanes are alone. - m2: a running row count per lane (`LaneIo::rows`, `pop_output`, `took`) instead of re-summing the queue on every result. - m3: "one result batch", not "one result". - m4: a peer's waker is taken under its lock and woken after release. Tests (M2) at the relation sink's production window (DEFAULT_WINDOW, 4): a peer batching values across 40 groups with a fast runtime that takes no frame while a result is unread (hung before, timeout 10 s); a peer ending with fewer values (the mismatch error, hung before); a peer failing while the fast lane is held (its error); the gated slow lane at the production window. The deep-window reproduction of the original stall is kept; once its slow lane opens it takes the whole window at once, so the hold ends and the lanes share the pool (slower, 12 s in a debug build, never stuck). Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
|
@codex review |
|
Codex Review: Didn't find any major issues. Chef's kiss. Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
|
@codex review |
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
|
@coderabbitai review |
Action not completed
|
…st-frame wake is tested (#746 review r6) - m2: when a task's `next` ends, its lane is marked stopped and its peers woken at once, before its send side finished, so a runtime that ends its output early while a send hangs cannot hold every peer. - m3: a new test, a quiet peer that batches values across groups and signals its reader only on output, hangs without the "took its last frame" wake (checked by deleting it: the test times out) and passes with it. No earlier test failed without that wake. - m1: the deep-window test's doc states what it checks (the hold while the gated peer has frames) and the bound once the gate opens; 3,000 groups instead of 12,000 (0.25 s instead of 21 s) and a 10 s timeout instead of 40 s. - `LaneIo::took` debug-asserts that the running row count does not drift. Signed-off-by: gopal tathe <44522021+techgopal@users.noreply.github.com>
|
Diff review passed: Opus review of dfb1320 (lane hold only while a lagging peer has frames it has not taken), verdict APPROVE_WITH_NITS, 0 blockers, 0 majors; no hang path, exact row counter, memory accounting unchanged. The minors are fixed in 1ce019a: a lane whose output ends is marked stopped and wakes its peers; a quiet-peer test covers the last-frame wake (it hangs without it); the deep-window test is 3,000 groups with a 10 s timeout; a debug_assert guards the row counter. |
|
main merged in for the CI disk fix (#776); lane: git pull before your next push |
P3-T014 PR-26a: the shared IR, runtime seam and operators for Python UDFs that see a relation rather than a column (design
P3-T085-embedded-python-udf-kinds.md§3.1–3.2, approved by the owner on 2026-11-12, Tier U;P3-T014-connect-udfs-sandbox.md§3.2 and §7).What it adds
krishiv-plan).MapPartitions,GroupMap,CoGroupMapandPythonTableFunction.GeneratorKind::PythonUdtf.Aggregate.PythonEvalTypegains Spark's codes 201–217, 250–254 and 300–302.INVALID_PANDAS_UDF_PLACEMENT. Physicalplanning also raises it, as Spark does.
krishiv-exec/src/python_relation.rs).UdfRuntime::prepare_relationis a default method that returnsKRISHIV_UDF_NOT_AVAILABLE, so the existingruntimes compile unchanged.
PreparedRelationUdf::openhands out aUdfTask, which is fedFrames:Batch{side},EndGroup,End.RelationCallSpeccarries the input column names of each side.Blockedwhile it is full);barrier=TrueraisesKRISHIV_UNSUPPORTED_FEATURE.DISTINCT,FILTERandWITHIN GROUPon a Python UDAF are refused.KRISHIV_PYTHON_UDF_RESULT_MISMATCH(39000).python_udf_relations, covers the nodes, the generator kind and the evaltypes. It has goldens, and a reader without the feature refuses the plan.
No runtime implements the relation seam yet, so no query can reach these nodes. The process-pool runtime comes in
PR-26c, the in-process runtime in PR-48, and Connect decode in PR-26d.
Review
An Opus diff review took three rounds, recorded in krishiv-internal
ops/reviews/2026-11-13-opus-B-t014a-udf-relations-ir-diff.md:CHANGES_REQUIRED, then CHANGES_REQUIRED, then APPROVE_WITH_NITS. All the findings that could give a wrong result,
hang, leak or crash are fixed: UDAF modifiers, keys reaching the task, empty cogroup sides, memory reservations,
cogroup side order, and a
finalizespin. One nit is fixed in the last commit: an aggregate lane holding only emptybatches no longer makes
finalizespin.Known follow-ups:
finalize, the lent bytes are freed before a spilled part is written. Fixing it means passing the creditinto
add_spilled_part.Checks
-D warnings, layering, file size andcargo check -p krishiv-pythonpass.size-approvedis needed; the Opus review has passed withevery blocking finding fixed.