feat(sirius): lower MO exact decimals for embedded GPU execution

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()),

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

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
+	}
+}

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))
+}

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
 }

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.

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:

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()) })

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)

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

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())