add check constraints

pkg/bootstrap/versions/upgrade_strategy.go

@@ -15,7 +15,10 @@
 package versions
 
 import (
+	"encoding/json"
 	"fmt"
+	"strconv"
+	"strings"
 
 	"go.uber.org/zap"
 
@@ -135,8 +138,13 @@ type UpgradeEntry struct {
 	// return true if the system is already in the final state and does not need to be upgraded,
 	// otherwise return false
 	CheckFunc func(txn executor.TxnExecutor, accountId uint32) (bool, error)
-	PreSql    string
-	PostSql   string
+	// RequiredProtocolVersion delays an upgrade entry until every service in the
+	// deployment reports the protocol understood by the persisted metadata it
+	// installs. The check is performed only when the entry still needs work, so
+	// an already-completed upgrade remains idempotent during a rolling restart.
+	RequiredProtocolVersion int64
+	PreSql                  string
+	PostSql                 string
 }
 
 // Upgrade entity execution upgrade entrance
@@ -151,33 +159,93 @@ func (u *UpgradeEntry) Upgrade(txn executor.TxnExecutor, accountId uint32) error
 
 	if exist {
 		return nil
-	} else {
-		// 1. First, judge whether there is prefix sql
-		if u.PreSql != "" {
-			res, err := txn.Exec(u.PreSql, statementOption)
-			if err != nil {
-				getLogger(txn.Txn().TxnOptions().CN).Error("execute upgrade entry pre-sql error", zap.Error(err), zap.String("upgrade entry", u.String()))
-				return err
-			}
-			res.Close()
+	}
+	if u.RequiredProtocolVersion > 0 {
+		if txn == nil {
+			return moerr.NewNotSupportedNoCtxf(
+				"upgrade %s requires protocol version %d, transaction is unavailable",
+				u.TableName, u.RequiredProtocolVersion)
+		}
+		if err := checkCommonProtocolVersion(txn, u.RequiredProtocolVersion); err != nil {
+			return err
+		}
+	}
+
+	// 1. First, judge whether there is prefix sql
+	if u.PreSql != "" {
+		res, err := txn.Exec(u.PreSql, statementOption)
+		if err != nil {
+			getLogger(txn.Txn().TxnOptions().CN).Error("execute upgrade entry pre-sql error", zap.Error(err), zap.String("upgrade entry", u.String()))
+			return err
 		}
+		res.Close()
+	}
+
+	// 2. Second, Execute upgrade sql
+	res, err := txn.Exec(u.UpgSql, statementOption)
+	if err != nil {
+		getLogger(txn.Txn().TxnOptions().CN).Error("execute upgrade entry sql error", zap.Error(err), zap.String("upgrade entry", u.String()))
+		return err
+	}
+	res.Close()
 
-		// 2. Second, Execute upgrade sql
-		res, err := txn.Exec(u.UpgSql, statementOption)
+	// 2. Third, after the upgrade is completed, judge whether there is post-sql
+	if u.PostSql != "" {
+		res, err = txn.Exec(u.PostSql, statementOption)
 		if err != nil {
-			getLogger(txn.Txn().TxnOptions().CN).Error("execute upgrade entry sql error", zap.Error(err), zap.String("upgrade entry", u.String()))
+			getLogger(txn.Txn().TxnOptions().CN).Error("execute upgrade entry post-sql error", zap.Error(err), zap.String("upgrade entry", u.String()))
 			return err
 		}
 		res.Close()
+	}
+	return nil
+}
+
+// checkCommonProtocolVersion asks every CN for its rollout value. A local
+// runtime can already be at the new version while a same-version, lower-offset
+// CN is still serving traffic, so checking only the upgrader would publish an
+// information_schema view that the older CN cannot plan.
+func checkCommonProtocolVersion(txn executor.TxnExecutor, required int64) error {
+	res, err := txn.Exec(
+		"SELECT mo_ctl('cn', 'GetProtocolVersion', '')",
+		executor.StatementOption{},
+	)
+	if err != nil {
+		return err
+	}
+	defer res.Close()
+
+	var encoded string
+	res.ReadRows(func(rows int, cols []*vector.Vector) bool {
+		if rows == 0 || len(cols) == 0 || cols[0].IsNull(0) {
+			return false
+		}
+		encoded = cols[0].GetStringAt(0)
+		return false
+	})
+	if encoded == "" {
+		return moerr.NewNotSupportedNoCtxf(
+			"upgrade requires all CNs to support protocol version %d: no protocol response", required)
+	}
 
-		// 2. Third, after the upgrade is completed, judge whether there is post-sql
-		if u.PostSql != "" {
-			res, err = txn.Exec(u.PostSql, statementOption)
-			if err != nil {
-				getLogger(txn.Txn().TxnOptions().CN).Error("execute upgrade entry post-sql error", zap.Error(err), zap.String("upgrade entry", u.String()))
-				return err
-			}
-			res.Close()
+	var envelope struct {
+		Result string `json:"result"`
+	}
+	if err := json.Unmarshal([]byte(encoded), &envelope); err != nil {
+		return moerr.NewNotSupportedNoCtxf(
+			"upgrade requires all CNs to support protocol version %d: invalid protocol response", required)
+	}
+	for _, nodeVersion := range strings.Split(envelope.Result, ",") {
+		nodeVersion = strings.TrimSpace(nodeVersion)
+		separator := strings.LastIndexByte(nodeVersion, ':')
+		if separator < 0 {
+			return moerr.NewNotSupportedNoCtxf(
+				"upgrade requires all CNs to support protocol version %d: invalid node version %q", required, nodeVersion)
+		}
+		version, parseErr := strconv.ParseInt(strings.TrimSpace(nodeVersion[separator+1:]), 10, 64)
+		if parseErr != nil || version < required {
+			return moerr.NewNotSupportedNoCtxf(
+				"upgrade requires all CNs to support protocol version %d: node %q is at version %d", required, nodeVersion[:separator], version)
 		}
 	}
 	return nil

pkg/bootstrap/versions/upgrade_strategy_test.go

@@ -15,12 +15,70 @@
 package versions
 
 import (
+	"errors"
 	"testing"
 
 	"github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/defines"
+	"github.com/matrixorigin/matrixone/pkg/util/executor"
 	"github.com/stretchr/testify/require"
 )
 
+func TestCheckCommonProtocolVersion(t *testing.T) {
+	for _, test := range []struct {
+		name    string
+		value   string
+		wantErr bool
+	}{
+		{name: "all CNs ready", value: `{"method":"GETPROTOCOLVERSION","result":"cn-a:14,cn-b:15"}`},
+		{name: "older CN blocks", value: `{"method":"GETPROTOCOLVERSION","result":"cn-a:14,cn-b:13"}`, wantErr: true},
+		{name: "malformed response blocks", value: `{"method":"GETPROTOCOLVERSION","result":"cn-a"}`, wantErr: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			txn := executor.NewMemTxnExecutor(func(sql string) (executor.Result, error) {
+				require.Equal(t, "SELECT mo_ctl('cn', 'GetProtocolVersion', '')", sql)
+				return newProtocolResult(t, test.value), nil
+			}, nil)
+			err := checkCommonProtocolVersion(txn, defines.MORPCVersion14)
+			if test.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+
+	txn := executor.NewMemTxnExecutor(func(string) (executor.Result, error) {
+		return executor.Result{}, errors.New("query unavailable")
+	}, nil)
+	require.ErrorContains(t, checkCommonProtocolVersion(txn, defines.MORPCVersion14), "query unavailable")
+}
+
+func TestUpgradeEntryWaitsForCommonProtocol(t *testing.T) {
+	upgraded := false
+	entry := UpgradeEntry{
+		TableName:               "CHECK_CONSTRAINTS",
+		RequiredProtocolVersion: defines.MORPCVersion14,
+		CheckFunc: func(executor.TxnExecutor, uint32) (bool, error) {
+			return false, nil
+		},
+		UpgSql: "CREATE VIEW information_schema.CHECK_CONSTRAINTS AS ...",
+	}
+	txn := executor.NewMemTxnExecutor(func(sql string) (executor.Result, error) {
+		if sql == "SELECT mo_ctl('cn', 'GetProtocolVersion', '')" {
+			return newProtocolResult(t, `{"method":"GETPROTOCOLVERSION","result":"cn-a:14,cn-b:13"}`), nil
+		}
+		upgraded = true
+		return executor.Result{}, nil
+	}, nil)
+
+	err := entry.Upgrade(txn, 0)
+	require.ErrorContains(t, err, "node")
+	require.False(t, upgraded)
+}
+
 func TestUpgradeStatementOption(t *testing.T) {
 	for _, test := range []struct {
 		name      string
@@ -44,3 +102,13 @@ func TestUpgradeStatementOption(t *testing.T) {
 		})
 	}
 }
+
+func newProtocolResult(t *testing.T, value string) executor.Result {
+	t.Helper()
+	mp := mpool.MustNewZeroNoFixed()
+	t.Cleanup(func() { mpool.DeleteMPool(mp) })
+	result := executor.NewMemResult([]types.Type{types.T_varchar.ToType()}, mp)
+	result.NewBatchWithRowCount(1)
+	require.NoError(t, executor.AppendStringRows(result, 0, []string{value}))
+	return result.GetResult()
+}

pkg/bootstrap/versions/v4_0_6/tenant_upgrade_list.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions"
 	"github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/defines"
 	"github.com/matrixorigin/matrixone/pkg/sql/mongodb"
 	"github.com/matrixorigin/matrixone/pkg/util/executor"
 	"github.com/matrixorigin/matrixone/pkg/util/sysview"
@@ -33,6 +34,8 @@ var tenantUpgEntries = []versions.UpgradeEntry{
 	upgradeInformationSchemaReferentialConstraints(),
 	populateInformationSchemaCharacterSets(),
 	upgradeInformationSchemaColumns(),
+	upgradeInformationSchemaCheckConstraints(),
+	upgradeInformationSchemaTableConstraints(),
 }
 
 // Keep this as a separate upgrade entry so tenants that already completed
@@ -122,6 +125,34 @@ func upgradeInformationSchemaReferentialConstraints() versions.UpgradeEntry {
 	}
 }
 
+func upgradeInformationSchemaCheckConstraints() versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:                  sysview.InformationDBConst,
+		TableName:               "CHECK_CONSTRAINTS",
+		UpgType:                 versions.CREATE_VIEW,
+		UpgSql:                  sysview.InformationSchemaCheckConstraintsDDL,
+		RequiredProtocolVersion: defines.MORPCVersion16,
+		CheckFunc: checkViewDefinition("CHECK_CONSTRAINTS",
+			sysview.InformationSchemaCheckConstraintsDDL),
+		PreSql: fmt.Sprintf("DROP VIEW IF EXISTS %s.%s;",
+			sysview.InformationDBConst, "CHECK_CONSTRAINTS"),
+	}
+}
+
+func upgradeInformationSchemaTableConstraints() versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:                  sysview.InformationDBConst,
+		TableName:               "TABLE_CONSTRAINTS",
+		UpgType:                 versions.MODIFY_VIEW,
+		UpgSql:                  sysview.InformationSchemaTableConstraintsDDL,
+		RequiredProtocolVersion: defines.MORPCVersion16,
+		CheckFunc: checkViewDefinition("TABLE_CONSTRAINTS",
+			sysview.InformationSchemaTableConstraintsDDL),
+		PreSql: fmt.Sprintf("DROP VIEW IF EXISTS %s.%s;",
+			sysview.InformationDBConst, "TABLE_CONSTRAINTS"),
+	}
+}
+
 func checkViewDefinition(viewName, definition string) func(executor.TxnExecutor, uint32) (bool, error) {
 	return func(txn executor.TxnExecutor, accountID uint32) (bool, error) {
 		exists, viewDef, err := versions.CheckViewDefinition(txn, accountID, sysview.InformationDBConst, viewName)