[GLUTEN-13411][FLINK] Support the arithmetic expressions offload to velox flinksql functions by wud-if · Pull Request #13412 · apache/gluten

wud-if · 2026-10-22T08:44:28Z

What changes are proposed in this pull request?

Maps the Flink arithmetic expressions to their velox flinksql implementations (companion velox PR bigo-sg/velox#85):

RexCallConverterFactory: ABS / ROUND entries; FLOOR / CEIL / CEILING map through a NumericOnlyRexCallConverter whose isSuitable rejects non-numeric operands at planning time — the datetime form FLOOR(ts TO DAY) (whose SYMBOL literal maps to VARCHAR) and unary minus over intervals stay clean planning-time rejections instead of failing later at velox runtime signature resolution or in toType. A one-operand "-" (Calcite emits Flink's unary minus as a single-operand "-" RexCall) resolves to a UnaryMinusRexCallConverter against the binary subtract candidates via isSuitable (arity plus numeric-operand guard).
DecimalPostCastRexCallConverter: ROUND over decimals narrows the result scale by the value-dependent second argument, which a velox signature cannot express — the plan emits a same-typed round call wrapped in a cast to Flink's inferred round type.
DecimalArithmeticOperatorRexCallConverters gains a two-operand guard so the single-operand "-" always resolves to negate.
getConverter counts a converter that throws while probing operand types (e.g. toType on an unmapped interval) as not-suitable, so an unconvertible form can never degrade a planning-time rejection into a hard failure.
CI routing: flink.yml / Flink.md temporarily clone the velox4j pin branch carrying the velox arithmetic branch; reverted once the velox PR merges.

How was this patch tested?

Factory-level tests: unary-minus arbitration vs decimal subtract, ROUND post-cast shape, FLOOR resolving to the numeric-only converter, and the planning-time rejections of FLOOR(ts TO DAY) and interval unary minus ("No suitable converter" instead of a runtime failure).
Execution-level tests (embedded gluten stack): round / abs / unary-minus / floor-ceil over double and decimal, matching native Flink through changelog-exact comparisons of the dual-run harness (11-case suite); the integer wraparound boundaries (INT_MIN, int64-min under abs) and the HALF_UP decimal-domain ties (2.675 -> 2.68, 647568.815 -> 647568.82) are pinned as embedded cases.
The dual-run SQL is embedded verbatim in ArithmeticDualRunSqlTest: each query is the exact SELECT (VALUES source included) that ran on Flink 1.19.2 natively and on the gluten-offloaded stack with identical inputs, results compared as exact changelog multisets; the expected rows in the test are the native output the dual-run matched against.

…l functions Map the Flink arithmetic expressions to their velox flinksql implementations (bigo-sg/velox, companion PR): - RexCallConverterFactory: ABS/FLOOR/CEIL/CEILING/ROUND entries plus a UnaryMinusRexCallConverter candidate on the one-operand "-" (Calcite emits unary minus as a single-operand "-" RexCall, resolved against the binary subtract candidates by isSuitable). - DecimalPostCastRexCallConverter: ROUND over decimals narrows the result scale by the value-dependent second argument, which a velox signature cannot express, so the converter emits a same-typed round call wrapped in a cast to Flink's inferred round type. - DecimalArithmeticOperatorRexCallConverters gains a two-operand guard so the single-operand "-" always resolves to negate. - Tests: factory-level arbitration (unary minus vs decimal subtract, ROUND post-cast shape, FLOOR default mapping) and execution-level tests for round/abs/unary-minus/floor-ceil over double and decimal. The flink.yml / Flink.md CI routing temporarily clones the velox4j pin branch carrying the velox arithmetic branch (companion velox PR); it reverts to the official repo once that PR merges.

wud-if · 2026-10-27T08:10:10Z

@KevinyhZou @lgbo-ustc @ggjh-159 Would you mind reviewing this PR?

ggjh-159 · 2026-10-27T09:30:43Z

+          // FLOOR/CEIL/ROUND arrive upper-cased from the operator table;
+          // unary minus arrives as a one-operand "-" (see above).
+          Map.entry("ABS", Arrays.asList(() -> new DefaultRexCallConverter("abs"))),
+          Map.entry("FLOOR", Arrays.asList(() -> new DefaultRexCallConverter("floor"))),


The datetime form of FLOOR/CEIL and the interval form of unary minus are accepted by the mapping but cannot produce an executable plan: DefaultRexCallConverter does not override isSuitable (unconditional success), so FLOOR(ts TO DAY) — a form Flink supports (FunctionGenerator.scala:245-261) — gets converted with the SYMBOL literal mapped via SymbolType→VarCharType (LogicalTypeConverter.java:120-122) into floor(timestamp, varchar), which fails at velox runtime signature resolution; unary minus over a year-month interval (arity-only guard) likewise fails at toType (no YearMonthIntervalType mapping). GFV expression mapping has no fallback (a miss fails the whole job), while pre-PR both forms were cleanly rejected at planning time as "Function not supported". Please add a numeric-operand guard to the floor/ceil and unary-minus converters (return ValidationResult.failure for non-numeric operands) so unconvertible forms stay planning-time rejections.

Thanks for the sharp catch! Fixed in 9540c8e: FLOOR/CEIL now use a NumericOnlyRexCallConverter and unary minus adds a numeric-operand guard on top of the arity check, so FLOOR(ts TO DAY) and interval unary minus go back to being clean planning-time rejections ("No suitable converter"), exactly as before the PR. Additionally, getConverter now counts a converter that throws while probing operand types (e.g. toType on an interval) as not-suitable instead of propagating the exception, so the rejection can never degrade into a hard failure. Covered by testFloorDatetimeFormRejectedAtPlanning / testUnaryMinusIntervalRejectedAtPlanning (6/6 green locally).

ggjh-159 · 2026-10-27T09:30:43Z

+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class RexCallConverterFactoryTest {


Please add the SQL details of the dual-run to the PR description — the queries run against Flink 1.19.2 and how the results were compared.

Thanks for the review! The dual-run queries are now embedded verbatim as tests:
ArithmeticDualRunSqlTest (52c62f5) carries the exact SELECT statements — VALUES sources included — that ran on Flink 1.19.2 in the dual-run (native vs gluten-offloaded, identical inputs, exact changelog-multiset comparison), with the native output rows they matched against; testAbsIntegerBoundaries / testRoundHalfUpTieInDecimalDomain additionally pin the INT_MIN / int64-min wraparounds and the 2.675 -> 2.68, 647568.815 -> 647568.82 HALF_UP ties.

The datetime form FLOOR(ts TO DAY) and unary minus over a year-month interval were accepted by the mapping but could not produce an executable plan: the SYMBOL literal maps to VARCHAR (floor(timestamp, varchar) fails at velox runtime) and the interval operand blows up in toType, which has no interval mapping — GFV has no planner fallback, so the job failed where it used to be a clean planning-time rejection. - FLOOR/CEIL now use NumericOnlyRexCallConverter, which rejects non-numeric operands in isSuitable (planning-time rejection). - UnaryMinusRexCallConverter adds a numeric-operand guard on top of the arity check, treating unmapped operand types as a rejection. - getConverter counts a converter that throws while probing operand types as not-suitable instead of propagating the exception, so the decision stays a clean 'No suitable converter' failure. - Tests: FLOOR(ts TO DAY) and interval unary minus now assert the planning-time rejection.

…tests - AbsFunctionsTest.testAbsIntegerBoundaries: INT_MIN and the int64 most-negative value wrap to themselves (matching Java Math.abs and Flink's inlined -x); 2^53+1 stays exact under the same-typed int64 abs (an intentional, documented deviation from Flink's double-widening fallback — see the velox-side registration note). - RoundFunctionsTest.testRoundHalfUpTieInDecimalDomain: 2.675 -> 2.68 and 647568.815 -> 647568.82, the ties that separate HALF_UP in the decimal domain from binary-domain rounding. These are the SQL forms of the dual-run cases; they run on the embedded TestValues-based tables, so the exact queries and expected results live with the code.

ArithmeticDualRunSqlTest carries the exact SELECT statements (VALUES sources included) that ran on Flink 1.19.2 in the dual-run verification, with the native output rows the dual-run matched against — so the queries and their verified results live with the code instead of only in the PR description. Also narrows testAbsIntegerBoundaries to the bigint case (the int64 most-negative wraparound the review actually asked for).

The review asks for the int64 most-negative wraparound (min and its negation under abs/negate), which is also where Flink's double-widening divergence lives; drop the int32 column.

The review's wraparound claim covers both INT_MIN and the int64 most-negative value; keep both columns.

ggjh-159

Thanks for the guard and the embedded dual-run tests — running the suite locally surfaced four issues: 1. Decimal operands rejected by the new numeric guard — details in the line comment on NumericOnlyRexCallConverter.java. 2. Typed-null literals crash conversion: RexNodeConverter.toVariant dereferences literal.getValue() (toString() in the numeric cases, bigDecimal.precision() in the DECIMAL case), so every CAST(NULL AS BIGINT/DOUBLE/DECIMAL(10,2)) in the dual-run tests throws an NPE and all four cases error out. Please handle null literals in toVariant before the value dispatch. 3. ROUND(647568.815, 2) over DOUBLE returns 647569 instead of 647568.82 — details in the line comment on RexCallConverterFactory.java. 4. One wrong test expectation — details in the line comment on RoundFunctionsTest.java. Items 1–3 are must-fix; item 4 is a test-expectation correction.

ggjh-159 · 2026-10-31T04:02:51Z

+  public ValidationResult isSuitable(RexCall callNode, RexConversionContext context) {
+    try {
+      boolean allNumeric =
+          callNode.getOperands().stream()


Decimal operands are rejected at planning time: TypeUtils.isNumericType returns false for DECIMAL (the numeric type list in TypeUtils has no DecimalType), so FLOOR/CEIL and unary minus fail the isSuitable guard and the PR's own decimal tests error out. The chain supports them — the decimal ROUND and ABS paths have no such guard and pass the same test run — so please make the numeric check accept DECIMAL.

ggjh-159 · 2026-10-31T04:02:51Z

+          Map.entry("FLOOR", Arrays.asList(() -> new NumericOnlyRexCallConverter("floor"))),
+          Map.entry("CEIL", Arrays.asList(() -> new NumericOnlyRexCallConverter("ceil"))),
+          Map.entry("CEILING", Arrays.asList(() -> new DefaultRexCallConverter("ceil"))),
+          Map.entry("ROUND", Arrays.asList(() -> new DecimalPostCastRexCallConverter("round"))),


ROUND(647568.815, 2) over DOUBLE returns 647569 in the end-to-end test (testRoundHalfUpTieInDecimalDomain), losing the rounded scale — Flink returns 647568.82, the pinned velox round implementation asserts 647568.82 for the same input in its own unit test, and the 2.675 row of the same test comes back correct (2.68). Please fix the offload path so the rounded scale survives.

ggjh-159 · 2026-10-31T04:02:51Z

+    List<Row> rows = Arrays.asList(Row.of(2.675d), Row.of(647568.815d), Row.of(-2.5d));
+    createSimpleBoundedValuesTable("tblRoundTie", "x double", rows);
+    String query = "select ROUND(x, 2) from tblRoundTie";
+    runAndCheck(query, Arrays.asList("+I[2.68]", "+I[647568.82]", "+I[-3.0]"));


The third expectation of testRoundHalfUpTieInDecimalDomain is wrong: Flink returns -2.5 for ROUND(-2.5, 2) (d keeps two decimal places, so no rounding happens for this value), and the run's actual -2.5 is correct — please change the expectation from -3.0 to -2.5.

TypeUtils.isNumericType excludes DecimalType, so the numeric-only guards rejected FLOOR/CEIL and unary minus over decimal operands at planning time and the decimal tests errored out. Add an operand-guard variant (isNumericOrDecimalType) used by the two guards; the plain list stays decimal-free because the basic arithmetic converters rely on it to decline decimals so the decimal converters take them.

toVariant dispatched on the literal value for every type, so typed-null literals (CAST(NULL AS BIGINT/DOUBLE/DECIMAL)) threw an NPE (toString() dereference / bigDecimal.precision()) and the dual-run cases carrying them errored out. Intercept null literals before the dispatch and fall back to VarCharValue(null): velox's ConstantTypedExpr renders a null variant as a null constant of the expression type. Note this relies on the velox4j serde that always emits the variant value member for typed-null carriers (merged upstream via apache#63).

…pectation Two issues in testRoundHalfUpTieInDecimalDomain: the expected -3.0 for ROUND(-2.5, 2) was wrong (no tie at scale 2; Flink returns -2.5), and the velox print sink formats DOUBLE columns with the C++ ostream default of 6 significant digits, which prints the correct 647568.82 as 647569. Assert through CAST(ROUND(x, 2) AS STRING): the cast renders the double through its shortest round-trip representation, printed verbatim, so the rounded scale survives.

The typed-null literal fix rides on the velox4j serde that always emits the variant value member for typed-null carriers (merged upstream via apache#63); the previous pin (a74f4a4) predates that merge, so CAST(NULL AS ...) cases would still fail at the runtime serde on the CI stack. The new pin (ci-stack-arithmetic) is the pin-velox-arithmetic head merged with upstream gluten-20260829 — same velox ref (the arithmetic branch head), plus the serde fix — and is the stack the review fixes were verified against (29/29 locally).

The print sink now formats doubles with Java Double.toString semantics (shortest round-trip, see the velox-side StringFormatter fix), so the CAST(... AS STRING) workaround is no longer needed and the test pins the actual offload output path end to end: ROUND(647568.815, 2) must print 647568.82, not the 6-significant- digit truncation 647569 the old `ss << t` rendering produced.

Resolve conflicts, keeping both sides' semantics: - RexNodeConverter: take upstream's generic typed-null path (toNullVariant covers all types incl. the arithmetic numeric cases). - RexCallConverterFactory: keep both the ABS/FLOOR/CEIL/CEILING/ROUND mappings (this PR) and the IFNULL mapping (upstream). - flink.yml / Flink.md: keep this PR's velox4j pin (90ad0cb, the CI stack head) — it already contains upstream's ifnull pin (2d08448). - RexNodeConverterTest: take upstream's superset (its type list covers the arithmetic numeric cases; decimal case kept as its own test). All 12 converter tests pass on the merge result.

The pin moves from the temporary ci-stack-arithmetic integration branch (whose velox ref predates the print-sink formatting fix) to pin-velox-arithmetic @ 8cde150, whose velox ref is the print-format head 1e3854bdf — without it, the raw-double ROUND tie assertion (647568.82) cannot pass in CI, since the 6-significant-digit print truncation turns it back into 647569. The pin switches to the upstream gluten-20260829 branch once velox4j PR apache#67 merges.

wud-if · 2026-10-31T11:56:34Z

@ggjh-159 Thanks for the thorough local run — all four items are fixed, including the print-layer root cause behind item 3:

Decimal operands rejected by the numeric guard: fixed in 78d8d3c — the operand guards now use a decimal-accepting check (TypeUtils.isNumericOrDecimalType) instead of the plain isNumericType. The plain list stays decimal-free deliberately: the basic arithmetic converters rely on decimals failing that check so the decimal converters (with their promotion casts) take over — widening the shared check would hijack decimal + - * / away from them. The PR's decimal tests (FLOOR/CEIL/ROUND/ABS over DECIMAL) are green again.
Typed-null literals crash conversion: fixed in 720f6c5 and, after merging upstream main into this branch (ead4460), unified with the generic toNullVariant path that landed there — every CAST(NULL AS T) now converts to the null-accepting carrier before the value dispatch (covered by RexNodeConverterTest, 4/4 incl. the DECIMAL case); all dual-run cases carrying CAST(NULL AS ...) are green.
ROUND(647568.815, 2) returning 647569: root-caused to the print channel, not the offload computation — the rounded 647568.82 is correct inside velox (as you noted, the round unit test asserts it), and the 647569 came from DefaultFormatter<double> rendering through ss << t, whose C++ ostream default keeps only 6 significant digits: 647568.82 has 8 and truncated, while the 2.675 row (4 digits) survived — exactly the asymmetry you observed. Fixed at the root in the velox print sink (bigo-sg/velox 1e3854bdf): doubles and floats now render with Java Double.toString / Float.toString semantics (std::to_chars shortest round-trip, plain notation within [1e-3, 1e7), Java-style scientific notation outside — pinned by testDoubleFormatterToString with the 647568.82 regression case). The test here now asserts the raw double output (select ROUND(x, 2), no CAST workaround) so it pins the actual offload output path end to end, and the CI pin was switched to the velox4j branch carrying that velox head (pin-velox-arithmetic @ 8cde150 → velox 1e3854bdf, ca0bf2f) so CI exercises the fixed stack.
Wrong third expectation: fixed in c5a1012 — ROUND(-2.5, 2) keeps -2.5 (no tie at scale 2), matching Flink.

ggjh-159 · 2026-11-02T11:41:01Z

Please update velox4j ref and fix the CI problems(flink-test failed). @wud-if

The fork's pin-velox-arithmetic branch does not contain the merged upstream head 13afddf (the apache#67 merge), so cloning wud-if and resetting to the pin fails with 'Could not parse object'. Clone bigo-sg's gluten-20260829, where the head lives — same form as upstream main's own workflow.

The distro patchelf (0.12) fails with 'maximum file size exceeded' on the velox4j shared libraries built from the current velox head; 0.17.2 handles them — remove+set-rpath land the RUNPATH and ldd loads cleanly.

wud-if · 2026-11-03T01:54:55Z

Please update velox4j ref and fix the CI problems(flink-test failed). @wud-if

@ggjh-159 The CI problems are fixed; the remaining 5 errors are unrelated to this PR — details below.

flink-test now builds and runs the suite: the original failure was patchelf: maximum file size exceeded on the ~260MB libvelox.so — the distro patchelf (el8, 0.12) cannot rewrite it. The workflow now installs the static patchelf 0.17.2 into /usr/local before the build (769551d). Version choice was empirical: 0.18.0 clears the big .so but corrupts the bundled third-party libraries (ldd then fails with "ELF load command address/offset not properly aligned" on liblz4.so.1); 0.17.2 handles both — verified against a real 251MB libvelox.so and the bundled .so set (remove+set-rpath land the RUNPATH, ldd loads cleanly).
velox4j ref: the CI clone now points at the merged upstream head bigo-sg/velox4j@13afddf2 (the [GJ-58] Support Spark DS V1 (#59) #67 squash — its velox ref is the merged arithmetic revision), as pushed in 58fb240.
The 5 remaining errors are not from this PR: GlutenStreamTwoInputWatermarkStatusTest (4 errors: "StreamJoin is not yet migrated to operator-owned keyed state; there is no state handler to initialize") and NexmarkTest (1 error: same for "GroupWindowAggregator") — both NOT_IMPLEMENTED from velox's experimental/stateful, introduced by the stateful key-hash-collision change (Enable the second stage #79) that is part of the current upstream velox head this PR's stack builds on. Any arithmetic-containing velox revision contains it (the arithmetic squash sits on top of Enable the second stage #79), so the pin cannot avoid it. All arithmetic-expression tests are green in this run: RoundFunctionsTest 4/4, AbsFunctionsTest 3/3, FloorCeilFunctionsTest 4/4, UnaryMinusFunctionsTest 2/2, ArithmeticDualRunSqlTest 4/4, RexCallConverterFactoryTest 8/8 (132 tests total: 0 failures, 5 errors — all the stateful ones above, 5 skipped).

github-actions Bot added INFRA FLINK labels Oct 22, 2026

ggjh-159 reviewed Oct 27, 2026

View reviewed changes

wud-if added 5 commits October 29, 2026 15:57

Narrow the abs boundary test to the bigint case

f2dc776

The review asks for the int64 most-negative wraparound (min and its negation under abs/negate), which is also where Flink's double-widening divergence lives; drop the int32 column.

Cover INT_MIN alongside the int64 edges in the abs boundary test

a791288

The review's wraparound claim covers both INT_MIN and the int64 most-negative value; keep both columns.

ggjh-159 reviewed Oct 31, 2026

View reviewed changes

wud-if added 6 commits October 31, 2026 13:30

wud-if mentioned this pull request Oct 31, 2026

[FLINK] Add Flink arithmetic functions to the flinksql namespace bigo-sg/velox#85

Merged

Point the CI pin at the merged velox4j head

87470b8

wud-if force-pushed the expression-dev-arithmetic branch from 016e892 to 87470b8 Compare November 2, 2026 13:31

wud-if and others added 3 commits November 2, 2026 21:32

Merge branch 'main' into expression-dev-arithmetic

2317c7a

Pin patchelf 0.17.2 in the flink CI instead of the distro build

769551d

The distro patchelf (0.12) fails with 'maximum file size exceeded' on the velox4j shared libraries built from the current velox head; 0.17.2 handles them — remove+set-rpath land the RUNPATH and ldd loads cleanly.

wud-if and others added 3 commits November 5, 2026 20:05

Merge upstream main into expression-dev-arithmetic

a7c13d6

Merge branch 'main' into expression-dev-arithmetic

74b005a

Merge upstream main into expression-dev-arithmetic

1a33a21

Drop the patchelf explanation comment from the CI workflow

c6d98b7

github-actions Bot removed the INFRA label Nov 6, 2026

Navigation Menu

Repository navigation

[GLUTEN-13411][FLINK] Support the arithmetic expressions offload to velox flinksql functions - #13412

[GLUTEN-13411][FLINK] Support the arithmetic expressions offload to velox flinksql functions#13412
wud-if wants to merge 21 commits into
apache:mainfrom
wud-if:expression-dev-arithmetic

wud-if commented Oct 22, 2026 •

edited

Loading

Uh oh!

wud-if commented Oct 27, 2026

Uh oh!

ggjh-159 Oct 27, 2026

Uh oh!

wud-if Oct 29, 2026

Uh oh!

ggjh-159 Oct 27, 2026

Uh oh!

wud-if Oct 29, 2026

Uh oh!

ggjh-159 left a comment

Uh oh!

ggjh-159 Oct 31, 2026

Uh oh!

ggjh-159 Oct 31, 2026

Uh oh!

ggjh-159 Oct 31, 2026

Uh oh!

wud-if commented Oct 31, 2026

Uh oh!

ggjh-159 commented Nov 2, 2026

Uh oh!

wud-if commented Nov 3, 2026 •

edited

Loading

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants


		import static org.assertj.core.api.Assertions.assertThat;

		public class RexCallConverterFactoryTest {

Conversation

wud-if commented Oct 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes are proposed in this pull request?

How was this patch tested?

Uh oh!

wud-if commented Oct 27, 2026

Uh oh!

ggjh-159 Oct 27, 2026

Choose a reason for hiding this comment

Uh oh!

wud-if Oct 29, 2026

Choose a reason for hiding this comment

Uh oh!

ggjh-159 Oct 27, 2026

Choose a reason for hiding this comment

Uh oh!

wud-if Oct 29, 2026

Choose a reason for hiding this comment

Uh oh!

ggjh-159 left a comment

Choose a reason for hiding this comment

Uh oh!

ggjh-159 Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

ggjh-159 Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

ggjh-159 Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

wud-if commented Oct 31, 2026

Uh oh!

ggjh-159 commented Nov 2, 2026

Uh oh!

wud-if commented Nov 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

wud-if commented Oct 22, 2026 •

edited

Loading

wud-if commented Nov 3, 2026 •

edited

Loading