Repository navigation
[GLUTEN-13480][FLINK] Support SHA2 / SHA224 / SHA256 / SHA384 / SHA512 offload to velox flinksql functions - #13484
TianJialong wants to merge 6 commits into
Conversation
|
@KevinyhZou @lgbo-ustc @philo-he @ggjh-159 Would you mind reviewing this PR? |
ggjh-159
left a comment
There was a problem hiding this comment.
Thanks for the clean five-way mapping and the thorough e2e coverage (including the job-fail assertions).
One corner divergence on a constant bad hashLength with empty/all-NULL input — details in the line comment on RexCallConverterFactory.java.
| // Direct name mapping: the SHA-2 family. The fixed-length spellings | ||
| // are the sha2 core with the length bound on the velox side; the | ||
| // hashLength operand is routed by the velox sha2 signature set. | ||
| Map.entry("SHA2", Arrays.asList(() -> new DefaultRexCallConverter("sha2"))), |
There was a problem hiding this comment.
Scenario: SHA2(s, 0) where 0 is a constant, and the input is an empty table (or s is all NULL). Flink fails this job even with zero rows: the constant length is checked in the generated operator's constructor (StringCallGen.scala:644 → CodeGeneratorContext.scala:1054-1056), which throws Unsupported algorithm. for 0 regardless of input. GFV succeeds and returns NULLs: the length check runs per row in velox, and an empty/all-NULL input never evaluates a row. Please validate a constant hashLength ∈ {224, 256, 384, 512} at conversion time in the SHA2 converter and fail the conversion otherwise; NULL constants and column references keep the existing runtime rejection, matching Flink. Add conversion tests (constant 0/385 fail to convert; valid and NULL constants still convert) and update testSha2LiteralZeroLengthPassesThrough, which currently pins constant-0 pass-through.
There was a problem hiding this comment.
@ggjh-159 Thanks for the review. Pushed 0f769d4, Done as suggested: the SHA2 converter now validates a constant hashLength at conversion time — anything outside {224, 256, 384, 512} fails the conversion with a clear message (Sha2RexCallConverter.isSuitable), so SHA2(s, 0) fails before any input is processed, empty table included — the same input-independent stage as Flink's check in the generated operator's constructor (CodeGeneratorContext.scala:1047-1056). NULL constants and column references keep the per-row runtime rejection in velox, as requested.
Tests: constant 0 / 385 / -1 (INT) and 0 (folded BIGINT) fail to convert; valid INT and BIGINT constants and a NULL constant still convert (RexNodeConverterTest.java:122-203) — the old constant-0 pass-through pin is replaced by these. End-to-end, the constant form now fails at submission while the column form still fails the job at runtime (HashFunctionsTest.java:104-149).
What changes are proposed in this pull request?
Fixes #13480
Maps the SHA-2 family hash expressions to their velox flinksql implementations (companion velox PR in bigo-sg/velox#95, velox4j PR in bigo-sg/velox4j#76):
RexCallConverterFactory:SHA2gainsSha2RexCallConverter, which validates a constant hashLength ∈ {224, 256, 384, 512} at conversion time and rejects the rest — Flink checks the constant length in the generated operator's constructor (input-independent: a bad constant length fails the job even on an empty table), while velox rejects per row, so the constant form must fail before any input is processed to keep both stacks consistent. NULL constants and column references cannot be checked at plan time and keep the per-row runtime rejection in velox.SHA224/SHA256/SHA384/SHA512stay plain name mappings (the length is bound on the velox side); thehashLengthoperand's integer width (INT or BIGINT, column ref or folded constant) resolves through the velox sha2 signature set, which registers every NUMERIC_INTEGER width.RexNodeConverterTest: covers the five mappings — SHA2's two-operand call (VARCHAR input + INTEGER length) and the four one-operand spellings, asserting each velox function name, arity and the VarCharType return (the operators' CHAR(56/64/96/128) / VARCHAR(128) return types map to velox VarCharType, matching the registered signatures).HashFunctionsTest(undergluten-flink/ut,table.runtime.stream.custom): the end-to-end suite for the family — bounded values tables feed the gluten streaming path and the print sink output is compared exactly: every spelling and every supported length over empty / plain / multi-byte UTF-8 / long inputs, planner-folded INT and BIGINT length constants, NULL propagation (NULL string and NULL length), the conversion-time rejection of constant 0 and 385 (testSha2ConstantInvalidLengthRejectedAtConversion), and the per-row runtime job-failure channel for a column length of 0 — which Flink 1.19.2 rejects outright, so the offloaded path must fail the same way — asserted through the job execution result.RexNodeConverterchange is required: the hash expressions read the input column directly, and any constant-input fold reduces to a VARCHAR literal, which the existing literal channel already handles — unlike the temporal family, no new typed-literal channels or cast elision are needed.How was this patch tested?
RexNodeConverterTest— the five name mappings with arity and return-type assertions (testSha2FamilyCallsMapToVeloxNames);testSha2LiteralInvalidLengthRejected(constant 0 / 385 / -1 in INT form and 0 in the folded-BIGINT form fail the conversion) andtestSha2LiteralValidAndNullLengthStillConvert(valid INT and BIGINT constants reach velox intact through their signatures; a NULL constant converts like any other NULL carrier and keeps the runtime rejection).HashFunctionsTest(10 cases on the mini-cluster through the gluten streaming path, bounded values source -> gluten-calc -> print sink) — digests pinned from Flink's own test suite for "" and "test" and independently probed for the multi-byte UTF-8 and long inputs, across every spelling and length; folded INT/BIGINT length constants; NULL propagation; a constant 0 / 385 is rejected at conversion time (testSha2ConstantInvalidLengthRejectedAtConversion, matching Flink 1.19.2's constructor-time check that fails the job even on empty input); a column length of 0 fails the job at runtime with the unsupported-length error (testSha2ZeroLengthFails, the per-row channel Flink uses for column and NULL lengths). The same scenario set ran on the GFV cluster (Flink 1.19.2, gluten-calc operator) with identical results, with both rejection channels exercised separately.