diff --git a/docs/design/sirius-mo-exact-lowering.md b/docs/design/sirius-mo-exact-lowering.md new file mode 100644 index 0000000000000..063d06e072577 --- /dev/null +++ b/docs/design/sirius-mo-exact-lowering.md @@ -0,0 +1,424 @@ +# Embedded MO exact-decimal lowering + +Design version: 1. Owner: MatrixOne planning, compile and CN execution. +Tracking: [#28968](https://github.com/matrixorigin/matrixone/issues/28968), +parent [#28966](https://github.com/matrixorigin/matrixone/issues/28966). + +The user approved this C/D design before implementation on 2026-10-08. +This document records PR C of the delivery sequence in merged #29690. +Semantic authority remains #29449 document blob +`42a89f09a1d168d02b9583cb3ea7b4de6dbb5634`. +Native B, Sirius #27, merged as +`5ea60cd31955d0dced2adcbcd3df0772207b79ef`, is the native dependency. + +Current delivery gate: C pins merged +[Sirius #29](https://github.com/matrixorigin/sirius/pull/29) at +`908ffc75a58b3be496b2172795e416326416e7ec`. Kernel prerequisite +[#29775](https://github.com/matrixorigin/matrixone/pull/29775) is merged. +C is ready for review; passing required CI and review approval gate merging. +Local integration covers all 22 native preparations and the complete public +numeric fixture. D implementation follows merged C. + +## Decision and invariant + +For each admitted embedded SELECT, validation, serialization, MO read +publication, native preparation and result reconstruction must agree on the +physical decimal width, precision, scale and nullability selected by MO. +Values, NULLs and public numeric error classes must match native MO. +The negation is any descriptor drift, coefficient narrowing, extra rounding, +error reclassification or fallback after explicit embedded selection. + +The native runtime owns one immutable capability snapshot. Expose it through +`Capabilities() uint64` on the bridge runtime and configured backend. Native +ABI-v1 MO-input/result capabilities retain their existing mandatory check. +Mask `16u` selects the complete MO exact-decimal v1 family; unknown bits do not +enable another family. Flight returns no embedded numeric capability. + +`ExportEmbeddedMO` accepts an explicit value export profile derived from that +snapshot. Candidate owns a copy, and every validation, read-schema and build +pass uses it. There is no mutable global profile or per-query capability +cache. The zero profile retains legacy emission. A capability change in another +runtime cannot change a previously validated candidate's serialization. + +## Wire and semantic closure + +The exact profile uses `urn:matrixone:sirius:exact-decimal:v1` and its +`mo_exact_decimal` type consistently for Decimal64/128/256, including narrow +declared precisions in physically wide carriers. It never infers width from +precision. Parameters are physical bits, precision and scale, plus Substrait +nullability. Internal precision 76 is valid for a 256-bit working type; public +results are limited to precision 65. + +Exact literals use the approved `ExactDecimalLiteral` protobuf with a single +fixed-width little-endian coefficient, exactly 8/16/32 bytes. Preserve MO's +canonical literal/protocol provenance. Decode a planner-owned numeric literal +with MO's exact parser where its internal representation requires that step; +this does not admit arbitrary runtime string-to-decimal conversion. + +Admit the native family's fourteen scalar and four aggregate operations by +bound overload ID and complete operand/result descriptors, rather than display +names. Normal checked casts retain the approved source/domain restrictions; +explicit, assignment/comparison and unsupported source conversions remain +ineligible. Preserve CASE/COALESCE types, exact comparisons, aggregate phases, +and existing structural group/join/sort consumers. Unsupported reachable +shapes fail before readers start. No new projection pushdown or MO fragment +execution substitutes for an unsupported native operation. + +Retain Native B's signature limits, including SUM/AVG results whose physical +width is at least their input width. A physically wide, low-precision input +whose MO-bound SUM returns Decimal128 is declined; changing that result to +Decimal256 would violate descriptor identity. Scan-owned filters remain MO +reader work in both validation and emission, rather than being validated as +Flight expressions and then omitted from the embedded wire plan. + +Embedded ordinary integer add/subtract/multiply remain ineligible: the pinned +consumer uses unchecked cuDF arithmetic for those operations. An inherited +registry label does not prove MO overflow equivalence. Checked exact-decimal +arithmetic, integral predicates and aggregates retain their existing domains. + +Exact CASE retains MO's complete bound result descriptor through the existing +checked annotation, including a nullable condition with required value arms. +Typed DATE extraction guards MO's `ZeroDate` sentinel before year/month/day, +returning zero for that sentinel and preserving normal/NULL dates. The pinned +component function's physical INT16 result is normalized through the proven +INT32 component domain to the declared BIGINT before conditional selection. +No published reader bytes, native ABI, Flight encoding or SQL text changes. + +All decimal-bearing parts of one exact closure use the extension. Standard +Flight emission stays on its existing type/function contract. Declaration +anchors and serialization are deterministic. Original and encoded plans retain +the existing 16 MiB limit. Current-main Q1-Q22 inventory includes the complete +reachable expression graph, not only the first reported decline. + +## Input, output and errors + +Decimal256 input is a 32-byte fixed-width coefficient through the existing +MO-native publisher. Credit is acquired before outgoing allocation/copy; +constants are charged for expansion and oversized batches split at row +boundaries. Keep 64 MiB input/result windows and the existing one-active-query, +sixteen-waiter, default-two-stream contract. + +Reuse the existing typed borrowed-result decoder: its fixed-width path already +supports 32-byte types. Validate native schema, lengths, validity and descriptor +identity before installing vectors. Native result credit survives the fill +callback, and the existing buffer lease owns the Go copy. Partial failure, +cancellation and result-writer errors unwind through the existing owners. + +Native status 12 maps to `moerr.ErrOutOfRange` (MySQL 1690 / SQLSTATE 22003). +Status 13 maps to `moerr.ErrInvalidInput` (MySQL 20301 / SQLSTATE HY000). +The operation that failed owns the error: scalar overflow inside SUM remains +out-of-range; SUM final overflow and checked-cast failure remain invalid-input. +Producer cancellation during cleanup cannot replace that primary error. +GPU-unavailable/quiescence failure keeps the fatal-runtime contract. +Messages remain bounded and do not add row contents. + +Extend the existing query-local terminal evidence with the selected profile, +capability snapshot and verified output descriptors/digests. Keep account/query +identity, GPU task counts, source/result accounting and terminal health together. +There is no new registry, unbounded collector or high-cardinality metric label. + +## Ownership and validation map + +| Closure | First owner and terminal path | Risk / proof | +| --- | --- | --- | +| Capability/profile | Service runtime -> candidate value -> serialization | R2: missing capability, legacy controls, candidate/profile retention | +| Types/functions/literals | MO bound plan -> exact encoder -> native prepared schema | R2: independent descriptor/wire vectors, overload negatives, complete inventory | +| Input/results | MO reader -> acquired native credit -> GPU -> result lease -> fill -> release | R3: width/NULL/constant/slice boundaries, partial failure, cancellation and reuse | +| Errors/evidence | Native terminal status -> typed Go error -> public MySQL result/terminal event | R3: failing-operation identity, cleanup races, query-local metadata | +| SDK/build | Merged native pin -> clean fingerprinted SDK -> linked MO consumer | R1/R2: source/header/proto/artifact identity, supported build modes and loadability | + +Host unit tests prove arithmetic encoding, admission and control contracts; +they do not replace GPU execution evidence. Required real GPU/public tests cover +all widths, high limbs/signs, NULLs, exact scalar/aggregate values, joins/sorting, +masked errors and healthy reuse. Reuse the approved prepared-division controls +at increments 0/4/10/30 and verify bound result metadata after re-execution. + +Keep public SQL fixtures small and deterministic, with native MO as the +independent value/metadata/error oracle. Test malformed input and absent +capabilities before readers start. Build CPU-only, Sirius-enabled and combined +Sirius/cuVS profiles using the current go.mod toolchain and controlled CGo +wrapper. Run repository SCA and self-review before delivery. + +The selected design preserves one typed execution boundary. Alternatives of a +global exporter toggle, mixed ordinary/exact decimal closure, float/narrowing +conversion or post-selection fallback violate the invariant. A second result +decoder/allocator would duplicate existing ownership without another need. + +## Delivery boundary + +C pins merged Native B and remains opt-in; current backend/default selection +does not change. D starts after C merges and versions the real native-MO and +embedded-MO campaign, with bounded transient row spools, all-22 SF1/SF10, +streams=2 plus 1/4 controls, one excluded warm-up/five measured suites and ten +separate Q9 repetitions. Existing performance/resource gates remain required. + +E/F are excluded from this implementation round. Their separate design must +resolve production recovery authority before default cutover and Flight +retirement. A nil/unready lease manager is not proof that Flight is empty. +No direct-TAE, storage/directory-lock, Docker image/base or fallback work is +introduced here. #28968 remains open until D acceptance; #28966 remains open +through cutover, verified release availability and retirement. + +## Historical implementation evidence (2026-10-09) + +The implementation base is `83d82b8ee0cd694e0c6a7146902d74ae4dfb415a`. +The selected native SDK is generated from clean merged Sirius +`5ea60cd31955d0dced2adcbcd3df0772207b79ef`, with importer +`95d9ce8d78490db3991ab6145653716aa3ec42c9`. Tests use the current +Go 1.27.1 toolchain and frozen Pixi `mo` environment. + +The complete Q1-Q22 MO exporter inventory passes validation and deterministic +serialization. Focused descriptor/overload/literal, Decimal256 publication, +borrowed-result, error-identity and schema-evidence tests pass. The exporter, +bridge and CN owning-package suites pass. The full compile suite fails +`TestRequiredIVFWorkersFallbackAsWholeQuery/{supported,canceled}`; both failures +reproduce at the verified clean base with the same GPU-linked host test mode. +They are not reported as a passing full compile suite. +All four owning-package suites also pass in normal CPU mode, including the full +compile suite. The reproduced IVF failure is specific to the GPU-tagged host +test mode; changing or skipping that unrelated test is outside C. + +The combined Sirius/cuVS release build passes SDK verification, linking and +packaging. Real MySQL tests pass fixed-width Decimal64/128/256 values and NULLs, +scalar overflow (including inside SUM), SUM final overflow, healthy runtime +reuse, and binary-prepared division at increments 0/4/10/30 with independent +value and public metadata assertions. Terminal events confirm exact profile, +capability mask 31, completed GPU tasks, no fallback and zero retained input +and result credit after cleanup. Public fixture setup remains one isolated +cluster; each selected run takes about 9 seconds including startup/teardown. +CASE/COALESCE inactive-overflow masking and division-by-zero NULL controls +also pass. CPU-only release compilation and binary loading pass with the frozen +compiler; the host compiler's failure in unchanged jemalloc is an environment +failure, not a code result. The default full repository static-check gate and +the affected closure with `gpu,sirius,sirius_integration` tags pass. +The Sirius-enabled release profile without MO's separate cuVS flag also builds, +packages, initializes the actual GPU runtime and passes the public all-width +round-trip control. Sirius execution remains on the GPU in this profile. + +The review traces each changed hunk through the five closure rows above. Q1 +retains the existing input-lease publication/release owner and result-batch +cleanup inside fill; query Close joins readers before destruction. Q2 retains +the existing independent query cancellation/deadline and credit-release paths. +Q3 retains 64 MiB windows, bounded descriptors and logical constant expansion; +schema evidence adds at most the already validated 1024 descriptors once per +terminal query. The capability/profile values are immutable before publication; +this change adds no shared mutable state, worker, wait or accumulating registry. +The independent public error/reuse cases close the new error-class path without +changing general frontend error policy. + +**Original merge blocker, resolved by importer #5 / Sirius #28:** +`TestExactEmbeddedTPCHNativePreparation/q1` crashed in the +pinned importer's `SubstraitToDuckDB::TransformRootOp`. Its root-name iterator +uses `SkipColumnNames` on DuckDB carrier types. Decimal256's private aliased +four-field STRUCT carrier is an opaque Substrait user-defined scalar, so its +limbs must not consume additional SQL root names. Advancing by four skips later +headings and eventually reads outside `RelRoot.names`. The native crash stack +and current importer source identify that exact path. Single-result numeric +tests passing do not close this multi-result failure. + +C must remain a draft until a prerequisite importer fix is merged, Sirius pins +that merged importer, and C pins the merged Sirius dependency. Then rerun all +22 native preparations and the full public numeric fixture. No MO-side carrier +name padding, coefficient narrowing, SQL rewrite, fallback or skipped assertion +is an acceptable substitute. D remains dependent on merged and validated C. + +The user approved the prerequisite fix PRs on 2026-10-09. Importer +[duckdb-substrait #5](https://github.com/matrixorigin/duckdb-substrait/pull/5) +adds a query-scoped opaque-carrier policy and bounds both root-name consumers; +it is ready for review with distribution CI limits recorded. The Sirius callback and width-preserving following-column +regression are prepared in draft +[Sirius #28](https://github.com/matrixorigin/sirius/pull/28). Its importer pin +remains at the prior merged revision until #5 merges. Joint development binding +validation passes 1009 assertions in 84 cases, including 157 assertions in three +canonical importer cases. Pinned changed-file Sirius hooks pass. These results +are not full C native/public acceptance or proof that the draft builds against +its old importer pin. Merge and pin dependencies in order before C delivery. +CI found an overly strict trailing cursor check; corrected importer head +`6693c3a4d779fb0e15fa07e4ce4166f0572f3952` retains legacy final-STRUCT +headings and checks every actual name read. Its independent value/name regression +passes. Corrected distribution CI has seven failures: the clean base's known +STRUCT field-selection failure/crash and five later failures now independently +reproduced at clean importer `95d9ce8d78490db3991ab6145653716aa3ec42c9` +and its bundled DuckDB `d8cdaa33fda8df955cc76ef58a280f68f4cd43fa`. +The five explicitly selected cases reproduce DISTINCT_FROM, CTE (TPCH/TPCDS), +the old user-defined-literal expected-message mismatch and an empty-plan root +failure. Full distribution CI is not green; C-unit CI remains pending. No SQL +assertion is relaxed, skipped or changed to obtain these results. + +## Merged importer follow-up (2026-10-09) + +Importer #5 merged as `99c7ca3b6f8f3159239e119ed2982d42f98c4690`, with +the same source tree as the tested final fix. Sirius #28 now pins that merged +revision at head `433cadd43a10442b0825a23234fb6d859207e9d5` and is ready +for review. Its clean frozen-Pixi build passes 852 assertions / 81 native +binding cases and 1468 / 8 production numeric C ABI GPU cases. The independent +C consumer passes real GPU work, credit accounting and runtime cleanup/reuse. +All eight SDK exporter tests and pinned changed-file hooks pass. The regenerated +ABI-v1 SDK records the exact clean head, with all 73 artifact fingerprints, +C header and canonical literal schema verified. Current-head CI is reported +separately; these local results do not imply its compiler jobs passed. + +The recorded MO pin remains merged Native B until Sirius #28 merges. Then +advance C to that merged Sirius revision and rerun all-22 native preparation +and the full public numeric fixture. No unmerged native pin or full acceptance +is claimed, and D still follows merged, validated C. + +## Consumer corrections after Sirius #28 (2026-10-09) + +Sirius #28 merged as `af4dc60152b3e14f263c7fe863b29ac7e154de30`; +C now pins that clean merged revision and merged importer #5. The original +root-name crash is closed. Full native preparation exposed two remaining MO +wire omissions: absent constant fetch bounds and declared nullability on +decimal column references. Emit both constant fetch modes for embedded plans, +using count -1 and offset 0 for absent bounds. Preserve each MO-bound decimal +column descriptor with the existing checked exact cast function: Substrait +field selections alone cannot carry the projected descriptor, and a grouped +aggregate's required internal value can be exposed by a nullable MO projection. +This retains physical width, precision and scale and leaves the result schema +comparison intact. Legacy Flight emission retains its existing wire shape. + +Condition-free joins in exact plans also need a native preparation fix: +DuckDB leaves JOIN ON true as ANY_JOIN when its ordinary optimizer is skipped. +Native preparation can represent that condition with equal constant keys in +its existing GPU join, preserving join kind, multiplicity, projection maps and +outer NULLs for empty inputs. MO retains its original JoinRel emission. Public +controls include scalar aggregates over empty/all-NULL input. + +For embedded DATE extraction, lower the already admitted year/month/day fields +to their direct native functions. DuckDB's generic date_part function is not a +supported Sirius GPU expression. Quarter must decline before readers start +until the native executor supports it. Do not broaden the DATE-only semantic +admission to time durations, session-zone timestamps or tolerant text parsing. + +The full public numeric fixture also exposed cuDF AST IS_NULL on Decimal256's +private STRUCT carrier. Correct that in the native expression owner using its +top-level validity mask and the caller's stream/resource, then publish a +separate prerequisite Sirius fix. Keep the public NULL-predicate assertion and +merged-pin delivery gate; do not substitute a SQL rewrite or fallback. + +## Joint consumer validation and remaining delivery gate (2026-10-09) + +[Sirius #29](https://github.com/matrixorigin/sirius/pull/29), clean head +`2a39339beb0c2312dc4b71fbdd472e822450637c`, fixes both remaining native +consumers: exact NULL predicates use the canonical validity mask, and literal +TRUE joins use constant equality keys in the existing GPU join. C retains its +original JoinRel emission. Native tests cover INNER/LEFT/RIGHT/FULL joins, +duplicate/NULL high-limb payloads and empty sides; FALSE/NULL predicates remain +rejected before work starts. Native decimal inequality keys remain unsupported. + +The joint MO consumer at `6ef0aadea257eb5c7fa3988674b9c81902261a4d` +passes all 22 native preparations with no reader/GPU work started and the full +public MySQL numeric fixture. The fixture checks exact values and public +metadata across all widths, arithmetic/aggregates/joins/sorting, NULL predicates, +DATE extraction, empty/all-NULL scalar joins, operation-owned errors, healthy +reuse, masked errors and prepared division increments 0/4/10/30/4. Its 31 +terminal events show capability 31, exact profile, real completed GPU tasks, +no fallback, healthy cleanup and zero retained input/result credit. + +That joint check used an explicitly marked development SDK from merged #28 +plus the source now committed in #29. C's delivered gitlink remains merged +`af4dc60152b3e14f263c7fe863b29ac7e154de30`; no unmerged native pin is +delivered. At #29's clean head, production C ABI checks pass 1964 assertions / +10 cases, exact/ordinary GPU controls pass 14879 / 11, and binding checks pass +852 / 81. Eight SDK exporter tests, an independent C99 consumer and all 73 +clean SDK fingerprints pass. These local results do not imply compiler CI is +green or establish D's SF1/SF10, lifecycle/resource or performance acceptance. + +The host's installed driver libraries differed from its loaded kernel driver. +Local GPU validation used the matching official NVIDIA 615.71.09 CUDA/NVML +libraries through a temporary test launcher. Their archive SHA-256 is +`8cd4b1fb55057db60342dd8e8f16a01f84486b46ab88185bfc19046a1e996772`. +Host packages were unchanged; these driver libraries are outside the repository +and release artifacts. Normal SDK/source/artifact checks remained enabled. + +At this checkpoint, the remaining merge order was #29, C's merged pin plus +release/owning-package/static delivery checks, then C. + +## Merged native delivery and kernel prerequisite (2026-10-09) + +Sirius #29 merged as `908ffc75a58b3be496b2172795e416326416e7ec` +on the authoritative `upstream-dev-merge` branch. C now pins that merged +revision. Its source tree is identical to the independently validated clean +`2a39339beb0c2312dc4b71fbdd472e822450637c` tree; regenerate the release +SDK and verify its merged source/artifact identity before using it for delivery. +The importer remains merged #5 at +`99c7ca3b6f8f3159239e119ed2982d42f98c4690`. + +C's pessimistic Compose BVT at head `8284b77a` exposed premature prepared +result-schema freezing when DDL commits after execute-time binding. The failure +also reproduces on clean main with an inert phase point. Independent kernel +PR [#29775](https://github.com/matrixorigin/matrixone/pull/29775) publishes +metadata from the actual executed generation before its first positive row, +or after successful zero-row completion; incompatible retries after publication +remain rejected. It also finalizes saved metadata after whole-query success. +Its eight public cases, complete frontend/compiler normal and race suites, +focused race stress and full pre-push SCA pass. Exact-head CI now passes, +including pessimistic Compose BVT, overall coverage and CI Required. + +Inherit the kernel fix only after #29775 merges, then complete C's final +merged-pin release/public/owning validation and required CI before readiness. +D implementation begins after C merges and stays one complete public campaign +PR. E/F remain behind their separate recovery-readiness design. Keep #28968 +and #28966 open. + +Merged-native delivery checks at executable C head `59aaa5d234`, on +authoritative MO main `0eb746f37c22f57d2081331b25bf379c80168c81`, pass: + +- Clean release SDK regeneration from merged #29, all 73 SDK fingerprints, + and source/compiler/header/proto/runtime-package identity checks. +- Default MO, Sirius-only and combined Sirius/cuVS release builds. An + independent C99 SDK consumer passes actual GPU execution, input/result credit, + cleanup and engine reuse. Same-process cuVS-before/Sirius/cuVS-after and the + complete combined native bridge suite pass. +- All 22 native preparations and the complete public numeric fixture pass + against the merged release SDK. Its 31 terminal events have capability 31, + the exact profile, completed GPU tasks, no fallback, healthy cleanup and zero + retained input/result credit. +- All four owning packages pass in default normal/race modes and with + Sirius-only native bindings. Native bridge data/cancellation passes 83 race + repetitions in one process (measured 0.36 seconds; 30-second budget). +- Eight SDK exporter tests and sixteen MO SDK verifier tests pass. + +These checks establish merged-native delivery evidence. They do not replace +the remaining merged-kernel integration/CI gate, D's full-data campaign, +actual-memory/process baselines, or any performance acceptance gate. + +## Merged kernel integration (2026-10-10) + +Kernel #29775 merged as `af3f7232a31879a2d6ed590785c8e9a560361c3f`. +C executable head `ad61b5b5d3c2794928dac9afc548d30e0ce769a8` inherits +that fix and the MORPC shutdown-test correction merged in #29794. The latter +removes global ants-pool reboot from the inline rejection test; the exact test +passes 100 race repetitions. C's earlier histogram-export assertion failure is +also fixed by waiting for its two asynchronous full-batch sends. + +Final local integration on this merged base passes: + +- Default, Sirius-only and combined Sirius/cuVS release builds, with the clean + merged #29 SDK, all 73 fingerprints and normal package/provenance guards. +- All four C-owning packages plus frontend in default normal/race modes; + Sirius native-tag owning checks and the full combined bridge suite, including + cuVS-before/Sirius/cuVS-after in one process. +- All 22 native preparations without reader or GPU-task admission, followed by + the complete public MySQL numeric fixture. Its 31 terminal events have + capability 31, the exact profile, completed GPU tasks, no fallback, healthy + cleanup and zero retained input/result credit. +- All nine prepared-metadata/saved-result public cases, including an actual + persisted-batch failure followed by successful save/replay on the same + physical connection. Normal mode and three race repetitions pass. +- Native data/cancellation passes 68 race repetitions in one process + (measured 0.44 seconds; 30-second budget). Full pre-push SCA passes. + +The host now has matching loaded and installed NVIDIA 615.78.08 drivers; +these GPU checks use the installed driver directly. The local native fixture +uses 256 MiB GPU capacity and 2 GiB host capacity. Its initial 512 MiB host +configuration correctly rejected larger preparation graphs at the existing +progress-window reservation boundary; only that local fixture was corrected. +Per-input/result windows and all production limits/assertions are unchanged. + +The eight SDK-exporter tests, sixteen MO SDK-verifier tests and independent C99 +consumer evidence above retain their unchanged native/script/ABI/artifact +scope. This integration does not establish D's SF1/SF10, actual-memory/process +or performance gates. The kernel merge prerequisite and local integration are +complete; readiness still requires passing required CI on the delivered C +head. D begins after C merges, and #28966/#28968 remain open. diff --git a/pkg/cnservice/sirius_embedded.go b/pkg/cnservice/sirius_embedded.go index 59f11e8135f97..6f5d0d9b6fe83 100644 --- a/pkg/cnservice/sirius_embedded.go +++ b/pkg/cnservice/sirius_embedded.go @@ -58,6 +58,7 @@ func validateSiriusEmbeddedConfig(c *SiriusConfig) error { } type embeddedRuntime interface { + Capabilities() uint64 Accepting() bool Close(context.Context) error Prepare(context.Context, siriusbridge.Request) (*siriusbridge.Query, error) @@ -70,6 +71,8 @@ type embeddedBackend struct { func (b *embeddedBackend) Accepting() bool { return b.native.Accepting() } +func (b *embeddedBackend) Capabilities() uint64 { return b.native.Capabilities() } + func (b *embeddedBackend) Close(ctx context.Context) error { return b.native.Close(ctx) } func (*embeddedBackend) CanFallbackBeforeVisibility(error) bool { return false } func (*embeddedBackend) Reconcile(uint64, []byte, func(context.Context) error) error { @@ -228,9 +231,12 @@ func (e *embeddedExecution) Cleanup(ctx context.Context) error { if stats.Fatal { health = "unavailable" } + schema, schemaDigest := embeddedSchemaEvidence(e.request) logutil.Info("Sirius embedded execution", zap.String("query_id", hex.EncodeToString(e.request.QueryID)), zap.Uint64("account_id", e.request.AccountID), zap.String("backend", "embedded"), zap.String("scan_mode", "mo"), zap.Bool("fallback", false), + zap.Uint64("native_capabilities", e.request.NativeCapabilities), zap.String("numeric_profile", e.request.NumericProfile), + zap.Any("verified_output_schema", schema), zap.String("output_schema_sha256", schemaDigest), zap.Uint32("gpu_streams", e.streams), zap.Int("read_bindings", len(e.request.Reads)), zap.Uint64("input_rows", e.counters.inputRows.Load()), zap.Uint64("input_payload_bytes", e.counters.inputBytes.Load()), zap.Float64("first_row_seconds", firstRow.Seconds()), zap.Float64("wall_seconds", wall.Seconds()), diff --git a/pkg/cnservice/sirius_embedded_test.go b/pkg/cnservice/sirius_embedded_test.go index f1948b46468e8..a89742673dc9b 100644 --- a/pkg/cnservice/sirius_embedded_test.go +++ b/pkg/cnservice/sirius_embedded_test.go @@ -40,6 +40,8 @@ type embeddedRuntimeTestRecorder struct { func (*embeddedRuntimeTestRecorder) Accepting() bool { return true } +func (*embeddedRuntimeTestRecorder) Capabilities() uint64 { return 0 } + func (r *embeddedRuntimeTestRecorder) Close(context.Context) error { r.closed = true return nil diff --git a/pkg/cnservice/sirius_numeric_test.go b/pkg/cnservice/sirius_numeric_test.go new file mode 100644 index 0000000000000..5f3a69bf7b97c --- /dev/null +++ b/pkg/cnservice/sirius_numeric_test.go @@ -0,0 +1,82 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cnservice + +import ( + "encoding/binary" + "testing" + + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/container/vector" + planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/compile" + "github.com/matrixorigin/matrixone/pkg/sql/compile/siriusbridge" + "github.com/matrixorigin/matrixone/pkg/testutil" + "github.com/stretchr/testify/require" +) + +func TestEmbeddedDecimal256ResultPreservesLimbsNullsAndBorrowing(t *testing.T) { + proc := testutil.NewProcess(t) + values := []types.Decimal256{{B0_63: 9007199254740993, B192_255: 1}, {B0_63: ^uint64(124), B64_127: ^uint64(0), B128_191: ^uint64(0), B192_255: ^uint64(0)}, {}} + backing := make([]byte, len(values)*32+8) + for row, value := range values { + for limb, word := range []uint64{value.B0_63, value.B64_127, value.B128_191, value.B192_255} { + binary.LittleEndian.PutUint64(backing[row*32+limb*8:], word) + } + } + backing[len(values)*32] = 4 + result := siriusbridge.Result{Rows: 3, Backing: backing, Vectors: []siriusbridge.Vector{{Data: backing[:96], Nulls: backing[96:]}}} + request := compile.SiriusPrepareRequest{Headings: []string{"v"}, OutputTypes: []planpb.Type{{Id: int32(types.T_decimal256), Width: 65, Scale: 4}}} + bat, err := decodeEmbeddedSiriusResult(result, request, proc.Mp()) + require.NoError(t, err) + t.Cleanup(func() { bat.Clean(proc.Mp()) }) + require.Equal(t, vector.BorrowedLease, bat.Vecs[0].DataBackingKind()) + require.Same(t, &backing[0], &bat.Vecs[0].GetData()[0]) + require.Equal(t, types.New(types.T_decimal256, 65, 4), *bat.Vecs[0].GetType()) + for row := 0; row < 2; row++ { + require.Equal(t, values[row], vector.GetFixedAtNoTypeCheck[types.Decimal256](bat.Vecs[0], row)) + } + require.True(t, bat.Vecs[0].IsNull(2)) + for _, bad := range []siriusbridge.Vector{ + {Data: backing[:95]}, + {Data: backing[:96], Class: 1}, + {Data: backing[:96], Nulls: []byte{8, 0, 0, 0, 0, 0, 0, 0}}, + } { + invalid, err := decodeEmbeddedSiriusResult(siriusbridge.Result{Rows: 3, Vectors: []siriusbridge.Vector{bad}}, request, proc.Mp()) + require.Error(t, err) + require.Nil(t, invalid) + } +} + +func TestEmbeddedSchemaEvidenceRetainsDescriptorIdentity(t *testing.T) { + request := compile.SiriusPrepareRequest{Headings: []string{"private expression heading"}, OutputTypes: []planpb.Type{{Id: int32(types.T_decimal256), Width: 15, Scale: 2}}} + columns, digest := embeddedSchemaEvidence(request) + require.Equal(t, []embeddedColumnEvidence{{OID: int32(types.T_decimal256), Precision: 15, Scale: 2, Nullable: true}}, columns) + require.Len(t, digest, 64) + _, same := embeddedSchemaEvidence(request) + require.Equal(t, digest, same) + for _, mutate := range []func(){ + func() { request.OutputTypes[0].Id = int32(types.T_decimal64) }, + func() { request.OutputTypes[0].Width++ }, + func() { request.OutputTypes[0].Scale++ }, + func() { request.OutputTypes[0].NotNullable = true }, + func() { request.Headings[0] = "another heading" }, + } { + mutate() + _, changed := embeddedSchemaEvidence(request) + require.NotEqual(t, digest, changed) + digest = changed + } +} diff --git a/pkg/cnservice/sirius_schema.go b/pkg/cnservice/sirius_schema.go new file mode 100644 index 0000000000000..5a241092c09db --- /dev/null +++ b/pkg/cnservice/sirius_schema.go @@ -0,0 +1,54 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cnservice + +import ( + "crypto/sha256" + "encoding/binary" + "encoding/hex" + + "github.com/matrixorigin/matrixone/pkg/sql/compile" +) + +type embeddedColumnEvidence struct { + OID int32 `json:"oid"` + Precision int32 `json:"precision"` + Scale int32 `json:"scale"` + Nullable bool `json:"nullable"` +} + +// Preparation verified these descriptors against the native output schema. +// The bridge bounds them to 1024 columns before an execution can exist. Names +// participate only in the digest: log neither expression headings nor values. +func embeddedSchemaEvidence(request compile.SiriusPrepareRequest) ([]embeddedColumnEvidence, string) { + columns := make([]embeddedColumnEvidence, len(request.OutputTypes)) + digest := sha256.New() + _, _ = digest.Write([]byte("sirius-mo-output-schema-v1\x00")) + var descriptor [17]byte + for i, typ := range request.OutputTypes { + columns[i] = embeddedColumnEvidence{OID: typ.Id, Precision: typ.Width, Scale: typ.Scale, Nullable: !typ.NotNullable} + binary.LittleEndian.PutUint32(descriptor[:4], uint32(typ.Id)) + binary.LittleEndian.PutUint32(descriptor[4:8], uint32(typ.Width)) + binary.LittleEndian.PutUint32(descriptor[8:12], uint32(typ.Scale)) + descriptor[12] = 0 + if !typ.NotNullable { + descriptor[12] = 1 + } + binary.LittleEndian.PutUint32(descriptor[13:], uint32(len(request.Headings[i]))) + _, _ = digest.Write(descriptor[:]) + _, _ = digest.Write([]byte(request.Headings[i])) + } + return columns, hex.EncodeToString(digest.Sum(nil)) +} diff --git a/pkg/sql/compile/sirius_backend.go b/pkg/sql/compile/sirius_backend.go index 5e4ccee41b108..6418d554097bb 100644 --- a/pkg/sql/compile/sirius_backend.go +++ b/pkg/sql/compile/sirius_backend.go @@ -30,15 +30,17 @@ import ( // may call Release only after its consumer has quiesced. The deadline includes // neither a new transaction nor permission to replay an externally visible query. type SiriusPrepareRequest struct { - AccountID uint64 - QueryID []byte - Plan []byte - OutputTypes []planpb.Type - Headings []string - Deadline time.Time - Release func(context.Context) error - Snapshot [12]byte - Reads []SiriusReadDescriptor + AccountID uint64 + QueryID []byte + Plan []byte + OutputTypes []planpb.Type + Headings []string + Deadline time.Time + Release func(context.Context) error + Snapshot [12]byte + Reads []SiriusReadDescriptor + NativeCapabilities uint64 + NumericProfile string } // SiriusInput reserves native credit before a producer materializes its outgoing @@ -92,12 +94,15 @@ type SiriusExecution interface { // Reconcile retains cleanup ownership by statement identity until the previous // consumer is quiescent. Close stops admission and drains that ownership. type SiriusBackend interface { + Capabilities() uint64 Prepare(context.Context, SiriusPrepareRequest) (SiriusExecution, error) Reconcile(uint64, []byte, func(context.Context) error) error Close(context.Context) error CanFallbackBeforeVisibility(error) bool } +func (*siriusFlightBackend) Capabilities() uint64 { return 0 } + type siriusFlightBackend struct { *sidecarflight.Runtime } diff --git a/pkg/sql/compile/sirius_backend_test.go b/pkg/sql/compile/sirius_backend_test.go index be564f7354b0c..56719ba337cc3 100644 --- a/pkg/sql/compile/sirius_backend_test.go +++ b/pkg/sql/compile/sirius_backend_test.go @@ -50,6 +50,7 @@ func (s *siriusExecutionStub) CleanupAfterRun(ctx context.Context, err error) er func TestSiriusFlightBackendPreservesPreparationOwnership(t *testing.T) { require.Nil(t, NewSiriusFlightBackend(nil)) backend := NewSiriusFlightBackend(&sidecarflight.Runtime{}) + require.Zero(t, backend.Capabilities(), "Flight cannot advertise embedded exact-decimal support") // Invalid preparation fails before any connection is required. Its release // callback still runs exactly once. Malformed requests remain terminal even // after successful cleanup, just as in the concrete Flight API. diff --git a/pkg/sql/compile/sirius_batch.go b/pkg/sql/compile/sirius_batch.go index e10c1554e775d..c4f5197b6e673 100644 --- a/pkg/sql/compile/sirius_batch.go +++ b/pkg/sql/compile/sirius_batch.go @@ -91,6 +91,8 @@ func siriusElementSize(oid types.T) (int, error) { return 8, nil case types.T_decimal128: return 16, nil + case types.T_decimal256: + return 32, nil case types.T_char, types.T_varchar, types.T_binary, types.T_varbinary: return types.VarlenaSize, nil default: diff --git a/pkg/sql/compile/sirius_batch_test.go b/pkg/sql/compile/sirius_batch_test.go index b66cc022f1e44..70f5c17504422 100644 --- a/pkg/sql/compile/sirius_batch_test.go +++ b/pkg/sql/compile/sirius_batch_test.go @@ -138,6 +138,41 @@ func TestSiriusSlicesBoundLogicalExpansion(t *testing.T) { require.Equal(t, 6, end) } +func TestSiriusDecimal256PublicationPreservesFixedWidthAndConstants(t *testing.T) { + proc := testutil.NewProcess(t) + typ := types.New(types.T_decimal256, 65, 4) + value := types.Decimal256{B0_63: 9007199254740993, B192_255: 1} + bat := batch.NewWithSize(2) + t.Cleanup(func() { bat.Clean(proc.Mp()) }) + bat.Vecs[0] = vector.NewVec(typ) + for row := 0; row < 3; row++ { + require.NoError(t, vector.AppendFixed(bat.Vecs[0], value, row == 2, proc.Mp())) + } + constant, err := vector.NewConstFixed(typ, value, 3, proc.Mp()) + require.NoError(t, err) + bat.Vecs[1] = constant + bat.SetRowCount(3) + columns := []SiriusReadColumn{{Type: planpb.Type{Id: int32(types.T_decimal256), Width: 65, Scale: 4}}, {Type: planpb.Type{Id: int32(types.T_decimal256), Width: 65, Scale: 4, NotNullable: true}}} + recorder := &siriusBatchRecorder{} + require.NoError(t, publishSiriusBatch(t.Context(), recorder, bat, columns)) + require.Len(t, recorder.vectors[0].Data, 96) + require.Len(t, recorder.vectors[1].Data, 32) + require.Equal(t, uint32(1), recorder.vectors[1].Class) + require.Equal(t, byte(4), recorder.vectors[0].Nulls[0]) + for _, payload := range [][]byte{recorder.vectors[0].Data[:32], recorder.vectors[1].Data} { + require.Equal(t, value.B0_63, binary.LittleEndian.Uint64(payload)) + require.Equal(t, value.B192_255, binary.LittleEndian.Uint64(payload[24:])) + } + require.Equal(t, 1, recorder.acquired) + require.Equal(t, 1, recorder.released) + end, err := siriusSliceEnd(bat, columns, 0, 16+2*66) + require.NoError(t, err) + require.Equal(t, 2, end, "wide constants count at their expanded logical width") + columns[0].Type.Scale++ + require.ErrorContains(t, publishSiriusBatch(t.Context(), recorder, bat, columns), "decimal type") + require.Equal(t, 1, recorder.acquired, "schema rejection precedes credit acquisition") +} + func TestSiriusSplitAndConstantVarlenaPublication(t *testing.T) { proc := testutil.NewProcess(t) columns := []SiriusReadColumn{{Type: planpb.Type{Id: int32(types.T_varchar)}}} @@ -176,6 +211,7 @@ func TestSiriusNativeElementSizesAndBatchRejections(t *testing.T) { {4, []types.T{types.T_int32, types.T_uint32, types.T_float32, types.T_date}}, {8, []types.T{types.T_int64, types.T_uint64, types.T_float64, types.T_decimal64, types.T_timestamp}}, {16, []types.T{types.T_decimal128}}, + {32, []types.T{types.T_decimal256}}, {24, []types.T{types.T_char, types.T_varchar, types.T_binary, types.T_varbinary}}, } { for _, oid := range group.oids { @@ -184,8 +220,8 @@ func TestSiriusNativeElementSizesAndBatchRejections(t *testing.T) { require.Equal(t, group.width, width) } } - _, err := siriusElementSize(types.T_decimal256) - require.Error(t, err, "wide types stay declined until numeric support is delivered") + _, err := siriusElementSize(types.T_uuid) + require.Error(t, err, "unsupported native types stay declined") proc := testutil.NewProcess(t) bat := batch.NewWithSize(1) t.Cleanup(func() { bat.Clean(proc.Mp()) }) diff --git a/pkg/sql/compile/sirius_embedded_admission_test.go b/pkg/sql/compile/sirius_embedded_admission_test.go index 9cc9749c26bf1..bc6de22b9d8ce 100644 --- a/pkg/sql/compile/sirius_embedded_admission_test.go +++ b/pkg/sql/compile/sirius_embedded_admission_test.go @@ -26,6 +26,7 @@ import ( mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test" planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/pb/txn" + "github.com/matrixorigin/matrixone/pkg/sql/plan/substrait" "github.com/matrixorigin/matrixone/pkg/testutil" "github.com/stretchr/testify/require" spb "github.com/substrait-io/substrait-protobuf/go/substraitpb" @@ -34,9 +35,16 @@ import ( type embeddedAdmissionRecorder struct { SiriusBackend - request SiriusPrepareRequest - err error - execution SiriusExecution + request SiriusPrepareRequest + err error + execution SiriusExecution + capabilities uint64 + capReads int +} + +func (b *embeddedAdmissionRecorder) Capabilities() uint64 { + b.capReads++ + return b.capabilities } func (b *embeddedAdmissionRecorder) Prepare(_ context.Context, request SiriusPrepareRequest) (SiriusExecution, error) { @@ -80,6 +88,27 @@ func TestEmbeddedSiriusAdmissionBindsWithoutStartingReaders(t *testing.T) { require.NoError(t, proto.Unmarshal(backend.request.Plan, &wire)) require.Equal(t, "__sirius_embedded_v1", wire.Relations[0].GetRoot().Input.GetRead().GetNamedTable().Names[0]) c.siriusRead = nil + for _, capabilities := range []uint64{11, 31} { + backend.capabilities, backend.capReads = capabilities, 0 + backend.request = SiriusPrepareRequest{} + query.Nodes[0].TableDef.Cols[0].Typ = planpb.Type{Id: int32(types.T_decimal256), Width: 65, Scale: 8} + offloaded, err = c.compileEmbeddedSiriusRead(ctx, plan, runtime) + require.Equal(t, 1, backend.capReads, "one immutable capability snapshot per admission") + if capabilities == 11 { + require.True(t, substrait.IsNotEligible(err)) + require.False(t, offloaded) + require.Empty(t, backend.request.Plan, "unsupported type must fail before native preparation or reader I/O") + } else { + require.NoError(t, err) + require.True(t, offloaded) + require.Equal(t, capabilities, backend.request.NativeCapabilities) + require.Equal(t, "mo-exact-decimal-v1", backend.request.NumericProfile) + require.NoError(t, proto.Unmarshal(backend.request.Plan, &wire)) + require.Equal(t, int64(256), wire.Relations[0].GetRoot().Input.GetRead().BaseSchema.Struct.Types[0].GetUserDefined().TypeParameters[0].GetInteger()) + c.siriusRead = nil + } + require.Zero(t, eng.buildBlockReadersCalls) + } backend.err = errors.New("native prepare rejected") offloaded, err = c.compileEmbeddedSiriusRead(ctx, plan, runtime) require.ErrorIs(t, err, backend.err) diff --git a/pkg/sql/compile/sirius_embedded_read.go b/pkg/sql/compile/sirius_embedded_read.go index cd92bc226fdc4..75cc0ee3fe141 100644 --- a/pkg/sql/compile/sirius_embedded_read.go +++ b/pkg/sql/compile/sirius_embedded_read.go @@ -35,7 +35,8 @@ func (c *Compile) compileEmbeddedSiriusRead(ctx context.Context, queryPlan *plan if queryPlan == nil || queryPlan.GetQuery() == nil || !siriusPlanEligible(queryPlan) { return false, moerr.NewNotSupported(ctx, "statement is not eligible for embedded Sirius") } - candidate, err := substrait.ExportEmbeddedMO(queryPlan.GetQuery()) + capabilities := runtime.Backend.Capabilities() + candidate, err := substrait.ExportEmbeddedMO(queryPlan.GetQuery(), substrait.NewEmbeddedExportProfile(capabilities)) if err != nil { return false, err // explicit selection never falls back } @@ -102,6 +103,7 @@ func (c *Compile) compileEmbeddedSiriusRead(ctx context.Context, queryPlan *plan AccountID: uint64(account), QueryID: append([]byte(nil), id[:]...), Snapshot: snapshotID, Plan: wire, OutputTypes: candidate.OutputTypes(), Headings: append([]string(nil), queryPlan.GetQuery().Headings...), Reads: descriptors, Deadline: deadline, + NativeCapabilities: capabilities, NumericProfile: candidate.NumericProfile(), }) if err != nil { return false, err diff --git a/pkg/sql/compile/sirius_runtime_test.go b/pkg/sql/compile/sirius_runtime_test.go index 7b53894a2a0bb..f2c280f8efee5 100644 --- a/pkg/sql/compile/sirius_runtime_test.go +++ b/pkg/sql/compile/sirius_runtime_test.go @@ -191,6 +191,8 @@ type siriusAdmissionBackend struct { func (b *siriusAdmissionBackend) Accepting() bool { return b.accepting } +func (*siriusAdmissionBackend) Capabilities() uint64 { return 0 } + func TestEmbeddedSiriusAdmissionNeverSilentlyFallsBack(t *testing.T) { proc := testutil.NewProcess(t) runtime := moruntime.ServiceRuntime(proc.GetService()) diff --git a/pkg/sql/compile/siriusbridge/bridge.go b/pkg/sql/compile/siriusbridge/bridge.go index ca9c833c91e7b..4ef19d32e6c77 100644 --- a/pkg/sql/compile/siriusbridge/bridge.go +++ b/pkg/sql/compile/siriusbridge/bridge.go @@ -257,6 +257,7 @@ func vectorPayloadBytes(vectors []Vector) (uint64, error) { type Runtime struct { mu sync.Mutex native driver + capabilities uint64 // Immutable after New publishes the runtime. closing bool queries map[*Query]struct{} inflight int @@ -270,6 +271,15 @@ type Runtime struct { cleanupTimeout time.Duration } +// Capabilities returns the native ABI snapshot taken before runtime publication. +// Query admission never polls or mutates process-global native capabilities. +func (r *Runtime) Capabilities() uint64 { + if r == nil { + return 0 + } + return r.capabilities +} + func newRuntime(d driver) *Runtime { done := make(chan struct{}) close(done) diff --git a/pkg/sql/compile/siriusbridge/native.go b/pkg/sql/compile/siriusbridge/native.go index e786cc4efc87a..dd57f0c1cf713 100644 --- a/pkg/sql/compile/siriusbridge/native.go +++ b/pkg/sql/compile/siriusbridge/native.go @@ -50,7 +50,11 @@ func status(code C.sirius_status, e *C.sirius_error) error { if code == C.SIRIUS_NOT_NEEDED { return errNotNeeded } - return &nativeError{uint32(code), C.GoStringN(&e.message[0], C.int(boundedMessage(e)))} + message := C.GoStringN(&e.message[0], C.int(boundedMessage(e))) + if err := nativeNumericError(uint32(code), message); err != nil { + return err + } + return &nativeError{uint32(code), message} } func boundedMessage(e *C.sirius_error) int { for i := range e.message { @@ -92,7 +96,8 @@ func New(config Config) (*Runtime, error) { if err := config.Validate(); err != nil { return nil, err } - if C.sirius_abi_version() != 1 || C.sirius_capabilities()&11 != 11 { + capabilities := uint64(C.sirius_capabilities()) + if C.sirius_abi_version() != 1 || capabilities&11 != 11 { return nil, moerr.NewBadConfigNoCtx("Sirius SDK lacks ABI v1 MO input/native result capability") } path := C.CString(config.ConfigPath) @@ -105,6 +110,7 @@ func New(config Config) (*Runtime, error) { return nil, err } runtime := newRuntime(d) + runtime.capabilities = capabilities runtime.cleanupTimeout = config.cleanupBudget() if config.MaxWaiting != 0 { runtime.maxQueries = 1 + int(config.MaxWaiting) diff --git a/pkg/sql/compile/siriusbridge/numeric_error.go b/pkg/sql/compile/siriusbridge/numeric_error.go new file mode 100644 index 0000000000000..9814f5a3704df --- /dev/null +++ b/pkg/sql/compile/siriusbridge/numeric_error.go @@ -0,0 +1,30 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package siriusbridge + +import "github.com/matrixorigin/matrixone/pkg/common/moerr" + +// Numeric statuses are append-only ABI-v1 values. Classify by the operation's +// native status, never by diagnostic text or the enclosing SQL expression. +func nativeNumericError(code uint32, message string) error { + switch code { + case 12: + return moerr.NewOutOfRangeNoCtx("decimal", message) + case 13: + return moerr.NewInvalidInputNoCtx(message) + default: + return nil + } +} diff --git a/pkg/sql/compile/siriusbridge/numeric_error_test.go b/pkg/sql/compile/siriusbridge/numeric_error_test.go new file mode 100644 index 0000000000000..203d61ca3ea4b --- /dev/null +++ b/pkg/sql/compile/siriusbridge/numeric_error_test.go @@ -0,0 +1,46 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package siriusbridge + +import ( + "context" + "errors" + "testing" + + "github.com/matrixorigin/matrixone/pkg/common/moerr" + "github.com/stretchr/testify/require" +) + +func TestNativeNumericStatusesPreservePublicErrorIdentity(t *testing.T) { + for _, tc := range []struct { + status uint32 + code uint16 + mysql uint16 + state string + }{{12, moerr.ErrOutOfRange, 1690, "22003"}, {13, moerr.ErrInvalidInput, 1105, "HY000"}} { + err := nativeNumericError(tc.status, "bounded native numeric failure") + require.True(t, moerr.IsMoErrCode(err, tc.code)) + joined := errors.Join(err, context.Canceled) + require.ErrorIs(t, joined, context.Canceled) + var typed *moerr.Error + require.ErrorAs(t, joined, &typed) + require.Equal(t, tc.code, typed.ErrorCode()) + require.Equal(t, tc.mysql, typed.MySQLCode()) + require.Equal(t, tc.state, typed.SqlState()) + } + for _, code := range []uint32{0, 1, 2, 4, 5, 6, 7, 8, 9, 10, 11, 14} { + require.Nil(t, nativeNumericError(code, "not a numeric status")) + } +} diff --git a/pkg/sql/plan/substrait/embedded.go b/pkg/sql/plan/substrait/embedded.go index aa292db814cdb..91c9b3d1c1d90 100644 --- a/pkg/sql/plan/substrait/embedded.go +++ b/pkg/sql/plan/substrait/embedded.go @@ -71,7 +71,7 @@ func (c *Candidate) EmbeddedMOReads() ([]EmbeddedMORead, error) { if err != nil { return nil, err } - result[i], _, err = embeddedMORead(node) + result[i], _, err = embeddedMORead(node, c.profile) if err != nil { return nil, err } @@ -90,7 +90,7 @@ func (c *Candidate) BuildEmbedded(bindings map[int32]EmbeddedReadBinding) ([]byt if err := c.validateEmbeddedBindings(bindings); err != nil { return nil, err } - e := exporter{query: c.query, embeddedBindings: bindings} + e := exporter{query: c.query, embeddedBindings: bindings, embeddedMO: true, profile: c.profile} relations := make([]*spb.PlanRel, 0, len(c.query.Steps)) for step, rootID := range c.query.Steps { e.stepOrdinal = int32(step) @@ -105,9 +105,10 @@ func (c *Candidate) BuildEmbedded(bindings map[int32]EmbeddedReadBinding) ([]byt } } plan := &spb.Plan{ - Version: &spb.Version{MajorNumber: 0, MinorNumber: 78, PatchNumber: 0, Producer: "matrixone"}, - Relations: relations, - Extensions: e.extensions(), + Version: &spb.Version{MajorNumber: 0, MinorNumber: 78, PatchNumber: 0, Producer: "matrixone"}, + Relations: relations, + Extensions: e.extensions(), + ExtensionUrns: e.extensionURNs(), } wire, err := proto.MarshalOptions{Deterministic: true}.Marshal(plan) if err != nil { @@ -150,7 +151,7 @@ func (c *Candidate) validateEmbeddedBindings(bindings map[int32]EmbeddedReadBind if err != nil { return err } - if _, _, err = embeddedMORead(node); err != nil { + if _, _, err = embeddedMORead(node, c.profile); err != nil { return err } } @@ -218,8 +219,8 @@ func (c *Candidate) embeddedReadNode(id int32) (*planpb.Node, error) { return node, nil } -func embeddedMORead(node *planpb.Node) (EmbeddedMORead, *spb.NamedStruct, error) { - physicalSchema, err := namedStruct(node.TableDef) +func embeddedMORead(node *planpb.Node, profile EmbeddedExportProfile) (EmbeddedMORead, *spb.NamedStruct, error) { + physicalSchema, err := namedStructForProfile(node.TableDef, profile) if err != nil { return EmbeddedMORead{}, nil, err } @@ -252,7 +253,7 @@ func embeddedMORead(node *planpb.Node) (EmbeddedMORead, *spb.NamedStruct, error) if expression.Typ.Id != physicalType.Id || expression.Typ.Width != physicalType.Width || expression.Typ.Scale != physicalType.Scale { return EmbeddedMORead{}, nil, notEligiblef(EligibilityType, "embedded MO scan node %d output %d type does not match source column %d", node.NodeId, output, ordinal) } - if _, typeErr := substraitType(&expression.Typ); typeErr != nil { + if _, typeErr := profile.substraitType(&expression.Typ); typeErr != nil { return EmbeddedMORead{}, nil, typeErr } ordinals[output] = ordinal @@ -273,7 +274,7 @@ func embeddedMORead(node *planpb.Node) (EmbeddedMORead, *spb.NamedStruct, error) for output, ordinal := range ordinals { column := visible[ordinal] name := "col_" + strconv.Itoa(output) - typ, typeErr := substraitType(&outputTypes[output]) + typ, typeErr := profile.substraitType(&outputTypes[output]) if typeErr != nil { return EmbeddedMORead{}, nil, typeErr } diff --git a/pkg/sql/plan/substrait/embedded_test.go b/pkg/sql/plan/substrait/embedded_test.go index dae6a049648d6..875835e892071 100644 --- a/pkg/sql/plan/substrait/embedded_test.go +++ b/pkg/sql/plan/substrait/embedded_test.go @@ -54,7 +54,7 @@ func TestEmbeddedMOAdmissionPreservesReaderOrderingWithoutChangingFlight(t *test query.Nodes[0].OrderBy = []*planpb.OrderBySpec{{Expr: col(0), Flag: planpb.OrderBySpec_ASC}} _, err := Export(query) require.ErrorContains(t, err, "sort semantics outside a SORT node") - candidate, err := ExportEmbeddedMO(query) + candidate, err := ExportEmbeddedMO(query, EmbeddedExportProfile{}) require.NoError(t, err) _, err = candidate.Build(map[int32][]byte{0: {1}}) require.ErrorContains(t, err, "cannot be emitted as Flight") @@ -64,7 +64,7 @@ func TestEmbeddedMOAdmissionPreservesReaderOrderingWithoutChangingFlight(t *test require.Len(t, query.Nodes[0].OrderBy, 1, "the MO producer must still receive the original ordering hint") query.Nodes = append(query.Nodes, &planpb.Node{NodeId: 1, NodeType: planpb.Node_PROJECT, Children: []int32{0}, ProjectList: []*planpb.Expr{col(0)}, OrderBy: query.Nodes[0].OrderBy}) query.Steps = []int32{1} - _, err = ExportEmbeddedMO(query) + _, err = ExportEmbeddedMO(query, EmbeddedExportProfile{}) require.ErrorContains(t, err, "sort semantics outside a SORT node", "only MO-owned scan annotations are admitted") } @@ -258,3 +258,40 @@ func embeddedPlan(t *testing.T, wire []byte) *spb.Plan { require.NoError(t, proto.Unmarshal(wire, plan)) return plan } + +func TestEmbeddedFetchIncludesBothConstantBounds(t *testing.T) { + for _, tc := range []struct { + name string + limit, offset *planpb.Expr + count, start int64 + }{ + {name: "limit only", limit: u64(5), count: 5}, + {name: "offset only", offset: u64(3), count: -1, start: 3}, + {name: "zero limit", limit: u64(0)}, + {name: "both", limit: u64(5), offset: u64(3), count: 5, start: 3}, + } { + t.Run(tc.name, func(t *testing.T) { + e := exporter{embeddedMO: true} + relation, err := e.fetch(&spb.Rel{}, &planpb.Node{Limit: tc.limit, Offset: tc.offset}) + require.NoError(t, err) + wire, err := proto.Marshal(relation) + require.NoError(t, err) + decoded := new(spb.Rel) + require.NoError(t, proto.Unmarshal(wire, decoded)) + fetch := decoded.GetFetch() + require.IsType(t, &spb.FetchRel_Count{}, fetch.CountMode) + require.IsType(t, &spb.FetchRel_Offset{}, fetch.OffsetMode) + require.Equal(t, tc.count, fetch.GetCount()) + require.Equal(t, tc.start, fetch.GetOffset()) + }) + } + input := &spb.Rel{} + e := exporter{embeddedMO: true} + unchanged, err := e.fetch(input, &planpb.Node{}) + require.NoError(t, err) + require.Same(t, input, unchanged) + e.embeddedMO = false + flight, err := e.fetch(input, &planpb.Node{Limit: u64(5)}) + require.NoError(t, err) + require.Nil(t, flight.GetFetch().OffsetMode, "preserve ordinary Flight emission") +} diff --git a/pkg/sql/plan/substrait/export.go b/pkg/sql/plan/substrait/export.go index 7ed88be31d12d..102d73fbf902f 100644 --- a/pkg/sql/plan/substrait/export.go +++ b/pkg/sql/plan/substrait/export.go @@ -53,6 +53,7 @@ type Candidate struct { headings []string types []planpb.Type embeddedMO bool + profile EmbeddedExportProfile } // Read identifies one physical table scan which needs a TaeRead lease. @@ -97,17 +98,17 @@ func (c *Candidate) OutputTypes() []planpb.Type { // Export validates q without performing I/O. func Export(q *planpb.Query) (*Candidate, error) { - return exportCandidate(q, false) + return exportCandidate(q, false, EmbeddedExportProfile{}) } // ExportEmbeddedMO validates a plan whose entire scans remain owned by MO. // Storage ordering hints are legal only at those scan boundaries; Flight // cannot claim their semantics because it bypasses the MO reader. -func ExportEmbeddedMO(q *planpb.Query) (*Candidate, error) { - return exportCandidate(q, true) +func ExportEmbeddedMO(q *planpb.Query, profile EmbeddedExportProfile) (*Candidate, error) { + return exportCandidate(q, true, profile) } -func exportCandidate(q *planpb.Query, embeddedMO bool) (*Candidate, error) { +func exportCandidate(q *planpb.Query, embeddedMO bool, profile EmbeddedExportProfile) (*Candidate, error) { if q == nil { return nil, moerr.NewInternalErrorNoCtx("substrait: missing query") } @@ -125,8 +126,8 @@ func exportCandidate(q *planpb.Query, embeddedMO bool) (*Candidate, error) { return nil, notEligiblef(EligibilityOperator, "node %d carries unsupported rank semantics", nodeID) } } - c := &Candidate{query: q, embeddedMO: embeddedMO} - e := exporter{query: q, readValues: make(map[int32][]byte), validateOnly: true, embeddedMO: embeddedMO} + c := &Candidate{query: q, embeddedMO: embeddedMO, profile: profile} + e := exporter{query: q, readValues: make(map[int32][]byte), validateOnly: true, embeddedMO: embeddedMO, profile: profile} for step, rootID := range q.Steps { if rootID < 0 || int(rootID) >= len(q.Nodes) { return nil, moerr.NewInternalErrorNoCtxf("substrait: invalid root node id %d at step %d", rootID, step) @@ -160,6 +161,13 @@ func exportCandidate(q *planpb.Query, embeddedMO bool) (*Candidate, error) { if len(c.types) != width { return nil, moerr.NewInternalErrorNoCtxf("substrait: query root has %d output types for %d columns", len(c.types), width) } + if profile.exactDecimalV1 { + for i := range c.types { + if exactDecimalBits(&c.types[i]) == 256 && c.types[i].Width > 65 { + return nil, notEligiblef(EligibilityType, "public exact decimal precision exceeds 65") + } + } + } return c, nil } @@ -205,6 +213,8 @@ func (c *Candidate) Build(readValues map[int32][]byte) ([]byte, error) { } type exporter struct { + profile EmbeddedExportProfile + exactUsed bool embeddedMO bool query *planpb.Query readValues map[int32][]byte @@ -243,7 +253,7 @@ func (e *exporter) node(id int32) (*spb.Rel, error) { switch n.NodeType { case planpb.Node_TABLE_SCAN: rel, err = e.read(n) - if err == nil && e.embeddedBindings != nil { + if err == nil && (e.embeddedBindings != nil || e.embeddedMO) { return rel, nil } case planpb.Node_FILTER: @@ -547,7 +557,7 @@ func (e *exporter) join(n *planpb.Node) (*spb.Rel, error) { inputs := []int{leftWidth, rightWidth} var condition *spb.Expression if len(n.OnList) == 0 { - condition, err = literal(&planpb.Literal{Value: &planpb.Literal_Bval{Bval: true}}, &planpb.Type{Id: int32(types.T_bool), NotNullable: true}) + condition, err = e.literal(&planpb.Literal{Value: &planpb.Literal_Bval{Bval: true}}, &planpb.Type{Id: int32(types.T_bool), NotNullable: true}) } else { if err = validateExprFields(n.OnList, inputs); err != nil { return nil, err @@ -611,7 +621,7 @@ func (e *exporter) read(n *planpb.Node) (*spb.Rel, error) { if n.IndexScanInfo.ProtoSize() != 0 || n.IndexReaderParam != nil { return nil, notEligiblef(EligibilityOperator, "node %d carries unsupported index scan semantics", n.NodeId) } - schema, err := namedStruct(n.TableDef) + schema, err := e.namedStruct(n.TableDef) if err != nil { return nil, err } @@ -639,22 +649,32 @@ func (e *exporter) read(n *planpb.Node) (*spb.Rel, error) { }) } } - if e.embeddedBindings != nil { - if e.embeddedReadSeen == nil { - e.embeddedReadSeen = make(map[int32]bool) - } - if e.embeddedReadSeen[n.NodeId] { - return nil, moerr.NewInternalErrorNoCtxf("substrait: embedded read node %d is replayed", n.NodeId) - } - e.embeddedReadSeen[n.NodeId] = true - binding, ok := e.embeddedBindings[n.NodeId] - if !ok { - return nil, moerr.NewInternalErrorNoCtxf("substrait: missing embedded binding for node %d", n.NodeId) + if e.embeddedBindings != nil || e.embeddedMO { + // Validation and emission consume the same post-scan contract. MO owns + // this scan's filters/projection/fetch; never validate them as a Flight + // expression and then silently omit them in the embedded build pass. + binding := EmbeddedReadBinding{BindingID: uint64(len(e.reads)), Source: EmbeddedReadMO} + if e.embeddedBindings != nil { + if e.embeddedReadSeen == nil { + e.embeddedReadSeen = make(map[int32]bool) + } + if e.embeddedReadSeen[n.NodeId] { + return nil, moerr.NewInternalErrorNoCtxf("substrait: embedded read node %d is replayed", n.NodeId) + } + e.embeddedReadSeen[n.NodeId] = true + var ok bool + binding, ok = e.embeddedBindings[n.NodeId] + if !ok { + return nil, moerr.NewInternalErrorNoCtxf("substrait: missing embedded binding for node %d", n.NodeId) + } } - _, outputSchema, embeddedErr := embeddedMORead(n) + _, outputSchema, embeddedErr := embeddedMORead(n, e.profile) if embeddedErr != nil { return nil, embeddedErr } + for _, typ := range outputSchema.Struct.Types { + e.exactUsed = e.exactUsed || typ.GetUserDefined() != nil + } return embeddedNamedRead(binding, outputSchema), nil } value := e.readValues[n.NodeId] @@ -732,7 +752,7 @@ func (e *exporter) aggregate(n *planpb.Node) (*spb.Rel, error) { if len(n.GroupBy) == 0 && aggregateCanReturnNullOnEmpty(functionID) { outputType.NotNullable = false } - supported, capabilityErr := hasSemanticCapability(semanticAggregate, name, f.Func, f.Args, &outputType) + supported, capabilityErr := e.hasSemanticCapability(semanticAggregate, name, f.Func, f.Args, &outputType) if capabilityErr != nil { return nil, capabilityErr } @@ -750,10 +770,13 @@ func (e *exporter) aggregate(n *planpb.Node) (*spb.Rel, error) { } args[j] = valueArg(a) } - out, xerr := substraitType(&outputType) + out, xerr := e.substraitType(&outputType) if xerr != nil { return nil, xerr } + if e.profile.exactDecimalV1 && exactDecimalBits(&outputType) != 0 { + name = "mo_decimal_" + name + } measures[i] = &spb.AggregateRel_Measure{Measure: &spb.AggregateFunction{FunctionReference: e.function(name), Arguments: args, OutputType: out, Phase: spb.AggregationPhase_AGGREGATION_PHASE_INITIAL_TO_RESULT, Invocation: spb.AggregateFunction_AGGREGATION_INVOCATION_ALL}} } relation := &spb.Rel{RelType: &spb.Rel_Aggregate{Aggregate: &spb.AggregateRel{Input: input, GroupingExpressions: groups, Groupings: []*spb.AggregateRel_Grouping{{ExpressionReferences: refs}}, Measures: measures}}} @@ -832,6 +855,12 @@ func (e *exporter) fetch(input *spb.Rel, n *planpb.Node) (*spb.Rel, error) { return input, nil } fetch := &spb.FetchRel{Input: input} + if e.embeddedMO { + // ABI-v1 preparation requires both constant modes to be explicit. + // Substrait's absent count means all rows and absent offset means zero. + fetch.CountMode = &spb.FetchRel_Count{Count: -1} + fetch.OffsetMode = &spb.FetchRel_Offset{Offset: 0} + } if n.Limit != nil { count, err := nonnegativeIntLiteral(n.Limit, 0) if err != nil { @@ -888,9 +917,18 @@ func (e *exporter) expr(x *planpb.Expr, inputs []int) (*spb.Expression, error) { if err != nil { return nil, err } + if e.profile.exactDecimalV1 && exactDecimalBits(&x.Typ) != 0 { + if _, err := e.substraitType(&x.Typ); err != nil { + return nil, err + } + // A field selection has no declared output type. Retain MO's + // descriptor, including a projection's nullable view of a required + // grouped aggregate, through the checked exact function. + return e.scalar("mo_decimal_cast", &x.Typ, field(ordinal)), nil + } return field(ordinal), nil case *planpb.Expr_Lit: - return literal(v.Lit, &x.Typ) + return e.literal(v.Lit, &x.Typ) case *planpb.Expr_List: return nil, notEligiblef(EligibilityExpression, "an expression list is only valid as an IN argument") case *planpb.Expr_F: @@ -898,6 +936,9 @@ func (e *exporter) expr(x *planpb.Expr, inputs []int) (*spb.Expression, error) { return nil, moerr.NewInternalErrorNoCtxf("substrait: malformed function") } functionID, _ := function.DecodeOverloadID(v.F.Func.Obj) + if e.profile.exactDecimalV1 && (decimalSignature(v.F.Args, &x.Typ) || functionID == function.COALESCE) { + return e.exactScalarExpr(x, v.F, inputs) + } switch functionID { case function.CAST: return e.castExpr(x, v.F, inputs) @@ -921,14 +962,14 @@ func (e *exporter) expr(x *planpb.Expr, inputs []int) (*spb.Expression, error) { if err := validateScalarSignature(name, &x.Typ, v.F.Args); err != nil { return nil, err } - supported, capabilityErr := hasSemanticCapability(semanticScalar, name, v.F.Func, v.F.Args, &x.Typ) + supported, capabilityErr := e.hasSemanticCapability(semanticScalar, name, v.F.Func, v.F.Args, &x.Typ) if capabilityErr != nil { return nil, capabilityErr } if !supported { return nil, notEligiblef(EligibilityExpression, "scalar overload %q has no declared Sirius semantic equivalence", v.F.Func.ObjName) } - if _, err := substraitType(&x.Typ); err != nil { + if _, err := e.substraitType(&x.Typ); err != nil { return nil, err } args := make([]*spb.Expression, len(v.F.Args)) @@ -949,7 +990,7 @@ func (e *exporter) unaryMinusExpr(result *planpb.Expr, call *planpb.Function, in if len(call.Args) != 1 || call.Args[0] == nil || !isDecimalType(types.T(call.Args[0].Typ.Id)) || semanticTypeFromPlan(&call.Args[0].Typ) != semanticTypeFromPlan(&result.Typ) { return nil, notEligiblef(EligibilityExpression, "unsupported unary minus signature") } - supported, err := hasSemanticCapability(semanticScalar, "unary_minus", call.Func, call.Args, &result.Typ) + supported, err := e.hasSemanticCapability(semanticScalar, "unary_minus", call.Func, call.Args, &result.Typ) if err != nil { return nil, err } @@ -966,7 +1007,7 @@ func (e *exporter) unaryMinusExpr(result *planpb.Expr, call *planpb.Function, in } else { zero.Value = &planpb.Literal_Decimal128Val{Decimal128Val: &planpb.Decimal128{}} } - zeroExpr, err := literal(zero, &result.Typ) + zeroExpr, err := e.literal(zero, &result.Typ) if err != nil { return nil, err } @@ -1000,7 +1041,7 @@ func (e *exporter) castExpr(result *planpb.Expr, call *planpb.Function, inputs [ if len(call.Args) != 2 { return nil, notEligiblef(EligibilityExpression, "cast requires a value and target type descriptor") } - supported, err := hasSemanticCapability(semanticScalar, "cast", call.Func, call.Args, &result.Typ) + supported, err := e.hasSemanticCapability(semanticScalar, "cast", call.Func, call.Args, &result.Typ) if err != nil { return nil, err } @@ -1015,7 +1056,7 @@ func (e *exporter) castExpr(result *planpb.Expr, call *planpb.Function, inputs [ if err != nil { return nil, err } - target, err := substraitType(&result.Typ) + target, err := e.substraitType(&result.Typ) if err != nil { return nil, err } @@ -1028,7 +1069,7 @@ func (e *exporter) caseExpr(result *planpb.Expr, call *planpb.Function, inputs [ if len(call.Args) < 3 || len(call.Args)%2 == 0 { return nil, notEligiblef(EligibilityExpression, "case requires condition/result pairs and an else expression") } - supported, err := hasSemanticCapability(semanticScalar, "if_then", call.Func, call.Args, &result.Typ) + supported, err := e.hasSemanticCapability(semanticScalar, "if_then", call.Func, call.Args, &result.Typ) if err != nil { return nil, err } @@ -1061,7 +1102,7 @@ func (e *exporter) inExpr(result *planpb.Expr, call *planpb.Function, inputs []i if len(call.Args) < 2 { return nil, notEligiblef(EligibilityExpression, "in requires a value and at least one option") } - supported, err := hasSemanticCapability(semanticScalar, "singular_or_list", call.Func, call.Args, &result.Typ) + supported, err := e.hasSemanticCapability(semanticScalar, "singular_or_list", call.Func, call.Args, &result.Typ) if err != nil { return nil, err } @@ -1169,7 +1210,10 @@ func (e *exporter) extractExpr(result *planpb.Expr, call *planpb.Function, input default: return nil, notEligiblef(EligibilityExpression, "extract field %q has no declared Sirius semantic equivalence", unit) } - supported, err := hasSemanticCapability(semanticScalar, "extract", call.Func, call.Args, &result.Typ) + if e.embeddedMO && unit == "quarter" { + return nil, notEligiblef(EligibilityExpression, "embedded extract field %q has no native GPU function", unit) + } + supported, err := e.hasSemanticCapability(semanticScalar, "extract", call.Func, call.Args, &result.Typ) if err != nil { return nil, err } @@ -1180,7 +1224,47 @@ func (e *exporter) extractExpr(result *planpb.Expr, call *planpb.Function, input if err != nil { return nil, err } - output, err := substraitType(&result.Typ) + if e.embeddedMO { + // The generic importer lowers extract to date_part, which has no + // Sirius GPU expression. Ordinary components also interpret MO's zero + // DATE sentinel as calendar 0000-12-31 after native input conversion. + // Guard that exact sentinel in the typed plan, preserving NULL behavior + // and the complete MO domain without changing published reader bytes. + zeroDate, err := e.literal(&planpb.Literal{Value: &planpb.Literal_Dateval{Dateval: int32(types.ZeroDate)}}, &planpb.Type{Id: int32(types.T_date), NotNullable: true}) + if err != nil { + return nil, err + } + zero, err := e.literal(&planpb.Literal{Value: &planpb.Literal_I64Val{}}, &planpb.Type{Id: int32(types.T_int64), NotNullable: true}) + if err != nil { + return nil, err + } + conditionType := &planpb.Type{Id: int32(types.T_bool), NotNullable: call.Args[1].Typ.NotNullable} + condition := e.scalar("equal", conditionType, value, zeroDate) + output, err := e.substraitType(&result.Typ) + if err != nil { + return nil, err + } + // The pinned component function physically returns INT16 but advertises + // BIGINT. A same-type cast is removed by binding; use the proven INT32 + // component domain before BIGINT so CASE receives equal physical types. + componentType := result.Typ + componentType.Id = int32(types.T_int32) + component, err := e.substraitType(&componentType) + if err != nil { + return nil, err + } + part := &spb.Expression{RexType: &spb.Expression_Cast_{Cast: &spb.Expression_Cast{ + Type: component, Input: e.scalar(unit, &result.Typ, value), FailureBehavior: spb.Expression_Cast_FAILURE_BEHAVIOR_THROW_EXCEPTION, + }}} + part = &spb.Expression{RexType: &spb.Expression_Cast_{Cast: &spb.Expression_Cast{ + Type: output, Input: part, FailureBehavior: spb.Expression_Cast_FAILURE_BEHAVIOR_THROW_EXCEPTION, + }}} + return &spb.Expression{RexType: &spb.Expression_IfThen_{IfThen: &spb.Expression_IfThen{ + Ifs: []*spb.Expression_IfThen_IfClause{{If: condition, Then: zero}}, + Else: part, + }}}, nil + } + output, err := e.substraitType(&result.Typ) if err != nil { return nil, err } @@ -1194,7 +1278,7 @@ func (e *exporter) extractExpr(result *planpb.Expr, call *planpb.Function, input } func (e *exporter) scalar(name string, typ *planpb.Type, args ...*spb.Expression) *spb.Expression { - out, _ := substraitType(typ) + out, _ := e.substraitType(typ) fargs := make([]*spb.FunctionArgument, len(args)) for i := range args { fargs[i] = valueArg(args[i]) @@ -1203,6 +1287,9 @@ func (e *exporter) scalar(name string, typ *planpb.Type, args ...*spb.Expression } func (e *exporter) function(name string) uint32 { + if strings.HasPrefix(name, "mo_decimal_") { + e.exactUsed = true + } if e.functions == nil { e.functions = make(map[string]uint32) } @@ -1220,13 +1307,28 @@ func (e *exporter) extensions() []*extensions.SimpleExtensionDeclaration { byAnchor[anchor] = name } result := make([]*extensions.SimpleExtensionDeclaration, 0, len(e.functions)) + if e.exactUsed { + result = append(result, &extensions.SimpleExtensionDeclaration{MappingType: &extensions.SimpleExtensionDeclaration_ExtensionType_{ + ExtensionType: &extensions.SimpleExtensionDeclaration_ExtensionType{ + ExtensionUrnReference: exactDecimalURNAnchor, TypeAnchor: exactDecimalTypeAnchor, Name: "mo_exact_decimal", + }, + }}) + } for anchor := 1; anchor < len(byAnchor); anchor++ { - result = append(result, &extensions.SimpleExtensionDeclaration{MappingType: &extensions.SimpleExtensionDeclaration_ExtensionFunction_{ExtensionFunction: &extensions.SimpleExtensionDeclaration_ExtensionFunction{FunctionAnchor: uint32(anchor), Name: byAnchor[anchor]}}}) + ref := uint32(0) + if strings.HasPrefix(byAnchor[anchor], "mo_decimal_") { + ref = exactDecimalURNAnchor + } + result = append(result, &extensions.SimpleExtensionDeclaration{MappingType: &extensions.SimpleExtensionDeclaration_ExtensionFunction_{ExtensionFunction: &extensions.SimpleExtensionDeclaration_ExtensionFunction{FunctionAnchor: uint32(anchor), Name: byAnchor[anchor], ExtensionUrnReference: ref}}}) } return result } func namedStruct(t *planpb.TableDef) (*spb.NamedStruct, error) { + return namedStructForProfile(t, EmbeddedExportProfile{}) +} + +func namedStructForProfile(t *planpb.TableDef, profile EmbeddedExportProfile) (*spb.NamedStruct, error) { names := make([]string, 0, len(t.Cols)) fields := make([]*spb.Type, 0, len(t.Cols)) hidden := false @@ -1241,7 +1343,7 @@ func namedStruct(t *planpb.TableDef) (*spb.NamedStruct, error) { if hidden { return nil, moerr.NewInternalErrorNoCtxf("substrait: table %q has non-suffix hidden columns", t.Name) } - typ, err := substraitType(&c.Typ) + typ, err := profile.substraitType(&c.Typ) if err != nil { if IsNotEligible(err) { return nil, err diff --git a/pkg/sql/plan/substrait/numeric_native_integration_test.go b/pkg/sql/plan/substrait/numeric_native_integration_test.go new file mode 100644 index 0000000000000..918afff5c05a6 --- /dev/null +++ b/pkg/sql/plan/substrait/numeric_native_integration_test.go @@ -0,0 +1,100 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build sirius && sirius_integration && cgo && linux && amd64 + +package substrait + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/matrixorigin/matrixone/pkg/sql/compile/siriusbridge" + planbuilder "github.com/matrixorigin/matrixone/pkg/sql/plan" + "github.com/stretchr/testify/require" +) + +// Preparation validates the complete native graph and output schema without +// admitting any reader work. Execution parity belongs to the public SQL test +// and the separately versioned full-data campaign. +func TestExactEmbeddedTPCHNativePreparation(t *testing.T) { + config := os.Getenv("MO_SIRIUS_TEST_CONFIG") + require.NotEmpty(t, config, "sirius_integration requires MO_SIRIUS_TEST_CONFIG") + config, err := filepath.Abs(config) + require.NoError(t, err) + runtime, err := siriusbridge.New(siriusbridge.Config{ConfigPath: config, GPUStreams: 2}) + require.NoError(t, err) + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + require.NoError(t, runtime.Close(ctx)) + }) + require.NotZero(t, runtime.Capabilities()&MOExactDecimalV1Capability) + mock := planbuilder.NewMockOptimizer(false, newPlanTestProcess(t)) + for number := 1; number <= 22; number++ { + t.Run(fmt.Sprintf("q%d", number), func(t *testing.T) { + bound := exactTPCHQuery(t, mock, number) + candidate, err := ExportEmbeddedMO(bound, NewEmbeddedExportProfile(runtime.Capabilities())) + require.NoError(t, err) + reads, err := candidate.EmbeddedMOReads() + require.NoError(t, err) + bindings := make(map[int32]EmbeddedReadBinding, len(reads)) + request := siriusbridge.Request{AccountID: 42, QueryID: []byte(fmt.Sprintf("mo-exact-q%d", number))} + for i, typ := range candidate.OutputTypes() { + request.Columns = append(request.Columns, siriusbridge.Column{ + OID: uint32(typ.Id), Width: typ.Width, Scale: typ.Scale, Nullable: !typ.NotNullable, Name: bound.Headings[i], + }) + } + var starts atomic.Int32 + for i, read := range reads { + binding := uint64(i + 1) + bindings[read.NodeID] = EmbeddedReadBinding{BindingID: binding, Source: EmbeddedReadMO} + nativeRead := siriusbridge.Read{BindingID: binding, Database: read.Database, Table: read.Table, Schema: read.Schema, + Producer: func(context.Context, *siriusbridge.Input) error { starts.Add(1); return nil }} + for _, col := range read.Columns { + nativeRead.Columns = append(nativeRead.Columns, siriusbridge.ReadColumn{ + Column: siriusbridge.Column{OID: uint32(col.Type.Id), Width: col.Type.Width, Scale: col.Type.Scale, Nullable: !col.Type.NotNullable, Name: col.Name}, + PhysicalID: col.PhysicalID, Sequence: col.Sequence, + }) + } + request.Reads = append(request.Reads, nativeRead) + } + request.Plan, err = candidate.BuildEmbedded(bindings) + require.NoError(t, err) + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + query, err := runtime.Prepare(ctx, request) + if err != nil { + t.Logf("admitted output descriptors: %+v", request.Columns) + } + require.NoError(t, err) + t.Cleanup(func() { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + require.NoError(t, query.Close(cleanupCtx)) + }) + require.NoError(t, query.Close(ctx)) + require.Zero(t, starts.Load(), "preparation must not start MO readers") + stats, ready := query.Statistics() + require.True(t, ready) + require.Zero(t, stats.GPUTasksStarted) + require.Zero(t, stats.MOInputUnits) + }) + } +} diff --git a/pkg/sql/plan/substrait/numeric_profile.go b/pkg/sql/plan/substrait/numeric_profile.go new file mode 100644 index 0000000000000..90f7d55f5d136 --- /dev/null +++ b/pkg/sql/plan/substrait/numeric_profile.go @@ -0,0 +1,174 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package substrait + +import ( + "encoding/binary" + + "github.com/matrixorigin/matrixone/pkg/common/moerr" + "github.com/matrixorigin/matrixone/pkg/container/types" + planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" + spb "github.com/substrait-io/substrait-protobuf/go/substraitpb" + "github.com/substrait-io/substrait-protobuf/go/substraitpb/extensions" + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/types/known/anypb" +) + +const ( + // Mask value, not a bit index: this is the append-only Sirius ABI-v1 flag. + MOExactDecimalV1Capability uint64 = 16 + MOExactDecimalV1URI = "urn:matrixone:sirius:exact-decimal:v1" + exactDecimalLiteralURL = "type.googleapis.com/matrixone.sirius.numeric.v1.ExactDecimalLiteral" + exactDecimalTypeAnchor uint32 = 1 + exactDecimalURNAnchor uint32 = 1 +) + +// EmbeddedExportProfile is copied into a Candidate before validation. Its +// private field cannot be changed by a caller between validation and Build. +type EmbeddedExportProfile struct{ exactDecimalV1 bool } + +func NewEmbeddedExportProfile(capabilities uint64) EmbeddedExportProfile { + return EmbeddedExportProfile{exactDecimalV1: capabilities&MOExactDecimalV1Capability != 0} +} + +func (p EmbeddedExportProfile) Name() string { + if p.exactDecimalV1 { + return "mo-exact-decimal-v1" + } + return "legacy" +} + +func (c *Candidate) NumericProfile() string { + if c == nil { + return "legacy" + } + return c.profile.Name() +} + +func exactDecimalBits(t *planpb.Type) int { + if t == nil { + return 0 + } + switch types.T(t.Id) { + case types.T_decimal64: + return 64 + case types.T_decimal128: + return 128 + case types.T_decimal256: + return 256 + default: + return 0 + } +} + +func (p EmbeddedExportProfile) substraitType(t *planpb.Type) (*spb.Type, error) { + bits := exactDecimalBits(t) + if !p.exactDecimalV1 || bits == 0 { + return substraitType(t) + } + maximum := int32(18) + if bits == 128 { + maximum = 38 + } else if bits == 256 { + maximum = 76 + } + if t.Width < 1 || t.Width > maximum || t.Scale < 0 || t.Scale > t.Width { + return nil, notEligiblef(EligibilityType, "invalid MO exact decimal descriptor (%d,%d,%d)", bits, t.Width, t.Scale) + } + nullable := spb.Type_NULLABILITY_NULLABLE + if t.NotNullable { + nullable = spb.Type_NULLABILITY_REQUIRED + } + return &spb.Type{Kind: &spb.Type_UserDefined_{UserDefined: &spb.Type_UserDefined{ + TypeReference: exactDecimalTypeAnchor, Nullability: nullable, + TypeParameters: []*spb.Type_Parameter{ + {Parameter: &spb.Type_Parameter_Integer{Integer: int64(bits)}}, + {Parameter: &spb.Type_Parameter_Integer{Integer: int64(t.Width)}}, + {Parameter: &spb.Type_Parameter_Integer{Integer: int64(t.Scale)}}, + }, + }}}, nil +} + +func (e *exporter) substraitType(t *planpb.Type) (*spb.Type, error) { + result, err := e.profile.substraitType(t) + if err == nil && result.GetUserDefined() != nil { + e.exactUsed = true + } + return result, err +} + +func (e *exporter) namedStruct(t *planpb.TableDef) (*spb.NamedStruct, error) { + return namedStructForProfile(t, e.profile) +} + +func (e *exporter) extensionURNs() []*extensions.SimpleExtensionURN { + if !e.exactUsed { + return nil + } + return []*extensions.SimpleExtensionURN{{ExtensionUrnAnchor: exactDecimalURNAnchor, Urn: MOExactDecimalV1URI}} +} + +func (e *exporter) literal(l *planpb.Literal, typ *planpb.Type) (*spb.Expression, error) { + if !e.profile.exactDecimalV1 || exactDecimalBits(typ) == 0 { + return literal(l, typ) + } + declared, err := e.substraitType(typ) + if err != nil { + return nil, err + } + if l == nil { + return nil, moerr.NewInternalErrorNoCtx("substrait: missing exact decimal literal") + } + if l.Isnull { + if typ.NotNullable { + return nil, moerr.NewInternalErrorNoCtx("substrait: NULL exact literal has a required type") + } + return &spb.Expression{RexType: &spb.Expression_Literal_{Literal: &spb.Expression_Literal{ + LiteralType: &spb.Expression_Literal_Null{Null: declared}, + }}}, nil + } + coefficient := make([]byte, exactDecimalBits(typ)/8) + switch value := l.Value.(type) { + case *planpb.Literal_Decimal64Val: + if exactDecimalBits(typ) != 64 || value.Decimal64Val == nil { + return nil, moerr.NewInternalErrorNoCtx("substrait: exact Decimal64 literal carrier mismatch") + } + binary.LittleEndian.PutUint64(coefficient, uint64(value.Decimal64Val.A)) + case *planpb.Literal_Decimal128Val: + if exactDecimalBits(typ) != 128 || value.Decimal128Val == nil { + return nil, moerr.NewInternalErrorNoCtx("substrait: exact Decimal128 literal carrier mismatch") + } + binary.LittleEndian.PutUint64(coefficient, uint64(value.Decimal128Val.A)) + binary.LittleEndian.PutUint64(coefficient[8:], uint64(value.Decimal128Val.B)) + default: + return nil, moerr.NewInternalErrorNoCtx("substrait: unsupported exact decimal literal carrier") + } + return exactCoefficientLiteral(coefficient, declared, !typ.NotNullable), nil +} + +func exactCoefficientLiteral(coefficient []byte, declared *spb.Type, nullable bool) *spb.Expression { + // Canonical SDK ExactDecimalLiteral: bytes coefficient_le = 1. Encoding + // through protowire preserves the protobuf contract without a second schema. + payload := protowire.AppendTag(nil, 1, protowire.BytesType) + payload = protowire.AppendBytes(payload, coefficient) + return &spb.Expression{RexType: &spb.Expression_Literal_{Literal: &spb.Expression_Literal{ + Nullable: nullable, + LiteralType: &spb.Expression_Literal_UserDefined_{UserDefined: &spb.Expression_Literal_UserDefined{ + TypeAnchorType: &spb.Expression_Literal_UserDefined_TypeReference{TypeReference: exactDecimalTypeAnchor}, + TypeParameters: declared.GetUserDefined().TypeParameters, + Val: &spb.Expression_Literal_UserDefined_Value{Value: &anypb.Any{TypeUrl: exactDecimalLiteralURL, Value: payload}}, + }}, + }}} +} diff --git a/pkg/sql/plan/substrait/numeric_profile_test.go b/pkg/sql/plan/substrait/numeric_profile_test.go new file mode 100644 index 0000000000000..95aaa24287edf --- /dev/null +++ b/pkg/sql/plan/substrait/numeric_profile_test.go @@ -0,0 +1,378 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package substrait + +import ( + "context" + "encoding/binary" + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/matrixorigin/matrixone/pkg/container/types" + planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/parsers" + "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect" + planbuilder "github.com/matrixorigin/matrixone/pkg/sql/plan" + "github.com/matrixorigin/matrixone/pkg/sql/plan/function" + "github.com/stretchr/testify/require" + spb "github.com/substrait-io/substrait-protobuf/go/substraitpb" + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/proto" +) + +func TestExactDecimalProfilePreservesPhysicalDescriptor(t *testing.T) { + profile := NewEmbeddedExportProfile(31) + require.Equal(t, "legacy", NewEmbeddedExportProfile(1<<16).Name(), "16 is a mask, not a bit index") + for _, oid := range []types.T{types.T_decimal64, types.T_decimal128, types.T_decimal256} { + for _, required := range []bool{false, true} { + typ := planpb.Type{Id: int32(oid), Width: 9, Scale: 2, NotNullable: required} + wire, err := profile.substraitType(&typ) + require.NoError(t, err) + user := wire.GetUserDefined() + require.NotNil(t, user) + require.Equal(t, []int64{int64(exactDecimalBits(&typ)), 9, 2}, []int64{user.TypeParameters[0].GetInteger(), user.TypeParameters[1].GetInteger(), user.TypeParameters[2].GetInteger()}) + require.Equal(t, required, user.Nullability == spb.Type_NULLABILITY_REQUIRED) + } + } + for _, typ := range []planpb.Type{ + {Id: int32(types.T_decimal64), Width: 19}, + {Id: int32(types.T_decimal128), Width: 39}, + {Id: int32(types.T_decimal256), Width: 77}, + {Id: int32(types.T_decimal256), Width: 15, Scale: 16}, + } { + _, err := profile.substraitType(&typ) + require.True(t, IsNotEligible(err)) + } + working := planpb.Type{Id: int32(types.T_decimal256), Width: 76, Scale: 40} + _, err := profile.substraitType(&working) + require.NoError(t, err) + _, err = NewEmbeddedExportProfile(11).substraitType(&working) + require.True(t, IsNotEligible(err)) +} + +func TestExactColumnReferencesRetainDeclaredDescriptor(t *testing.T) { + for _, oid := range []types.T{types.T_decimal64, types.T_decimal128, types.T_decimal256} { + for _, required := range []bool{false, true} { + e := exporter{profile: NewEmbeddedExportProfile(31)} + typ := planpb.Type{Id: int32(oid), Width: 15, Scale: 2, NotNullable: required} + x := &planpb.Expr{Typ: typ, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: 0}}} + wire, err := e.expr(x, []int{1}) + require.NoError(t, err) + cast := wire.GetScalarFunction() + require.NotNil(t, cast) + require.Equal(t, e.functions["mo_decimal_cast"], cast.FunctionReference) + declared := cast.OutputType.GetUserDefined() + require.Equal(t, []int64{int64(exactDecimalBits(&typ)), 15, 2}, []int64{ + declared.TypeParameters[0].GetInteger(), declared.TypeParameters[1].GetInteger(), declared.TypeParameters[2].GetInteger(), + }) + require.Equal(t, required, declared.Nullability == spb.Type_NULLABILITY_REQUIRED) + require.Equal(t, int32(0), cast.Arguments[0].GetValue().GetSelection().GetDirectReference().GetStructField().Field) + legacy := exporter{} + wire, err = legacy.expr(x, []int{1}) + require.NoError(t, err) + require.NotNil(t, wire.GetSelection()) + } + } +} + +func TestExactCandidateRetainsProfileThroughReadAndBuild(t *testing.T) { + query := embeddedProjectedScanQuery() + for _, column := range query.Nodes[0].TableDef.Cols { + column.Typ = planpb.Type{Id: int32(types.T_decimal256), Width: 15, Scale: 2} + } + for _, expr := range query.Nodes[0].ProjectList { + expr.Typ = planpb.Type{Id: int32(types.T_decimal256), Width: 15, Scale: 2} + } + _, err := ExportEmbeddedMO(query, NewEmbeddedExportProfile(11)) + require.True(t, IsNotEligible(err)) + profile := NewEmbeddedExportProfile(31) + candidate, err := ExportEmbeddedMO(query, profile) + require.NoError(t, err) + profile = NewEmbeddedExportProfile(11) + require.Equal(t, "legacy", profile.Name()) + require.Equal(t, "mo-exact-decimal-v1", candidate.NumericProfile()) + reads, err := candidate.EmbeddedMOReads() + require.NoError(t, err) + require.Equal(t, int32(types.T_decimal256), reads[0].Columns[0].Type.Id) + wire, err := candidate.BuildEmbedded(map[int32]EmbeddedReadBinding{0: {BindingID: 1, Source: EmbeddedReadMO}}) + require.NoError(t, err) + plan := embeddedPlan(t, wire) + require.Len(t, plan.ExtensionUrns, 1) + require.Equal(t, MOExactDecimalV1URI, plan.ExtensionUrns[0].Urn) + user := plan.Relations[0].GetRoot().Input.GetRead().BaseSchema.Struct.Types[0].GetUserDefined() + require.Equal(t, int64(256), user.TypeParameters[0].GetInteger(), "precision 15 cannot narrow a 256-bit carrier") + _, err = candidate.Build(map[int32][]byte{0: {1}}) + require.ErrorContains(t, err, "cannot be emitted as Flight") +} + +func exactLiteralBytes(t *testing.T, expression *spb.Expression) []byte { + t.Helper() + user := expression.GetLiteral().GetUserDefined() + require.NotNil(t, user) + value := user.GetValue() + require.Equal(t, exactDecimalLiteralURL, value.TypeUrl) + number, kind, n := protowire.ConsumeTag(value.Value) + require.Equal(t, protowire.Number(1), number) + require.Equal(t, protowire.BytesType, kind) + require.Greater(t, n, 0) + bytes, consumed := protowire.ConsumeBytes(value.Value[n:]) + require.Equal(t, len(value.Value), n+consumed) + return bytes +} + +func TestExactLiteralsUseCanonicalWidthsAndWideNumericProvenance(t *testing.T) { + e := exporter{profile: NewEmbeddedExportProfile(31)} + for _, bits := range []int{64, 128} { + typ := planpb.Type{Id: int32(types.T_decimal64), Width: 15, Scale: 2, NotNullable: true} + literal := &planpb.Literal{Value: &planpb.Literal_Decimal64Val{Decimal64Val: &planpb.Decimal64{A: -125}}} + if bits == 128 { + typ.Id = int32(types.T_decimal128) + literal.Value = &planpb.Literal_Decimal128Val{Decimal128Val: &planpb.Decimal128{A: -125, B: -1}} + } + wire, err := e.literal(literal, &typ) + require.NoError(t, err) + bytes := exactLiteralBytes(t, wire) + require.Len(t, bytes, bits/8) + require.Equal(t, int64(-125), int64(binary.LittleEndian.Uint64(bytes))) + if bits == 128 { + require.Equal(t, ^uint64(0), binary.LittleEndian.Uint64(bytes[8:])) + } + } + typ := planpb.Type{Id: int32(types.T_decimal256), Width: 39, NotNullable: true} + for _, negative := range []bool{false, true} { + text := "340282366920938463463374607431768211457" // 2^128 + 1 + if negative { + text = "-" + text + } + call := &planpb.Function{Func: &planpb.ObjectRef{ObjName: "cast", Obj: int64(function.CAST) << 32}, Args: []*planpb.Expr{ + {Typ: planpb.Type{Id: int32(types.T_varchar), NotNullable: true}, Expr: &planpb.Expr_Lit{Lit: &planpb.Literal{Value: &planpb.Literal_Sval{Sval: text}, DecimalLiteralRequiresV82: true}}}, + {Typ: typ, Expr: &planpb.Expr_T{T: &planpb.TargetType{}}}, + }} + wire, handled, err := e.exactNumericLiteralCast(&planpb.Expr{Typ: typ}, call) + require.True(t, handled) + require.NoError(t, err) + bytes := exactLiteralBytes(t, wire) + require.Len(t, bytes, 32) + want := [4]uint64{1, 0, 1, 0} + if negative { + want = [4]uint64{^uint64(0), ^uint64(0), ^uint64(1), ^uint64(0)} + } + for i := range want { + require.Equal(t, want[i], binary.LittleEndian.Uint64(bytes[8*i:])) + } + call.Args[0].GetLit().DecimalLiteralRequiresV82 = false + _, handled, err = e.exactNumericLiteralCast(&planpb.Expr{Typ: typ}, call) + require.False(t, handled, "ordinary string casts cannot become numeric literals") + require.NoError(t, err) + } +} + +func TestExactEmbeddedTPCHInventory(t *testing.T) { + mock := planbuilder.NewMockOptimizer(false, newPlanTestProcess(t)) + for number := 1; number <= 22; number++ { + t.Run(fmt.Sprintf("q%d", number), func(t *testing.T) { + query := exactTPCHQuery(t, mock, number) + candidate, err := ExportEmbeddedMO(query, NewEmbeddedExportProfile(31)) + require.NoError(t, err, "signatures: %v", numericFunctionSignatures(query)) + bindings := make(map[int32]EmbeddedReadBinding) + for ordinal, read := range candidate.Reads() { + bindings[read.NodeID] = EmbeddedReadBinding{BindingID: uint64(ordinal + 1), Source: EmbeddedReadMO} + } + wire, err := candidate.BuildEmbedded(bindings) + require.NoError(t, err) + var plan spb.Plan + require.NoError(t, proto.Unmarshal(wire, &plan)) + require.Equal(t, query.Headings, plan.Relations[len(plan.Relations)-1].GetRoot().Names) + }) + } +} + +func exactTPCHQuery(t *testing.T, mock *planbuilder.MockOptimizer, number int) *planpb.Query { + t.Helper() + sql, err := os.ReadFile(filepath.Join("..", "tpch", fmt.Sprintf("q%d.sql", number))) + require.NoError(t, err) + statements, err := parsers.Parse(t.Context(), dialect.MYSQL, string(sql), 1) + require.NoError(t, err) + query, err := mock.Optimize(statements[0]) + require.NoError(t, err) + for _, node := range query.Nodes { + if node != nil && node.TableDef != nil && node.ObjRef != nil { + node.TableDef.DbId, node.TableDef.TblId, node.ObjRef.Obj = 7, 42, 42 + } + } + return query +} + +func TestExactScalarAndAggregateFamiliesUseBoundOverloads(t *testing.T) { + for _, oid := range []types.T{types.T_decimal64, types.T_decimal128, types.T_decimal256} { + for _, name := range []string{"+", "-", "*", "/", "div", "mod", "unary_minus", "=", "!=", "<", "<=", ">", ">="} { + t.Run(fmt.Sprintf("%s/%s", oid, name), func(t *testing.T) { + n := 2 + if name == "unary_minus" { + n = 1 + } + inputTypes := make([]types.Type, n) + for i := range inputTypes { + inputTypes[i] = types.New(oid, 15, 2) + } + bound, err := function.GetFunctionByName(t.Context(), name, inputTypes) + require.NoError(t, err) + if casts, needed := bound.ShouldDoImplicitTypeCast(); needed { + inputTypes = casts + } + args := make([]*planpb.Expr, len(inputTypes)) + for i, input := range inputTypes { + args[i] = &planpb.Expr{Typ: planpb.Type{Id: int32(input.Oid), Width: input.Width, Scale: input.Scale, NotNullable: true}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: int32(i)}}} + } + out := bound.GetReturnType() + result := &planpb.Expr{Typ: planpb.Type{Id: int32(out.Oid), Width: out.Width, Scale: out.Scale, NotNullable: function.DeduceNotNullable(bound.GetEncodedOverloadID(), args)}} + call := &planpb.Function{Func: &planpb.ObjectRef{ObjName: name, Obj: bound.GetEncodedOverloadID()}, Args: args} + e := exporter{profile: NewEmbeddedExportProfile(31)} + wire, err := e.exactScalarExpr(result, call, []int{n}) + require.NoError(t, err) + id, _ := function.DecodeOverloadID(call.Func.Obj) + require.Equal(t, e.function("mo_decimal_"+exactScalarName(id)), wire.GetScalarFunction().FunctionReference) + call.Func.Obj++ // another overload cannot inherit the semantic declaration + _, err = e.exactScalarExpr(result, call, []int{n}) + require.True(t, IsNotEligible(err)) + }) + } + for _, name := range []string{"sum", "avg", "min", "max"} { + input := types.New(oid, 15, 2) + if oid == types.T_decimal256 && name == "sum" { + input.Width = 53 + } + bound, err := function.GetFunctionByName(t.Context(), name, []types.Type{input}) + require.NoError(t, err) + args := []*planpb.Expr{{Typ: planpb.Type{Id: int32(oid), Width: input.Width, Scale: 2, NotNullable: true}}} + out := bound.GetReturnType() + output := planpb.Type{Id: int32(out.Oid), Width: out.Width, Scale: out.Scale} + e := exporter{profile: NewEmbeddedExportProfile(31)} + ok, err := e.hasExactSemanticCapability(semanticAggregate, name, &planpb.ObjectRef{ObjName: name, Obj: bound.GetEncodedOverloadID()}, args, &output) + require.NoError(t, err) + require.True(t, ok, "%s/%s", oid, name) + } + } +} + +func TestExactAggregateDeclinesUnsupportedPhysicalNarrowing(t *testing.T) { + input := types.New(types.T_decimal256, 15, 2) + bound, err := function.GetFunctionByName(t.Context(), "sum", []types.Type{input}) + require.NoError(t, err) + output := bound.GetReturnType() + require.Equal(t, types.T_decimal128, output.Oid) + e := exporter{profile: NewEmbeddedExportProfile(31)} + ok, err := e.hasExactSemanticCapability(semanticAggregate, "sum", &planpb.ObjectRef{ObjName: "sum", Obj: bound.GetEncodedOverloadID()}, []*planpb.Expr{{Typ: planpb.Type{Id: int32(input.Oid), Width: input.Width, Scale: input.Scale}}}, &planpb.Type{Id: int32(output.Oid), Width: output.Width, Scale: output.Scale}) + require.NoError(t, err) + require.False(t, ok, "retain the MO result and decline; do not widen it to fit a native signature") +} + +func TestExactIntegerCastRequiresDomainOrProvenLiteral(t *testing.T) { + target := planpb.Type{Id: int32(types.T_decimal64), Width: 15, Scale: 2, NotNullable: true} + argument := &planpb.Expr{Typ: planpb.Type{Id: int32(types.T_int64), NotNullable: true}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{}}} + call := &planpb.Function{Func: &planpb.ObjectRef{ObjName: "cast", Obj: int64(function.CAST) << 32}, Args: []*planpb.Expr{argument, {Typ: target, Expr: &planpb.Expr_T{T: &planpb.TargetType{}}}}} + e := exporter{profile: NewEmbeddedExportProfile(31)} + _, err := e.exactScalarExpr(&planpb.Expr{Typ: target}, call, []int{1}) + require.True(t, IsNotEligible(err), "a Decimal64 target cannot contain the entire BIGINT domain") + argument.Expr = &planpb.Expr_Lit{Lit: &planpb.Literal{Value: &planpb.Literal_I64Val{I64Val: 12}}} + wire, err := e.exactScalarExpr(&planpb.Expr{Typ: target}, call, []int{1}) + require.NoError(t, err) + require.Equal(t, uint64(1200), binary.LittleEndian.Uint64(exactLiteralBytes(t, wire))) + target.Id, target.Width = int32(types.T_decimal128), 38 + call.Args[1].Typ = target + argument.Expr = &planpb.Expr_Col{Col: &planpb.ColRef{}} + _, err = e.exactScalarExpr(&planpb.Expr{Typ: target}, call, []int{1}) + require.NoError(t, err) + for _, mode := range []int64{1, 2, 3, 4} { + call.Func.Obj = int64(function.CAST)<<32 | mode + _, err = e.exactScalarExpr(&planpb.Expr{Typ: target}, call, []int{1}) + require.True(t, IsNotEligible(err), "only normal checked casts belong to v1") + } +} + +func TestExactDivisionRetainsBoundStatementResult(t *testing.T) { + for _, increment := range []int32{0, 4, 10, 30} { + inputs := []types.Type{types.New(types.T_decimal128, 10, 2), types.New(types.T_decimal128, 10, 2)} + bound, err := function.GetFunctionByName(function.WithDivPrecisionIncrement(context.Background(), increment), "/", inputs) + require.NoError(t, err) + if casts, needed := bound.ShouldDoImplicitTypeCast(); needed { + inputs = casts + } + args := make([]*planpb.Expr, 2) + for i, input := range inputs { + args[i] = &planpb.Expr{Typ: planpb.Type{Id: int32(input.Oid), Width: input.Width, Scale: input.Scale, NotNullable: true}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: int32(i)}}} + } + out := bound.GetReturnType() + result := &planpb.Expr{Typ: planpb.Type{Id: int32(out.Oid), Width: out.Width, Scale: out.Scale, NotNullable: function.DeduceNotNullable(bound.GetEncodedOverloadID(), args)}} + e := exporter{profile: NewEmbeddedExportProfile(31)} + wire, err := e.exactScalarExpr(result, &planpb.Function{Func: &planpb.ObjectRef{ObjName: "/", Obj: bound.GetEncodedOverloadID()}, Args: args}, []int{2}) + require.NoError(t, err) + user := wire.GetScalarFunction().OutputType.GetUserDefined() + require.Equal(t, int64(out.Width), user.TypeParameters[1].GetInteger()) + require.Equal(t, int64(out.Scale), user.TypeParameters[2].GetInteger()) + } +} + +func TestEmbeddedIntegerArithmeticDeclinesUncheckedConsumer(t *testing.T) { + for _, tc := range []struct{ mo, wire string }{{"+", "add"}, {"-", "subtract"}, {"*", "multiply"}} { + t.Run(tc.wire, func(t *testing.T) { + bound, err := function.GetFunctionByName(t.Context(), tc.mo, []types.Type{types.T_int64.ToType(), types.T_int64.ToType()}) + require.NoError(t, err) + args := []*planpb.Expr{{Typ: planpb.Type{Id: int32(types.T_int64)}}, {Typ: planpb.Type{Id: int32(types.T_int64)}}} + ret := bound.GetReturnType() + out := planpb.Type{Id: int32(ret.Oid), Width: ret.Width, Scale: ret.Scale, NotNullable: function.DeduceNotNullable(bound.GetEncodedOverloadID(), args)} + ref := &planpb.ObjectRef{ObjName: tc.mo, Obj: bound.GetEncodedOverloadID()} + for _, capabilities := range []uint64{11, 31} { + e := exporter{embeddedMO: true, profile: NewEmbeddedExportProfile(capabilities)} + ok, err := e.hasSemanticCapability(semanticScalar, tc.wire, ref, args, &out) + require.NoError(t, err) + require.False(t, ok, "embedded cuDF integer arithmetic does not preserve MO overflow") + } + flight := exporter{} + ok, err := flight.hasSemanticCapability(semanticScalar, tc.wire, ref, args, &out) + require.NoError(t, err) + require.True(t, ok, "retain the separate Flight admission contract") + }) + } +} + +func TestExactCASEPreservesBoundNullableConditionDescriptor(t *testing.T) { + for _, required := range []bool{false, true} { + t.Run(fmt.Sprint(required), func(t *testing.T) { + decimal := types.New(types.T_decimal64, 15, 2) + bound, err := function.GetFunctionByName(t.Context(), "case", []types.Type{types.T_bool.ToType(), decimal, decimal}) + require.NoError(t, err) + args := []*planpb.Expr{ + {Typ: planpb.Type{Id: int32(types.T_bool), NotNullable: required}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: 0}}}, + {Typ: planpb.Type{Id: int32(types.T_decimal64), Width: 15, Scale: 2, NotNullable: true}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: 1}}}, + {Typ: planpb.Type{Id: int32(types.T_decimal64), Width: 15, Scale: 2, NotNullable: true}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: 2}}}, + } + result := &planpb.Expr{Typ: planpb.Type{Id: int32(types.T_decimal64), Width: 15, Scale: 2, NotNullable: function.DeduceNotNullable(bound.GetEncodedOverloadID(), args)}} + require.Equal(t, required, result.Typ.NotNullable) + e := exporter{embeddedMO: true, profile: NewEmbeddedExportProfile(31)} + wire, err := e.exactScalarExpr(result, &planpb.Function{Func: &planpb.ObjectRef{ObjName: "case", Obj: bound.GetEncodedOverloadID()}, Args: args}, []int{3}) + require.NoError(t, err) + annotation := wire.GetScalarFunction() + require.NotNil(t, annotation, "IfThen cannot carry MO's bound descriptor") + require.Equal(t, e.functions["mo_decimal_cast"], annotation.FunctionReference) + declared := annotation.OutputType.GetUserDefined() + require.Equal(t, []int64{64, 15, 2}, []int64{declared.TypeParameters[0].GetInteger(), declared.TypeParameters[1].GetInteger(), declared.TypeParameters[2].GetInteger()}) + require.Equal(t, required, declared.Nullability == spb.Type_NULLABILITY_REQUIRED) + require.NotNil(t, annotation.Arguments[0].GetValue().GetIfThen()) + }) + } +} diff --git a/pkg/sql/plan/substrait/numeric_semantics.go b/pkg/sql/plan/substrait/numeric_semantics.go new file mode 100644 index 0000000000000..066fa3f31ddce --- /dev/null +++ b/pkg/sql/plan/substrait/numeric_semantics.go @@ -0,0 +1,374 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package substrait + +import ( + "context" + "encoding/binary" + + "github.com/matrixorigin/matrixone/pkg/common/moerr" + "github.com/matrixorigin/matrixone/pkg/container/types" + planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/plan/function" + spb "github.com/substrait-io/substrait-protobuf/go/substraitpb" +) + +func decimalSignature(args []*planpb.Expr, result *planpb.Type) bool { + if exactDecimalBits(result) != 0 { + return true + } + for _, arg := range args { + if arg != nil && exactDecimalBits(&arg.Typ) != 0 { + return true + } + } + return false +} + +func exactScalarName(id int32) string { + switch id { + case function.PLUS: + return "add" + case function.MINUS: + return "subtract" + case function.MULTI: + return "multiply" + case function.DIV: + return "divide" + case function.INTEGER_DIV: + return "integer_divide" + case function.MOD: + return "modulo" + case function.UNARY_MINUS: + return "negate" + case function.CAST: + return "cast" + case function.EQUAL: + return "equal" + case function.NOT_EQUAL: + return "not_equal" + case function.LESS_THAN: + return "less" + case function.LESS_EQUAL: + return "less_equal" + case function.GREAT_THAN: + return "greater" + case function.GREAT_EQUAL: + return "greater_equal" + case function.CASE: + return "if_then" + case function.COALESCE: + return "coalesce" + case function.ISNULL: + return "is_null" + case function.ISNOTNULL: + return "is_not_null" + default: + return "" + } +} + +func (e *exporter) hasSemanticCapability(kind semanticCapabilityKind, name string, ref *planpb.ObjectRef, args []*planpb.Expr, out *planpb.Type) (bool, error) { + if e.embeddedMO && kind == semanticScalar && !decimalSignature(args, out) && + (name == "add" || name == "subtract" || name == "multiply") { + // The embedded engine's ordinary integer operators use cuDF arithmetic + // without MO's checked overflow. The legacy registry label is not proof + // that this consumer preserves errors; decline before starting readers. + return false, nil + } + if e.profile.exactDecimalV1 && (decimalSignature(args, out) || name == "coalesce") { + return e.hasExactSemanticCapability(kind, name, ref, args, out) + } + return hasSemanticCapability(kind, name, ref, args, out) +} + +func sameDecimalDescriptor(a, b *planpb.Type) bool { + return exactDecimalBits(a) != 0 && exactDecimalBits(a) == exactDecimalBits(b) && a.Width == b.Width && a.Scale == b.Scale +} + +func (e *exporter) hasExactSemanticCapability(kind semanticCapabilityKind, name string, ref *planpb.ObjectRef, args []*planpb.Expr, out *planpb.Type) (bool, error) { + if ref == nil || out == nil || uint64(ref.Obj)&function.Distinct != 0 { + return false, nil + } + id, overload := function.DecodeOverloadID(ref.Obj) + if _, exists := function.GetFunctionByIdWithoutError(ref.Obj); !exists { + return false, nil + } + if _, err := e.profile.substraitType(out); err != nil { + return false, err + } + inputTypes := make([]types.Type, len(args)) + for i, arg := range args { + if arg == nil { + return false, nil + } + if _, err := e.profile.substraitType(&arg.Typ); err != nil { + return false, err + } + inputTypes[i] = types.New(types.T(arg.Typ.Id), arg.Typ.Width, arg.Typ.Scale) + } + if kind == semanticAggregate { + if len(args) != 1 { + return false, nil + } + switch id { + case function.SUM, function.AVG: + if (id == function.SUM && name != "sum") || (id == function.AVG && name != "avg") || + exactDecimalBits(out) < 128 || exactDecimalBits(out) < exactDecimalBits(&args[0].Typ) || out.Width > 65 || + (exactDecimalBits(&args[0].Typ) == 0 && types.T(args[0].Typ.Id) != types.T_int64) || + (id == function.SUM && out.Scale != args[0].Typ.Scale) || + (id == function.AVG && out.Scale < args[0].Typ.Scale) { + return false, nil + } + case function.MIN, function.MAX: + if (id == function.MIN && name != "min") || (id == function.MAX && name != "max") || !sameDecimalDescriptor(&args[0].Typ, out) || out.Width > 65 { + return false, nil + } + case function.COUNT: + if name != "count" || types.T(out.Id) != types.T_int64 { + return false, nil + } + default: + return false, nil + } + } else { + if name == "" || exactScalarName(id) != name { + return false, nil + } + switch id { + case function.CAST: + if overload != 0 || len(args) != 2 || args[1].GetT() == nil || exactDecimalBits(out) == 0 || !sameDecimalDescriptor(&args[1].Typ, out) { + return false, nil + } + if exactDecimalBits(&args[0].Typ) == 0 { + digits := int32(0) + switch types.T(args[0].Typ.Id) { + case types.T_int8: + digits = 3 + case types.T_int16: + digits = 5 + case types.T_int32: + digits = 10 + case types.T_int64: + digits = 19 + } + // A valid decimal domain with this many integer digits contains + // both signed endpoints at every scale, not just sampled values. + if digits == 0 || out.Width-out.Scale < digits { + return false, nil + } + } + case function.CASE: + if len(args) < 3 || len(args)%2 != 1 || !tpchCaseArgs(args, out) { + return false, nil + } + case function.COALESCE: + if len(args) == 0 { + return false, nil + } + if exactDecimalBits(out) == 0 { + switch types.T(out.Id) { + case types.T_bool, types.T_int8, types.T_int16, types.T_int32, types.T_int64: + default: + return false, nil + } + } + for _, arg := range args { + if arg.Typ.Id != out.Id || arg.Typ.Width != out.Width || arg.Typ.Scale != out.Scale { + return false, nil + } + } + case function.ISNULL, function.ISNOTNULL: + if len(args) != 1 || types.T(out.Id) != types.T_bool { + return false, nil + } + case function.UNARY_MINUS: + if len(args) != 1 || exactDecimalBits(&args[0].Typ) == 0 || exactDecimalBits(out) < exactDecimalBits(&args[0].Typ) || out.Width < args[0].Typ.Width || out.Scale != args[0].Typ.Scale { + return false, nil + } + default: + if len(args) != 2 || exactDecimalBits(&args[0].Typ) == 0 || exactDecimalBits(&args[1].Typ) == 0 { + return false, nil + } + if id == function.INTEGER_DIV { + if types.T(out.Id) != types.T_int64 { + return false, nil + } + } else if id == function.EQUAL || id == function.NOT_EQUAL || id == function.LESS_THAN || id == function.LESS_EQUAL || id == function.GREAT_THAN || id == function.GREAT_EQUAL { + if types.T(out.Id) != types.T_bool { + return false, nil + } + } else if exactDecimalBits(out) == 0 || out.Width > 65 { + return false, nil + } + if exactDecimalBits(out) != 0 { + if exactDecimalBits(out) < max(exactDecimalBits(&args[0].Typ), exactDecimalBits(&args[1].Typ)) { + return false, nil + } + switch id { + case function.PLUS, function.MINUS, function.MOD: + if out.Scale != max(args[0].Typ.Scale, args[1].Typ.Scale) { + return false, nil + } + case function.MULTI: + if exactDecimalBits(out) < 128 || out.Scale != min(args[0].Typ.Scale+args[1].Typ.Scale, max(int32(12), args[0].Typ.Scale, args[1].Typ.Scale)) { + return false, nil + } + case function.DIV: + if exactDecimalBits(out) < 128 || out.Scale > 30 { + return false, nil + } + } + } + } + } + resolved, err := function.GetFunctionByName(context.Background(), ref.ObjName, inputTypes) + if err != nil || resolved.GetEncodedOverloadID() != ref.Obj { + return false, nil + } + // Arithmetic results were bound before MO inserted physical working casts. + // Re-deriving precision from those casts would change valid MO results, and + // re-deriving division would lose the bound session increment. Validate the + // v1 domain above and retain that declared descriptor verbatim instead. + if kind == semanticAggregate || exactDecimalBits(out) == 0 || id == function.CAST || id == function.CASE || id == function.COALESCE { + result := resolved.GetReturnType() + if int32(result.Oid) != out.Id || result.Width != out.Width || result.Scale != out.Scale { + return false, nil + } + } + if kind == semanticAggregate && aggregateCanReturnNullOnEmpty(id) && !out.NotNullable { + return true, nil + } + return function.DeduceNotNullable(ref.Obj, args) == out.NotNullable || semanticNotNullable(ref.Obj, args) == out.NotNullable, nil +} + +func (e *exporter) exactScalarExpr(result *planpb.Expr, call *planpb.Function, inputs []int) (*spb.Expression, error) { + id, _ := function.DecodeOverloadID(call.Func.Obj) + name := exactScalarName(id) + if id == function.CAST { + if literal, handled, err := e.exactNumericLiteralCast(result, call); handled { + return literal, err + } + } + supported, err := e.hasExactSemanticCapability(semanticScalar, name, call.Func, call.Args, &result.Typ) + if err != nil { + return nil, err + } + if name == "" || !supported { + return nil, notEligiblef(EligibilityExpression, "exact %s overload %d has no declared Sirius semantic equivalence (result %d,%d,%d,required=%t)", name, call.Func.Obj, result.Typ.Id, result.Typ.Width, result.Typ.Scale, result.Typ.NotNullable) + } + if id == function.CASE { + branch, err := e.caseExpr(result, call, inputs) + if err != nil { + return nil, err + } + // IfThen carries no declared output type. Native inference from only + // the value arms loses MO's nullable-condition result annotation. + return e.scalar("mo_decimal_cast", &result.Typ, branch), nil + } + arguments := call.Args + if id == function.CAST { + arguments = arguments[:1] + } + args := make([]*spb.Expression, len(arguments)) + for i, argument := range arguments { + args[i], err = e.expr(argument, inputs) + if err != nil { + return nil, err + } + } + if id != function.COALESCE && id != function.ISNULL && id != function.ISNOTNULL { + name = "mo_decimal_" + name + } + return e.scalar(name, &result.Typ, args...), nil +} + +func (e *exporter) exactNumericLiteralCast(result *planpb.Expr, call *planpb.Function) (*spb.Expression, bool, error) { + id, overload := function.DecodeOverloadID(call.Func.Obj) + if id != function.CAST || overload != 0 || len(call.Args) != 2 || call.Args[0] == nil || call.Args[1] == nil || call.Args[1].GetT() == nil || !sameDecimalDescriptor(&call.Args[1].Typ, &result.Typ) { + return nil, false, nil + } + source := call.Args[0].GetLit() + if source == nil || source.IsBin || source.IsSerialized { + return nil, false, nil + } + declared, err := e.substraitType(&result.Typ) + if err != nil { + return nil, true, err + } + if source.Isnull { + literal, err := e.literal(source, &result.Typ) + return literal, true, err + } + var coefficient types.Decimal256 + if value, ok := source.Value.(*planpb.Literal_Sval); ok { + if !source.DecimalLiteralRequiresV82 || types.T(call.Args[0].Typ.Id) != types.T_varchar || len(value.Sval) > MaxPlanBytes { + return nil, false, nil + } + coefficient, err = types.ParseDecimal256(value.Sval, result.Typ.Width, result.Typ.Scale) + if err != nil { + return nil, true, moerr.NewInvalidInputNoCtx("invalid MO exact numeric literal") + } + } else { + var value int64 + var bits int + switch literal := source.Value.(type) { + case *planpb.Literal_I8Val: + if types.T(call.Args[0].Typ.Id) == types.T_int8 { + value, bits = int64(literal.I8Val), 8 + } + case *planpb.Literal_I16Val: + if types.T(call.Args[0].Typ.Id) == types.T_int16 { + value, bits = int64(literal.I16Val), 16 + } + case *planpb.Literal_I32Val: + if types.T(call.Args[0].Typ.Id) == types.T_int32 { + value, bits = int64(literal.I32Val), 32 + } + case *planpb.Literal_I64Val: + if types.T(call.Args[0].Typ.Id) == types.T_int64 { + value, bits = literal.I64Val, 64 + } + } + if bits == 0 { + return nil, false, nil + } + if bits < 64 && (value < -(int64(1)<<(bits-1)) || value >= int64(1)<<(bits-1)) { + return nil, true, moerr.NewInternalErrorNoCtx("substrait: integer literal carrier mismatch") + } + coefficient.B0_63 = uint64(value) + if value < 0 { + coefficient.B64_127, coefficient.B128_191, coefficient.B192_255 = ^uint64(0), ^uint64(0), ^uint64(0) + } + coefficient, err = coefficient.Scale(result.Typ.Scale) + if err != nil { + return nil, false, nil + } + limit, err := (types.Decimal256{B0_63: 1}).Scale(result.Typ.Width) + if err != nil || (!coefficient.Sign() && !coefficient.Less(limit)) || (coefficient.Sign() && !limit.Minus().Less(coefficient)) { + return nil, false, nil + } + // A proven constant becomes literal data, not a runtime cast whose + // target would have to contain the entire signed source domain. + } + data := make([]byte, exactDecimalBits(&result.Typ)/8) + for i, word := range []uint64{coefficient.B0_63, coefficient.B64_127, coefficient.B128_191, coefficient.B192_255} { + if i*8 >= len(data) { + break + } + binary.LittleEndian.PutUint64(data[i*8:], word) + } + return exactCoefficientLiteral(data, declared, !result.Typ.NotNullable), true, nil +} diff --git a/pkg/sql/plan/substrait/substrait_test.go b/pkg/sql/plan/substrait/substrait_test.go index ee77414f7de33..996bf13056bf8 100644 --- a/pkg/sql/plan/substrait/substrait_test.go +++ b/pkg/sql/plan/substrait/substrait_test.go @@ -68,7 +68,7 @@ func TestCollationMetadataSubstraitBoundaries(t *testing.T) { require.ErrorContains(t, err, "unsupported collation metadata") q = embeddedProjectedScanQuery() - embedded, err := ExportEmbeddedMO(q) + embedded, err := ExportEmbeddedMO(q, EmbeddedExportProfile{}) require.NoError(t, err) reads, err := embedded.EmbeddedMOReads() require.NoError(t, err) @@ -78,7 +78,7 @@ func TestCollationMetadataSubstraitBoundaries(t *testing.T) { require.NoError(t, err) require.NotEmpty(t, wire) mutate(q.Nodes[0].TableDef) - _, err = ExportEmbeddedMO(q) + _, err = ExportEmbeddedMO(q, EmbeddedExportProfile{}) require.ErrorContains(t, err, "unsupported collation metadata") reads, err = embedded.EmbeddedMOReads() require.ErrorContains(t, err, "unsupported collation metadata") diff --git a/pkg/sql/plan/substrait/tpch_test.go b/pkg/sql/plan/substrait/tpch_test.go index bd0bf73834e64..16266b4fc149c 100644 --- a/pkg/sql/plan/substrait/tpch_test.go +++ b/pkg/sql/plan/substrait/tpch_test.go @@ -21,9 +21,12 @@ import ( "path/filepath" "testing" + "github.com/matrixorigin/matrixone/pkg/container/types" + planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/sql/parsers" "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect" planbuilder "github.com/matrixorigin/matrixone/pkg/sql/plan" + "github.com/matrixorigin/matrixone/pkg/sql/plan/function" "github.com/stretchr/testify/require" spb "github.com/substrait-io/substrait-protobuf/go/substraitpb" "google.golang.org/protobuf/proto" @@ -129,6 +132,49 @@ func TestExportExtractSemanticBoundary(t *testing.T) { require.True(t, ok) require.Equal(t, EligibilityExpression, reason) } + embedded, embeddedErr := ExportEmbeddedMO(query, NewEmbeddedExportProfile(31)) + if tc.eligible && tc.expression != "extract(quarter from l_shipdate)" { + require.NoError(t, embeddedErr) + require.NotNil(t, embedded) + } else { + require.Nil(t, embedded) + require.True(t, IsNotEligible(embeddedErr), "%v", embeddedErr) + } }) } } + +func TestEmbeddedDATEExtractGuardsZeroDate(t *testing.T) { + for _, unit := range []string{"year", "month", "day"} { + for _, required := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/%t", unit, required), func(t *testing.T) { + args := []*planpb.Expr{ + {Typ: planpb.Type{Id: int32(types.T_varchar), NotNullable: true}, Expr: &planpb.Expr_Lit{Lit: &planpb.Literal{Value: &planpb.Literal_Sval{Sval: unit}}}}, + {Typ: planpb.Type{Id: int32(types.T_date), NotNullable: required}, Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: 0}}}, + } + bound, err := function.GetFunctionByName(t.Context(), "extract", []types.Type{types.T_varchar.ToType(), types.T_date.ToType()}) + require.NoError(t, err) + result := &planpb.Expr{Typ: planpb.Type{Id: int32(types.T_int64), NotNullable: function.DeduceNotNullable(bound.GetEncodedOverloadID(), args)}} + call := &planpb.Function{Func: &planpb.ObjectRef{ObjName: "extract", Obj: bound.GetEncodedOverloadID()}, Args: args} + e := exporter{embeddedMO: true, profile: NewEmbeddedExportProfile(31)} + wire, err := e.extractExpr(result, call, []int{1}) + require.NoError(t, err) + guard := wire.GetIfThen() + require.NotNil(t, guard) + require.Len(t, guard.Ifs, 1) + equality := guard.Ifs[0].If.GetScalarFunction() + require.Equal(t, e.functions["equal"], equality.FunctionReference) + require.Equal(t, types.ZeroDate.DaysSinceUnixEpoch(), equality.Arguments[1].GetValue().GetLiteral().GetDate()) + require.Equal(t, int64(0), guard.Ifs[0].Then.GetLiteral().GetI64()) + require.NotNil(t, guard.Else.GetCast().Type.GetI64()) + component := guard.Else.GetCast().Input.GetCast() + require.NotNil(t, component.Type.GetI32()) + require.Equal(t, e.functions[unit], component.Input.GetScalarFunction().FunctionReference) + flight := exporter{} + legacy, err := flight.extractExpr(result, call, []int{1}) + require.NoError(t, err) + require.NotNil(t, legacy.GetScalarFunction(), "Flight keeps the existing extract encoding") + }) + } + } +} diff --git a/pkg/tests/sqlintegration/sirius_embedded_test.go b/pkg/tests/sqlintegration/sirius_embedded_test.go index 5b18385299e5a..b0f742b4ddb98 100644 --- a/pkg/tests/sqlintegration/sirius_embedded_test.go +++ b/pkg/tests/sqlintegration/sirius_embedded_test.go @@ -22,9 +22,11 @@ import ( "fmt" "os" "path/filepath" + "strings" "testing" "time" + "github.com/go-sql-driver/mysql" "github.com/matrixorigin/matrixone/pkg/cnservice" "github.com/matrixorigin/matrixone/pkg/embed" "github.com/matrixorigin/matrixone/pkg/pb/metadata" @@ -52,7 +54,9 @@ func TestEmbeddedSiriusPublicMOReader(t *testing.T) { db, err := sql.Open("mysql", fmt.Sprintf("dump:111@tcp(127.0.0.1:%d)/", cn.GetServiceConfig().CN.Frontend.Port)) require.NoError(t, err) t.Cleanup(func() { require.NoError(t, db.Close()) }) - ctx, cancel := context.WithTimeout(t.Context(), 60*time.Second) + // USE, transactions and prepared division must share one public session. + db.SetMaxOpenConns(1) + ctx, cancel := context.WithTimeout(t.Context(), 120*time.Second) t.Cleanup(cancel) for _, statement := range []string{ "create database sirius_reader_contract", "use sirius_reader_contract", @@ -80,12 +84,178 @@ func TestEmbeddedSiriusPublicMOReader(t *testing.T) { compare("select n,s from t order by n") execSQLRequire(t, ctx, db, "rollback") compare("select n,s from t order by n") + t.Run("exact decimal values and metadata", func(t *testing.T) { + execSQLRequire(t, ctx, db, "create table d (id int not null primary key, a decimal(15,2), b decimal(38,6), c decimal(65,8))") + execSQLRequire(t, ctx, db, "insert into d values (1,9007199254740.99,9007199254740993.000001,340282366920938463463374607431768211457.12345678),(2,-12.50,-25.000001,-340282366920938463463374607431768211457.12345678),(3,NULL,NULL,NULL),(4,0,0,0)") + for _, statement := range []string{ + "select id,a,b,c from d order by id", + "select id,a+a,b-b,c+c,-c,a*b,c/c,c div c,c mod c from d order by id", + "select id,a is null,a is not null,b is null,b is not null,c is null,c is not null from d order by id", + "select id,case when c>0 then c else -c end,coalesce(c,0),c is null,c is not null from d order by id", + "select sum(a),avg(a),min(a),max(a),sum(b),avg(b),min(b),max(b),sum(c),avg(c),min(c),max(c),count(c) from d", + "select sum(c),avg(c),min(c),max(c),count(c) from d where id=3", + "select sum(c),avg(c),min(c),max(c),count(c) from d where id<0", + "select x.id,x.c,y.c from d x join d y on x.c=y.c order by x.c,x.id,y.id", + "select c,count(c) from d group by c order by c", + "select x.id,x.c,y.v from d x left join (select sum(c) as v from d where id<0) y on true order by x.id", + "select x.id,x.c,y.v from d x left join (select sum(c) as v from d where id=3) y on true order by x.id", + } { + t.Run(statement, func(t *testing.T) { + names, values := readSiriusPublicRows(t, ctx, db, statement) + gpuNames, gpuValues := readSiriusPublicRows(t, ctx, db, "/*+ SIDECAR GPU */ "+statement) + require.Equal(t, names, gpuNames) + require.Equal(t, values, gpuValues) + var queryID string + require.NoError(t, db.QueryRowContext(ctx, "select last_query_id()").Scan(&queryID)) + t.Logf("embedded decimal query_id=%s", queryID) + }) + } + }) + t.Run("CASE bound nullable condition with required arms", func(t *testing.T) { + execSQLRequire(t, ctx, db, "create table case_required (id int not null primary key, a decimal(15,2) not null, b decimal(15,2) not null)") + execSQLRequire(t, ctx, db, "insert into case_required values (1,1.25,2.50),(2,0,2.50)") + statement := "select id,case when a/a>0 then a else b end as chosen from case_required order by id" + names, values := readSiriusPublicRows(t, ctx, db, statement) + gpuNames, gpuValues := readSiriusPublicRows(t, ctx, db, "/*+ SIDECAR GPU */ "+statement) + require.Equal(t, names, gpuNames) + require.Equal(t, values, gpuValues) + require.Contains(t, gpuNames[1], "nullable=true/true") + require.Contains(t, gpuNames[1], "decimal=15,2/true") + require.Equal(t, [][]sql.NullString{{{String: "1", Valid: true}, {String: "1.25", Valid: true}}, {{String: "2", Valid: true}, {String: "2.50", Valid: true}}}, gpuValues) + }) + t.Run("DATE extraction values and metadata", func(t *testing.T) { + var original string + require.NoError(t, db.QueryRowContext(ctx, "select @@session.sql_mode").Scan(&original)) + t.Cleanup(func() { + cleanup, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + execSQLRequire(t, cleanup, db, "set session sql_mode='"+strings.ReplaceAll(original, "'", "''")+"'") + }) + execSQLRequire(t, ctx, db, "set session sql_mode=''") + execSQLRequire(t, ctx, db, "create table date_input (id int not null primary key, v date)") + execSQLRequire(t, ctx, db, "insert into date_input values (1,'0001-01-01'),(2,'2000-02-29'),(3,'9999-12-31'),(4,NULL),(5,'0000-00-00')") + compare("select id,extract(year from v),extract(month from v),extract(day from v) from date_input order by id") + names, values := readSiriusPublicRows(t, ctx, db, "/*+ SIDECAR GPU */ select extract(year from v),extract(month from v),extract(day from v) from date_input where id=5") + require.Len(t, names, 3) + require.Equal(t, [][]sql.NullString{{{String: "0", Valid: true}, {String: "0", Valid: true}, {String: "0", Valid: true}}}, values) + }) + t.Run("unchecked integer arithmetic declines before execution", func(t *testing.T) { + for _, statement := range []string{ + "select max(c),count(*)+9223372036854775807 from d", + "select max(c),-9223372036854775807-count(*) from d", + "select max(c),count(*)*9223372036854775807 from d", + } { + t.Run(statement, func(t *testing.T) { + var control, declined *mysql.MySQLError + require.ErrorAs(t, readSiriusPublicError(t, ctx, db, statement), &control) + require.Equal(t, uint16(1690), control.Number) + require.Equal(t, "22003", string(control.SQLState[:])) + require.ErrorAs(t, readSiriusPublicError(t, ctx, db, "/*+ SIDECAR GPU */ "+statement), &declined) + require.Equal(t, uint16(1105), declined.Number) + require.Contains(t, declined.Message, "not eligible") + }) + } + // The same runtime still accepts integral aggregates and checked decimal + // arithmetic after declining the unsupported ordinary integer operators. + compare("select max(c),count(*) from d") + }) + t.Run("numeric errors and healthy reuse", func(t *testing.T) { + execSQLRequire(t, ctx, db, "create table numeric_errors (id int not null primary key, v decimal(38,0), c decimal(65,0))") + execSQLRequire(t, ctx, db, "insert into numeric_errors values (1,99999999999999999999999999999999999999,99999999999999999999999999999999999999999999999999999999999999999),(2,99999999999999999999999999999999999999,99999999999999999999999999999999999999999999999999999999999999999)") + for _, tc := range []struct { + statement string + number uint16 + state string + }{ + {"select v*v from numeric_errors", 1690, "22003"}, + {"select sum(v*v) from numeric_errors", 1690, "22003"}, + {"select sum(c) from numeric_errors", 20301, "HY000"}, + } { + t.Run(tc.statement, func(t *testing.T) { + for _, prefix := range []string{"", "/*+ SIDECAR GPU */ "} { + err := readSiriusPublicError(t, ctx, db, prefix+tc.statement) + var mysqlErr *mysql.MySQLError + require.ErrorAs(t, err, &mysqlErr) + require.Equal(t, tc.number, mysqlErr.Number) + require.Equal(t, tc.state, string(mysqlErr.SQLState[:])) + } + names, values := readSiriusPublicRows(t, ctx, db, "select sum(v) from numeric_errors") + gpuNames, gpuValues := readSiriusPublicRows(t, ctx, db, "/*+ SIDECAR GPU */ select sum(v) from numeric_errors") + require.Equal(t, names, gpuNames) + require.Equal(t, values, gpuValues) + }) + } + for _, statement := range []string{ + "select case when id<0 then v*v else v end from numeric_errors order by id", + "select coalesce(v,v*v) from numeric_errors order by id", + "select v/0 from numeric_errors order by id", + } { + t.Run(statement, func(t *testing.T) { + names, values := readSiriusPublicRows(t, ctx, db, statement) + gpuNames, gpuValues := readSiriusPublicRows(t, ctx, db, "/*+ SIDECAR GPU */ "+statement) + require.Equal(t, names, gpuNames) + require.Equal(t, values, gpuValues) + }) + } + }) + t.Run("prepared division rebinds values and metadata", func(t *testing.T) { + execSQLRequire(t, ctx, db, "create table division_input (a decimal(10,2), b decimal(10,2))") + execSQLRequire(t, ctx, db, "insert into division_input values (1,3)") + var original int + require.NoError(t, db.QueryRowContext(ctx, "select @@session.div_precision_increment").Scan(&original)) + t.Cleanup(func() { + cleanup, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + execSQLRequire(t, cleanup, db, fmt.Sprintf("set session div_precision_increment=%d", original)) + }) + control, err := db.PrepareContext(ctx, "select a/b as q from division_input") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, control.Close()) }) + embedded, err := db.PrepareContext(ctx, "/*+ SIDECAR GPU */ select a/b as q from division_input") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, embedded.Close()) }) + for _, increment := range []int{0, 4, 10, 30, 4} { + t.Run(fmt.Sprint(increment), func(t *testing.T) { + execSQLRequire(t, ctx, db, fmt.Sprintf("set session div_precision_increment=%d", increment)) + rows, err := control.QueryContext(ctx) + require.NoError(t, err) + names, values := readSiriusPublicResult(t, rows) + rows, err = embedded.QueryContext(ctx) + require.NoError(t, err) + gpuNames, gpuValues := readSiriusPublicResult(t, rows) + require.Equal(t, names, gpuNames) + require.Equal(t, values, gpuValues) + scale := min(2+increment, 30) + require.Equal(t, [][]sql.NullString{{{String: "0." + strings.Repeat("3", scale), Valid: true}}}, gpuValues) + require.Contains(t, gpuNames[0], fmt.Sprintf("decimal=%d,%d/true", 12+increment, scale)) + }) + } + }) +} + +func readSiriusPublicError(t *testing.T, ctx context.Context, db *sql.DB, statement string) error { + t.Helper() + rows, err := db.QueryContext(ctx, statement) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + // Numeric errors may arrive after the result metadata. Drain to the + // terminal packet without retaining any rows from a failing query. + } + return rows.Err() } func readSiriusPublicRows(t *testing.T, ctx context.Context, db *sql.DB, statement string) ([]string, [][]sql.NullString) { t.Helper() rows, err := db.QueryContext(ctx, statement) require.NoError(t, err) + return readSiriusPublicResult(t, rows) +} + +func readSiriusPublicResult(t *testing.T, rows *sql.Rows) ([]string, [][]sql.NullString) { + t.Helper() defer rows.Close() names, err := rows.Columns() require.NoError(t, err) diff --git a/third_party/sirius b/third_party/sirius index e2e2f08f9fd1e..908ffc75a58b3 160000 --- a/third_party/sirius +++ b/third_party/sirius @@ -1 +1 @@ -Subproject commit e2e2f08f9fd1eaa1253297493df0ab11a6078664 +Subproject commit 908ffc75a58b3be496b2172795e416326416e7ec