Migrate to ptabler - #123
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request migrates the tables aggregation logic in tables-aggregation.lib.tengo to use the ptabler library, replacing the external software execution with native dataframe operations. The feedback suggests adding defensive checks to prevent runtime panics when accessing nested properties of axis and col specifications, and adding an assertion to ensure unsupported aggregation types are not silently ignored when fallback logic is executed.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| ensureUniquenessParamsFromPconvParams := func(pfConvParams) { | ||
| return { | ||
| axes: slices.map(pfConvParams.axes, func(axis) { | ||
| return axis.column | ||
| return { column: axis.column, type: axis.spec.type } | ||
| }), | ||
| columns: slices.map(pfConvParams.columns, func(col) { | ||
| return col.column | ||
| return { column: col.column, type: col.spec.valueType } | ||
| }) | ||
| } | ||
| } |
There was a problem hiding this comment.
To prevent potential runtime panics, we should defensively check if axis.spec and col.spec are defined before accessing their properties (type and valueType). In Tengo, accessing properties on an undefined value will cause a runtime panic.
ensureUniquenessParamsFromPconvParams := func(pfConvParams) {
return {
axes: slices.map(pfConvParams.axes, func(axis) {
return { column: axis.column, type: is_undefined(axis.spec) ? undefined : axis.spec.type }
}),
columns: slices.map(pfConvParams.columns, func(col) {
return { column: col.column, type: is_undefined(col.spec) ? undefined : col.spec.valueType }
})
}
}
| } else { | ||
| // first: dedup rows by key, keeping the first. | ||
| result = df.unique({ subset: keyNames, keep: "first", maintainOrder: true }) | ||
| } |
There was a problem hiding this comment.
The previous implementation supported arbitrary aggregation types passed via aggParams[0]. The new implementation hardcodes keep: "first" when len(aggParams) <= 1, silently ignoring any other aggregation type if specified. To prevent unexpected behavior or silent bugs, we should assert that aggParams[0] is indeed "first" when provided.
} else {
if len(aggParams) > 0 {
ll.assert(aggParams[0] == "first", "Unsupported aggregation type: " + aggParams[0])
}
// first: dedup rows by key, keeping the first.
result = df.unique({ subset: keyNames, keep: "first", maintainOrder: true })
}
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| if len(aggParams) > 1 { | ||
| // max_by: keep, per key, the row with the greatest ranking column value. | ||
| rankCol := aggParams[1] | ||
| aggs := slices.map(params.columns, func(col) { | ||
| return pt.col(col.column).maxBy(pt.col(rankCol)).alias(col.column) | ||
| }) | ||
| result = df.groupBy(keyNames...).agg(aggs...) | ||
| } else { | ||
| // first: dedup rows by key, keeping the first. | ||
| result = df.unique({ subset: keyNames, keep: "first", maintainOrder: true }) | ||
| } |
There was a problem hiding this comment.
aggParams[0] type string silently ignored
The old API used aggParams[0] as the aggregation-type discriminator ("max_by" or "first"). The new implementation discards it entirely and dispatches only on len(aggParams) > 1. A call like ensureUniqueness(tsv, params, "max_by") (type provided but ranking column omitted) now silently falls into the df.unique() branch instead of raising an error, and a hypothetical future caller passing a third aggregation type with two arguments would silently be treated as max_by. Since all current call sites happen to map 1:1 (1 arg → first, 2 args → max_by), there is no current regression, but the contract is no longer enforced.
Prompt To Fix With AI
This is a comment left during a code review.
Path: workflow/src/tables-aggregation.lib.tengo
Line: 46-56
Comment:
**`aggParams[0]` type string silently ignored**
The old API used `aggParams[0]` as the aggregation-type discriminator ("max_by" or "first"). The new implementation discards it entirely and dispatches only on `len(aggParams) > 1`. A call like `ensureUniqueness(tsv, params, "max_by")` (type provided but ranking column omitted) now silently falls into the `df.unique()` branch instead of raising an error, and a hypothetical future caller passing a third aggregation type with two arguments would silently be treated as `max_by`. Since all current call sites happen to map 1:1 (1 arg → first, 2 args → max_by), there is no current regression, but the contract is no longer enforced.
How can I resolve this? If you propose a fix, please make it concise.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
23dda11 to
28054d9
Compare
2 participants
Greptile Summary
This PR replaces the external
paggregateSwbinary (ptransform) with thept(ptabler) library for the TSV deduplication step inensureUniqueness, and extendsensureUniquenessParamsFromPconvParamsto carry column type metadata so that integer columns are correctly re-coerced after reading the TSV withinferSchema: false.Touched Terms:
ensureUniqueness— Deduplicates a multi-sample TSV by key axes. Now implemented as an in-process ptabler DataFrame workflow instead of shelling out to a separate ptransform binary; integer columns (Long/Int) are explicitly re-cast viacast(\"Double\") → round() → cast(type)before dedup.ensureUniquenessParamsFromPconvParams— Converts pfconv export parameters into the shape expected byensureUniqueness. Previously extracted onlycolumnnames; now also extractstype(axis.spec.type) andvalueType(col.spec.valueType) soensureUniquenesscan apply type-correct casts.aggParams— Variadic arguments controlling deduplication strategy:aggParams[0]was the aggregation-type string ("max_by" / "first"),aggParams[1]was the ranking column for max_by. In the new code,aggParams[0]is ignored; the branch is selected purely bylen(aggParams) > 1. Current call sites are unaffected but the type string is no longer validated.intCasts— New internal variable. A list of ptabler column expressions that re-cast integer columns from String (the result ofinferSchema: falseTSV loading) back to their declared Long/Int types before aggregation/dedup.pt(ptabler) — New import replacingexec,assets, and thepaggregateSwsoftware handle. Provides a DataFrame API (workflow,frame,withColumns,groupBy,agg,unique,col,save,run) used to implement in-process table operations.The old
max_bycode path had a variable-shadowing bug (pickCols := []re-declared inside the block, then immediately iterated) that caused it to pass an emptypickColsto the aggregation binary; the new implementation fixes this.The
aggParams[0]aggregation-type string is now silently discarded; dispatch is by argument count only. All current callers still work correctly but the contract is no longer enforced.Confidence Score: 4/5
Safe to merge — all current call sites behave correctly; only the dropped aggParams[0] type-string leaves a latent gap in input validation.
The functional change is well-scoped: all three concrete call sites pass either ("first") or ("max_by", rankCol), mapping cleanly onto the new arity-based dispatch. The old max_by path had a variable-shadowing bug that caused an empty column list to be sent to the binary; the new code fixes that. The one gap is that aggParams[0] is now silently ignored, so a mistaken call such as passing only "max_by" without a ranking column silently falls through to dedup-by-first instead of failing fast.
workflow/src/tables-aggregation.lib.tengo — verify the aggParams arity contract is documented or enforced, and confirm ptabler's maxBy semantics on Long-typed columns match the previous ptransform max_by behaviour.
Important Files Changed
Flowchart
%%{init: {'theme': 'neutral'}}%% flowchart TD A[ensureUniqueness called\ninputTsv, params, ...aggParams] --> B[Build allCols\naxes + value columns] B --> C[Collect keyNames\nfrom params.axes] C --> D[pt.workflow\nwf.frame inputTsv as TSV\ninferSchema: false] D --> E{Any Long/Int\ncolumns in allCols?} E -- Yes --> F[withColumns: cast String to Double to round to Long/Int\nfor each integer column] E -- No --> G{len aggParams > 1?} F --> G G -- Yes: max_by path --> H[rankCol = aggParams 1\nmap params.columns to maxBy rankCol\ndf.groupBy keyNames .agg aggs...] G -- No: first path --> I[df.unique subset: keyNames\nkeep: first, maintainOrder: true] H --> J[result.save output.tsv] I --> J J --> K[wf.run .getFile output.tsv\nreturn deduplicated TSV]%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%% flowchart TD A[ensureUniqueness called\ninputTsv, params, ...aggParams] --> B[Build allCols\naxes + value columns] B --> C[Collect keyNames\nfrom params.axes] C --> D[pt.workflow\nwf.frame inputTsv as TSV\ninferSchema: false] D --> E{Any Long/Int\ncolumns in allCols?} E -- Yes --> F[withColumns: cast String to Double to round to Long/Int\nfor each integer column] E -- No --> G{len aggParams > 1?} F --> G G -- Yes: max_by path --> H[rankCol = aggParams 1\nmap params.columns to maxBy rankCol\ndf.groupBy keyNames .agg aggs...] G -- No: first path --> I[df.unique subset: keyNames\nkeep: first, maintainOrder: true] H --> J[result.save output.tsv] I --> J J --> K[wf.run .getFile output.tsv\nreturn deduplicated TSV]Prompt To Fix All With AI
Reviews (1): Last reviewed commit: "Migrate to ptabler" | Re-trigger Greptile
Context used: