feat(sql): add experimental native materialized views with atomic full refresh

cmd/mo-service/materialized_view_faults.go

@@ -0,0 +1,36 @@
+// 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 matrixone_mv_faults
+
+package main
+
+import (
+	"context"
+	"os"
+
+	"github.com/matrixorigin/matrixone/pkg/util/fault"
+)
+
+// Only the explicitly instrumented native recovery binary accepts startup
+// faults. Production builds have no environment-controlled injection path.
+// Startup injection is needed because ordinary login cannot precede migration.
+func init() {
+	if point := os.Getenv("MO_MV_STARTUP_FAULT"); point != "" {
+		fault.Enable()
+		if result := fault.HandleFaultInject(context.Background(), fault.AddFault, point); result != "OK" {
+			panic(result)
+		}
+	}
+}

etc/launch-tae-compose/compose.yaml

@@ -42,7 +42,7 @@ services:
     depends_on:
       - tn
     ports:
-      - "6002"
+      - "6001"
       - "12345:12345"
     environment:
       - AWS_ACCESS_KEY_ID=minio
@@ -61,7 +61,7 @@ services:
     depends_on:
       - tn
     ports:
-      - "6002"
+      - "6001"
       - "22345:12345"
     profiles:
       - launch

optools/run_bvt_group.sh

@@ -70,3 +70,13 @@ printf '%s\n' "${included_scripts[@]}"
 
 cd "${tester_dir}"
 ./run.sh -n -g -o -p "${case_root}" -i "${include_list}" "${resource_args[@]}"
+
+# Native MV checks need two independently addressed CNs and typed SQL oracles.
+# Run once, after mo-tester's global restore/account fixtures have finished.
+if (( group_id == 0 )); then
+    repo_root=$(cd "${case_root}/../../.." && pwd)
+    if [[ -f "${repo_root}/test/materialized_view/main.go" ]]; then
+        bash "${repo_root}/optools/run_materialized_view_bvt.sh" \
+            "${repo_root}" "${tester_dir}/report/materialized_view"
+    fi
+fi

optools/run_materialized_view_bvt.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+# SQL contract checks reuse the existing multi-CN BVT deployment. Performance
+# and restart experiments remain separate, explicitly invoked validation modes.
+set -euo pipefail
+if [[ $# != 2 ]]; then
+    echo "Usage: $0 <repository-root> <report-directory>" >&2
+    exit 2
+fi
+repo_root=$(cd "$1" && pwd)
+mkdir -p "$2" "${repo_root}/coverage"
+report_dir=$(cd "$2" && pwd)
+coverage_dir=$(cd "${repo_root}/coverage" && pwd)
+# A rerun must not accept a previous success if the client cannot write JSON.
+rm -f "${report_dir}/consumers.json" "${report_dir}/functional.json" "${report_dir}/failure.json" "${report_dir}/multi-cn.json" "${report_dir}/native-probe.json"
+
+cn_dsn() {
+    local endpoint port
+    endpoint=$(docker compose -f "${repo_root}/etc/launch-tae-compose/compose.yaml" \
+        --profile launch-multi-cn port "$1" 6001)
+    port=${endpoint##*:}
+    [[ "${port}" =~ ^[0-9]+$ ]]
+    printf 'dump:111@tcp(127.0.0.1:%s)/?timeout=5s&readTimeout=330s&writeTimeout=330s' "${port}"
+}
+
+# Overrides support an existing local deployment without starting more servers.
+primary_dsn=${MV_BVT_DSN:-$(cn_dsn cn-0)}
+second_dsn=${MV_BVT_SECOND_DSN:-$(cn_dsn cn-1)}
+build_dir=$(mktemp -d)
+trap 'rm -rf "${build_dir}"' EXIT
+cd "${repo_root}"
+# Match mo-service's default -cover mode so covdata can combine the producer.
+CGO_ENABLED=0 go build -mod=readonly -cover -covermode=set \
+    -o "${build_dir}/mv-validation" ./test/materialized_view
+CGO_ENABLED=0 go build -mod=readonly -cover -covermode=set \
+    -o "${build_dir}/mv-native-probe" ./test/materialized_view/native_probe
+export GOCOVERDIR="${coverage_dir}"
+# The same JSON bridge drives the destructive native recovery suite. Exercise
+# its public stream/session/error contract on this existing deployment too;
+# its coverage counters join the unchanged UT/BVT coverage aggregation.
+python3 "${repo_root}/test/materialized_view/native_probe/check.py" \
+    "${build_dir}/mv-native-probe" "${primary_dsn}" > "${report_dir}/native-probe.json"
+"${build_dir}/mv-validation" -dsn "${primary_dsn}" -functional \
+    -output "${report_dir}/functional.json"
+"${build_dir}/mv-validation" -dsn "${primary_dsn}" -edge-check \
+    -output "${report_dir}/failure.json"
+"${build_dir}/mv-validation" -dsn "${primary_dsn}" -consumer-check -second-dsn "${second_dsn}" \
+    -output "${report_dir}/consumers.json"
+"${build_dir}/mv-validation" -dsn "${primary_dsn}" -second-dsn "${second_dsn}" \
+    -output "${report_dir}/multi-cn.json"
+# Fail closed if a mode exits without its nonempty, successful evidence report.
+python3 - "${report_dir}" <<'PY'
+import json
+import pathlib
+import sys
+
+probe = json.loads((pathlib.Path(sys.argv[1]) / "native-probe.json").read_text())
+if probe != {"runs": [{"passed_requests": 9, "process_exit": 0}] * 2}:
+    raise SystemExit("Incomplete materialized-view native SQL bridge checks")
+print("Materialized-view native SQL bridge: 2 x 9 requests passed")
+
+for name in ("functional", "failure", "consumers", "multi-cn"):
+    result = json.loads((pathlib.Path(sys.argv[1]) / (name + ".json")).read_text())
+    if result.get("error") or not result.get("passed_cases"):
+        raise SystemExit("Incomplete materialized-view BVT: " + name)
+    print("Materialized-view BVT:", name, len(result["passed_cases"]), "checks passed")
+PY

pkg/bootstrap/columns_upgrade_test.go

@@ -37,7 +37,8 @@ func TestColumnsUpgradeReachesExistingTenants(t *testing.T) {
 		current := versions.Version{Version: "4.0.16", VersionOffset: 1, State: versions.StateReady}
 		definition := sysview.InformationSchemaColumnsV109DDL()
 		tenantVersion := "4.0.16"
-		var queuedVersion, queuedRoute bool
+		var queuedVersion bool
+		var queuedRoutes []string
 		var writes []string
 		failCreate := false
 		injected := errors.New("injected COLUMNS create failure")
@@ -55,12 +56,11 @@ func TestColumnsUpgradeReachesExistingTenants(t *testing.T) {
 			case strings.HasPrefix(sql, "select version from mo_version"):
 				return newBootstrapStringResult("4.0.16"), nil
 			case strings.HasPrefix(sql, "insert into mo_version"):
-				require.Contains(t, sql, "'4.0.17', 1, 0")
+				require.Contains(t, sql, "'4.0.18', 5, 0")
 				queuedVersion = true
 			case strings.HasPrefix(sql, "insert into mo_upgrade"):
-				require.Contains(t, sql, "'4.0.16'")
-				require.Contains(t, sql, "'4.0.17'")
-				queuedRoute = true
+				require.Contains(t, sql, "'4.0.18'")
+				queuedRoutes = append(queuedRoutes, sql)
 			case strings.HasPrefix(sql, "select create_version from mo_account"):
 				return newBootstrapStringResult(tenantVersion), nil
 			case strings.HasPrefix(sql, "SELECT tbl.rel_createsql"):
@@ -80,6 +80,11 @@ func TestColumnsUpgradeReachesExistingTenants(t *testing.T) {
 					"tenant completion must follow the successful view refresh")
 				tenantVersion = "4.0.17"
 				return executor.Result{AffectedRows: 1}, nil
+			case sql == "update mo_account set create_version = '4.0.18' where account_id = 7":
+				require.Equal(t, "4.0.17", tenantVersion, "retain the upstream tenant repair hop")
+				require.Equal(t, sysview.InformationSchemaColumnsDDL, definition)
+				tenantVersion = "4.0.18"
+				return executor.Result{AffectedRows: 1}, nil
 			default:
 				return executor.Result{}, fmt.Errorf("unexpected upgrade SQL: %s", sql)
 			}
@@ -91,18 +96,22 @@ func TestColumnsUpgradeReachesExistingTenants(t *testing.T) {
 		final := s.getFinalVersionHandle().Metadata()
 		require.NoError(t, s.doCheckUpgrade(t.Context()))
 		require.True(t, queuedVersion)
-		require.True(t, queuedRoute, "a completed 4.0.16 cluster must queue the forward repair")
-		require.Equal(t, "4.0.17", final.Version)
-		require.Equal(t, "4.0.16", final.MinUpgradeVersion)
-		require.Equal(t, uint32(1), final.VersionOffset)
-		require.Equal(t, defines.MORPCVersion118, final.RequiredProtocolVersion)
+		require.Len(t, queuedRoutes, 2, "retain the tenant repair and later MV catalog hops")
+		require.Contains(t, queuedRoutes[0], "'4.0.16'")
+		require.Contains(t, queuedRoutes[0], "'4.0.17'")
+		require.Contains(t, queuedRoutes[1], "'4.0.17'")
+		require.Contains(t, queuedRoutes[1], "'4.0.18'")
+		require.Equal(t, "4.0.18", final.Version)
+		require.Equal(t, "4.0.17", final.MinUpgradeVersion)
+		require.Equal(t, uint32(5), final.VersionOffset)
+		require.Equal(t, defines.MORPCVersion122, final.RequiredProtocolVersion)
 
 		// Model a tenant created by an old CN after batch enumeration: the
 		// cluster route has finished, but its persisted account version is old.
 		current = final
 		current.State = versions.StateReady
 		failCreate = true
-		fetchTenant := func() (int32, string, error) { return 7, "4.0.17", nil }
+		fetchTenant := func() (int32, string, error) { return 7, "4.0.18", nil }
 		upgraded, err := s.MaybeUpgradeTenant(t.Context(), fetchTenant, nil)
 		require.ErrorIs(t, err, injected)
 		require.False(t, upgraded)
@@ -114,7 +123,7 @@ func TestColumnsUpgradeReachesExistingTenants(t *testing.T) {
 		require.NoError(t, err)
 		require.True(t, upgraded)
 		require.Equal(t, []string{"DROP VIEW IF EXISTS information_schema.COLUMNS;", sysview.InformationSchemaColumnsDDL}, writes)
-		require.Equal(t, "4.0.17", tenantVersion)
+		require.Equal(t, "4.0.18", tenantVersion)
 		require.Equal(t, sysview.InformationSchemaColumnsDDL, definition)
 		writes = nil
 		upgraded, err = s.MaybeUpgradeTenant(t.Context(), fetchTenant, nil)

pkg/bootstrap/service_statistics_upgrade_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_14"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_15"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_16"
+	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_17"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_6"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_7"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_8"
@@ -82,6 +83,7 @@ func TestDoCheckUpgradeQueuesStatisticsRefresh(t *testing.T) {
 		{name: "current_4.0.14", version: "4.0.14", offset: v4_0_14.Handler.Metadata().VersionOffset, upgrade: true},
 		{name: "current_4.0.15", version: "4.0.15", offset: v4_0_15.Handler.Metadata().VersionOffset, upgrade: true},
 		{name: "current_4.0.16", version: "4.0.16", offset: v4_0_16.Handler.Metadata().VersionOffset, upgrade: true},
+		{name: "current_4.0.17", version: "4.0.17", offset: v4_0_17.Handler.Metadata().VersionOffset, upgrade: true},
 		{name: "current", version: final.Version, offset: final.VersionOffset},
 	} {
 		t.Run(test.name, func(t *testing.T) {
@@ -117,6 +119,9 @@ func TestDoCheckUpgradeQueuesStatisticsRefresh(t *testing.T) {
 				require.NoError(t, b.doCheckUpgrade(context.Background()))
 				if test.upgrade {
 					hops := []versions.Version{final}
+					if versions.Compare(test.version, "4.0.17") < 0 {
+						hops = append([]versions.Version{v4_0_17.Handler.Metadata()}, hops...)
+					}
 					if versions.Compare(test.version, "4.0.16") < 0 {
 						hops = append([]versions.Version{v4_0_16.Handler.Metadata()}, hops...)
 					}

pkg/bootstrap/upgrade.go

@@ -36,6 +36,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_15"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_16"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_17"
+	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_18"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_2"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_3"
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions/v4_0_4"
@@ -81,6 +82,7 @@ func (s *service) initUpgrade() {
 	s.handles = append(s.handles, v4_0_15.Handler)
 	s.handles = append(s.handles, v4_0_16.Handler)
 	s.handles = append(s.handles, v4_0_17.Handler)
+	s.handles = append(s.handles, v4_0_18.Handler)
 }
 
 func (s *service) getFinalVersionHandle() VersionHandle {

pkg/bootstrap/versions/v4_0_18/upgrade.go

@@ -0,0 +1,84 @@
+// 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 v4_0_18
+
+import (
+	"context"
+	"strings"
+
+	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions"
+	"github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/defines"
+	"github.com/matrixorigin/matrixone/pkg/util/executor"
+	"github.com/matrixorigin/matrixone/pkg/util/fault"
+)
+
+var Handler = &versionHandle{
+	metadata: versions.Version{
+		Version:                 "4.0.18",
+		MinUpgradeVersion:       "4.0.17",
+		UpgradeCluster:          versions.Yes,
+		UpgradeTenant:           versions.No,
+		VersionOffset:           uint32(len(catalog.MaterializedViewCatalogDDLs) + 1),
+		RequiredProtocolVersion: defines.MORPCVersion122,
+	},
+}
+
+type versionHandle struct {
+	metadata versions.Version
+}
+
+func (v *versionHandle) Metadata() versions.Version {
+	return v.metadata
+}
+
+func (v *versionHandle) Prepare(_ context.Context, txn executor.TxnExecutor, _ bool) error {
+	txn.Use(catalog.MO_CATALOG)
+	return nil
+}
+
+func (v *versionHandle) HandleTenantUpgrade(_ context.Context, tenantID int32, txn executor.TxnExecutor) error {
+	return nil
+}
+
+func (v *versionHandle) HandleClusterUpgrade(ctx context.Context, tx executor.TxnExecutor) error {
+	// This version has no tenant-upgrade phase, so the framework's tenant
+	// protocol check does not run. Fence old CNs before installing MV catalogs.
+	if err := versions.CheckCommonProtocolVersion(tx, v.metadata.RequiredProtocolVersion); err != nil {
+		return err
+	}
+	for _, ddl := range catalog.MaterializedViewCatalogDDLs {
+		ddl = strings.Replace(ddl, "create cluster table ", "create cluster table if not exists ", 1)
+		result, err := tx.Exec(ddl, executor.StatementOption{}.WithAccountID(catalog.System_Account))
+		if err != nil {
+			return err
+		}
+		result.Close()
+		fault.TriggerFaultWithContext(ctx, "mv_catalog_upgrade_after_ddl")
+	}
+	// Format 1 lacks the captured division context. Pause legacy policies and
+	// fence queued work without guessing semantics or rewriting stored rows.
+	r, err := tx.Exec("update mo_catalog.mo_materialized_views set interval_seconds=0,next_due=null,epoch=epoch+1 where interval_seconds>0 and cast(json_extract(definition,'$.format_version') as int)=1", executor.StatementOption{}.WithAccountID(catalog.System_Account))
+	if err != nil {
+		return err
+	}
+	r.Close()
+	return nil
+}
+
+func (v *versionHandle) HandleCreateFrameworkDeps(_ executor.TxnExecutor) error {
+	return moerr.NewInternalErrorNoCtxf("Only v1.2.0 can initialize upgrade framework, current version is:%s", v.metadata.Version)
+}