feat(ivfflat): add INCLUDE column support for IVF indexes (#24168)

Add INCLUDE column support across IVF DDL, entries-table maintenance, planner rewrites, and ivf_search so covered vector queries can push include predicates and avoid base-table joins when mode=include.

Keep mode=post and mode=pre on their original single-round path, switch mode=include fallback from cumulative bucket expansion to non-overlapping bucket slices, reset per-input search round state, and collapse EXPLAIN ANALYZE background ivf_search plans by default while keeping capped verbose expansion.

Add parser, planner, ALTER, explain, unit, and distributed SQL coverage for the new INCLUDE behavior.

Approved by: @ouyuanning, @LeftHandCold, @heni02, @cpegeric, @aunjgr, @fengttt, @XuPeng-SH, @gouhongshen

Author: iamlinjunhong

pkg/bootstrap/versions/v4_0_0/upgrade_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/prashantv/gostub"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions"
 	"github.com/matrixorigin/matrixone/pkg/catalog"
@@ -48,6 +49,10 @@ func mockTenantUpgrade(t *testing.T) {
 }
 
 func Test_Upgrade(t *testing.T) {
+	originalTenantUpgEntries := append([]versions.UpgradeEntry(nil), tenantUpgEntries...)
+	defer func() {
+		tenantUpgEntries = originalTenantUpgEntries
+	}()
 	mockTenantUpgrade(t)
 
 	sid := ""
@@ -85,11 +90,11 @@ func Test_Upgrade(t *testing.T) {
 }
 
 func Test_versionHandle_HandleClusterUpgrade(t *testing.T) {
-	originalEntries := clusterUpgEntries
-	clusterUpgEntries = nil
+	originalClusterUpgEntries := append([]versions.UpgradeEntry(nil), clusterUpgEntries...)
 	defer func() {
-		clusterUpgEntries = originalEntries
+		clusterUpgEntries = originalClusterUpgEntries
 	}()
+	clusterUpgEntries = []versions.UpgradeEntry{}
 
 	v := &versionHandle{
 		metadata: versions.Version{
@@ -109,6 +114,16 @@ func Test_versionHandle_HandleClusterUpgrade(t *testing.T) {
 	assert.Nil(t, err)
 }
 
+func TestUpgradeDefersMoIndexesIncludeColumn(t *testing.T) {
+	for _, entries := range [][]versions.UpgradeEntry{clusterUpgEntries, tenantUpgEntries} {
+		for _, entry := range entries {
+			require.False(t,
+				entry.Schema == catalog.MO_CATALOG && entry.TableName == catalog.MO_INDEXES && entry.UpgType == versions.ADD_COLUMN,
+				"mo_indexes schema changes require a staged release after all writers use explicit column lists")
+		}
+	}
+}
+
 func Test_upg_create_mo_task_sql_task_check_exists(t *testing.T) {
 	stubs := gostub.Stub(&versions.CheckTableDefinition, func(txn executor.TxnExecutor, accountId uint32, schema string, table string) (bool, error) {
 		assert.Equal(t, uint32(0), accountId)

pkg/bootstrap/versions/v4_0_1/upgrade_test.go

@@ -21,8 +21,10 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/prashantv/gostub"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"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/common/runtime"
 	mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test"
@@ -94,6 +96,16 @@ func Test_versionHandle_HandleClusterUpgrade(t *testing.T) {
 	assert.Nil(t, err)
 }
 
+func TestUpgradeDefersMoIndexesIncludeColumn(t *testing.T) {
+	for _, entries := range [][]versions.UpgradeEntry{clusterUpgEntries, tenantUpgEntries} {
+		for _, entry := range entries {
+			require.False(t,
+				entry.Schema == catalog.MO_CATALOG && entry.TableName == catalog.MO_INDEXES && entry.UpgType == versions.ADD_COLUMN,
+				"mo_indexes schema changes require a staged release after all writers use explicit column lists")
+		}
+	}
+}
+
 func Test_upg_statistics_view_check_error(t *testing.T) {
 	stubs := gostub.Stub(&versions.CheckViewDefinition, func(txn executor.TxnExecutor, accountId uint32, schema string, viewName string) (bool, string, error) {
 		return true, "", moerr.NewInternalErrorNoCtx("return error")

pkg/catalog/secondary_index_utils.go

@@ -113,7 +113,10 @@ const (
 	IntermediateGraphDegree = "intermediate_graph_degree"
 	GraphDegree             = "graph_degree"
 	ITopkSize               = "itopk_size"
-	IncludedColumns         = "included_columns"
+	// IncludedColumns persists INCLUDE metadata inside algo_params. Consumers
+	// prefer plan.IndexDef.IncludedColumns when present and fall back to this key
+	// for catalog-loaded definitions.
+	IncludedColumns = "included_columns"
 
 	// Index-defining build params, settable as CREATE INDEX options (parsed by
 	// each plugin's ParamsFromTree). Written into flat algo_params only when
@@ -127,6 +130,49 @@ const (
 	IndexAlgoParamPrefixLengths = "prefix_lengths"
 )
 
+func ParseIncludeColumnsValue(raw string) ([]string, error) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return nil, nil
+	}
+
+	if strings.HasPrefix(raw, "[") {
+		var cols []string
+		if err := json.Unmarshal([]byte(raw), &cols); err == nil {
+			return cols, nil
+		}
+	}
+
+	parts := strings.Split(raw, ",")
+	cols := make([]string, 0, len(parts))
+	for _, part := range parts {
+		part = strings.TrimSpace(part)
+		if part == "" {
+			continue
+		}
+		cols = append(cols, part)
+	}
+	if len(cols) == 0 {
+		return nil, nil
+	}
+	return cols, nil
+}
+
+// MarshalIncludeColumnsValue encodes INCLUDE column names without losing
+// commas or leading/trailing whitespace that are valid inside quoted SQL
+// identifiers. ParseIncludeColumnsValue retains support for the historical
+// comma-separated representation when catalog metadata is reloaded.
+func MarshalIncludeColumnsValue(cols []string) (string, error) {
+	if len(cols) == 0 {
+		return "", nil
+	}
+	encoded, err := json.Marshal(cols)
+	if err != nil {
+		return "", err
+	}
+	return string(encoded), nil
+}
+
 /* 1. ToString Functions */
 
 // IndexParamsToStringList used by buildShowCreateTable and restoreDDL
@@ -223,19 +269,6 @@ func IndexParamsToStringList(indexParams string) (string, error) {
 	if val, ok := result[IndexAlgoParamQuantizerTrainLimit]; ok {
 		res += fmt.Sprintf(" %s = %s ", IndexAlgoParamQuantizerTrainLimit, val)
 	}
-
-	if val, ok := result[IncludedColumns]; ok && len(val) > 0 {
-		raw := strings.Split(val, ",")
-		parts := make([]string, 0, len(raw))
-		for _, p := range raw {
-			if p = strings.TrimSpace(p); p != "" {
-				parts = append(parts, p)
-			}
-		}
-		if len(parts) > 0 {
-			res += " INCLUDE (" + strings.Join(parts, ", ") + ") "
-		}
-	}
 	return res, nil
 }
 
@@ -425,7 +458,7 @@ func indexParamsToMap(def interface{}) (map[string]string, error) {
 		case tree.INDEX_TYPE_BTREE, tree.INDEX_TYPE_INVALID, tree.INDEX_TYPE_RTREE:
 			// do nothing
 		case tree.INDEX_TYPE_MASTER:
-			// do nothing
+		// do nothing
 		default:
 			// Vector algorithms (IVFFLAT / HNSW / CAGRA / IVFPQ) build their
 			// algo_params via the per-plugin plan hook BuildIndexParams; they

pkg/catalog/secondary_index_utils_test.go

@@ -59,7 +59,6 @@ func newPrefixKeyPart(colName string, length int) *tree.KeyPart {
 }
 
 func TestIsIndexAsync(t *testing.T) {
-
 	var (
 		json string
 		err  error
@@ -252,3 +251,56 @@ func TestIndexPrefixLengthsFromParams(t *testing.T) {
 	require.Nil(t, IndexPrefixLengthsFromParams("{bad json"))
 	require.Nil(t, IndexPrefixLengthsFromParams(`{"prefix_lengths":"t:0"}`))
 }
+
+func unresolvedName(name string) *tree.UnresolvedName {
+	return tree.NewUnresolvedName(tree.NewCStr(name, 1))
+}
+
+func TestIndexParamsToJsonString_RejectsPluginIvfFlatPath(t *testing.T) {
+	idx := tree.NewIndex(
+		false,
+		[]*tree.KeyPart{tree.NewKeyPart(unresolvedName("embedding"), -1, tree.DefaultDirection, nil)},
+		"idx1",
+		tree.INDEX_TYPE_IVFFLAT,
+		&tree.IndexOption{
+			AlgoParamList:         10,
+			AlgoParamVectorOpType: "vector_l2_ops",
+			IncludeColumns: []*tree.UnresolvedName{
+				unresolvedName("title"),
+				unresolvedName("category"),
+			},
+		},
+	)
+
+	params, err := IndexParamsToJsonString(idx)
+	require.Error(t, err)
+	require.Empty(t, params)
+	require.Contains(t, err.Error(), "invalid index alogorithm type")
+}
+
+func TestIndexParamsToStringList_DoesNotRenderIncludeColumns(t *testing.T) {
+	paramList, err := IndexParamsToStringList(`{"lists":"10","op_type":"vector_l2_ops","included_columns":"title,category"}`)
+	require.NoError(t, err)
+	require.Contains(t, paramList, "lists = 10")
+	require.Contains(t, paramList, "op_type 'vector_l2_ops'")
+	require.NotContains(t, paramList, "INCLUDE")
+	require.NotContains(t, paramList, "title")
+}
+
+func TestParseIncludeColumnsValue_BackwardCompatible(t *testing.T) {
+	encoded, err := MarshalIncludeColumnsValue([]string{"title, label", " rank "})
+	require.NoError(t, err)
+	require.Equal(t, `["title, label"," rank "]`, encoded)
+
+	cols, err := ParseIncludeColumnsValue(encoded)
+	require.NoError(t, err)
+	require.Equal(t, []string{"title, label", " rank "}, cols)
+
+	cols, err = ParseIncludeColumnsValue("title,category")
+	require.NoError(t, err)
+	require.Equal(t, []string{"title", "category"}, cols)
+
+	cols, err = ParseIncludeColumnsValue("[legacy,other")
+	require.NoError(t, err)
+	require.Equal(t, []string{"[legacy", "other"}, cols)
+}

pkg/catalog/types.go

@@ -406,6 +406,7 @@ const (
 	SystemSI_IVFFLAT_TblCol_Entries_id      = "__mo_index_centroid_fk_id"
 	SystemSI_IVFFLAT_TblCol_Entries_pk      = IndexTablePrimaryColName
 	SystemSI_IVFFLAT_TblCol_Entries_entry   = "__mo_index_centroid_fk_entry"
+	SystemSI_IVFFLAT_IncludeColPrefix       = "__mo_index_include_"
 
 	/************ 3. FULLTEXT Index **************/
 

pkg/frontend/predefined.go

@@ -457,6 +457,7 @@ var (
 			ordinal_position    int unsigned  not null,
 			options             text,
 			index_table_name    varchar(5000),
+			included_columns    text,
 			primary key(id, column_name)
 		)`, catalog.MO_CATALOG, catalog.MO_INDEXES)
 

pkg/indexplugin/plan/alter_column.go

@@ -0,0 +1,37 @@
+// 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 plan
+
+import planpb "github.com/matrixorigin/matrixone/pkg/pb/plan"
+
+// AlterColumnHooks is an optional plan-layer hook for algorithm-specific
+// metadata maintenance during ALTER TABLE column mutations.
+type AlterColumnHooks interface {
+	HandleAlterDropColumn(tableDef *planpb.TableDef, indexDef *planpb.IndexDef, colName string) (bool, error)
+	HandleAlterRenameColumn(tableDef *planpb.TableDef, oldColName, newColName string) ([]string, error)
+}
+
+type RenameColumnRebuildHook interface {
+	RenameColumnRequiresIndexRebuild(tableDef *planpb.TableDef, indexDef *planpb.IndexDef, oldColName string) (bool, error)
+}
+
+type UpdateColumnRewriteHook interface {
+	UpdateColumnRequiresIndexRewrite(tableDef *planpb.TableDef, indexDef *planpb.IndexDef, colName string) (bool, error)
+}
+
+var (
+	IncludedColumnAffected       func(indexDef *planpb.IndexDef, colName string) (bool, error)
+	RenameIncludedColumnsForAlgo func(tableDef *planpb.TableDef, algo, oldColName, newColName string) ([]string, error)
+)

pkg/iscp/index_sqlwriter.go

@@ -81,6 +81,7 @@ type IvfflatSqlWriter struct {
 	entries_tbl   string
 	meta_tbl      string
 	ivfparam      vectorindex.IvfParam
+	includeCols   []string
 }
 
 // Hnsw Sql Writer.  Use the vectorindex.VectorIndeXCdc JSON format
@@ -589,17 +590,36 @@ func NewIvfflatSqlWriter(algo string, jobID JobID, info *ConsumerInfo, tabledef
 	w.pkType = &types.Type{Oid: types.T(typ.Id), Width: typ.Width, Scale: typ.Scale}
 
 	nparts := len(w.indexdef[0].Parts)
-	w.partsPos = make([]int32, nparts)
-	w.partsType = make([]*types.Type, nparts)
+	includeCols, err := ivfflatIncludeColumnsFromIndexDefs(w.indexdef)
+	if err != nil {
+		return nil, err
+	}
+	w.includeCols = append(w.includeCols[:0], includeCols...)
+	w.partsPos = make([]int32, nparts+len(includeCols))
+	w.partsType = make([]*types.Type, nparts+len(includeCols))
 
 	for i, part := range w.indexdef[0].Parts {
 		w.partsPos[i] = tabledef.Name2ColIndex[part]
 		typ = tabledef.Cols[w.partsPos[i]].Typ
 		w.partsType[i] = &types.Type{Oid: types.T(typ.Id), Width: typ.Width, Scale: typ.Scale}
 	}
+	for i, includeCol := range includeCols {
+		pos, ok := tabledef.Name2ColIndex[includeCol]
+		if !ok {
+			resolvedCol := catalog.ResolveAlias(includeCol)
+			pos, ok = tabledef.Name2ColIndex[resolvedCol]
+		}
+		if !ok {
+			return nil, moerr.NewInternalErrorNoCtxf("ivfflat include column %q not found in source table", includeCol)
+		}
+		dst := nparts + i
+		w.partsPos[dst] = pos
+		typ = tabledef.Cols[pos].Typ
+		w.partsType[dst] = &types.Type{Oid: types.T(typ.Id), Width: typ.Width, Scale: typ.Scale}
+	}
 
-	w.srcPos = make([]int32, nparts+1)
-	w.srcType = make([]*types.Type, nparts+1)
+	w.srcPos = make([]int32, len(w.partsPos)+1)
+	w.srcType = make([]*types.Type, len(w.partsType)+1)
 
 	w.srcPos[0] = w.pkPos
 	w.srcType[0] = w.pkType
@@ -615,6 +635,36 @@ func NewIvfflatSqlWriter(algo string, jobID JobID, info *ConsumerInfo, tabledef
 	return w, nil
 }
 
+func ivfflatIncludeColumnsFromIndexDefs(indexdefs []*plan.IndexDef) ([]string, error) {
+	for _, idx := range indexdefs {
+		if idx != nil && len(idx.IncludedColumns) > 0 {
+			return idx.IncludedColumns, nil
+		}
+	}
+	for _, idx := range indexdefs {
+		if idx == nil || idx.IndexAlgoParams == "" {
+			continue
+		}
+		params, err := catalog.IndexParamsStringToMap(idx.IndexAlgoParams)
+		if err != nil {
+			return nil, err
+		}
+		raw := params[catalog.IncludedColumns]
+		if raw == "" {
+			raw = params["include_columns"]
+		}
+		if raw == "" {
+			continue
+		}
+		cols, err := catalog.ParseIncludeColumnsValue(raw)
+		if err != nil {
+			return nil, err
+		}
+		return cols, nil
+	}
+	return nil, nil
+}
+
 // REPLACE INTO __mo_index_secondary_0197786c-285f-70cb-9337-e484a3ff92c4(__mo_index_centroid_fk_version, __mo_index_centroid_fk_id, __mo_index_pri_col, __mo_index_centroid_fk_entry)
 // with centroid as (select * from __mo_index_secondary_0197786c-285f-70bb-b277-2cef56da590a where __mo_index_centroid_version = 0),
 // src as (select column_0 as id, cast(column_1 as vecf32(3)) as embed from (values row(2005,'[0.4532634, 0.7297859, 0.48885703]'), row(2009, '[0.68150306, 0.6950923, 0.16590895] ')))
@@ -708,11 +758,20 @@ func (w *IvfflatSqlWriter) toIvfflatUpsert(upsert bool) ([]byte, error) {
 		sql += fmt.Sprintf("REPLACE INTO %s ", sqlquote.QualifiedIdent(w.info.DBName, w.entries_tbl))
 	}
 
-	sql += fmt.Sprintf("(`%s`, `%s`, `%s`, `%s`) ",
+	targetCols := []string{
 		catalog.SystemSI_IVFFLAT_TblCol_Entries_version,
 		catalog.SystemSI_IVFFLAT_TblCol_Entries_id,
 		catalog.SystemSI_IVFFLAT_TblCol_Entries_pk,
-		catalog.SystemSI_IVFFLAT_TblCol_Entries_entry)
+		catalog.SystemSI_IVFFLAT_TblCol_Entries_entry,
+	}
+	for _, includeCol := range w.includeCols {
+		targetCols = append(targetCols, catalog.SystemSI_IVFFLAT_IncludeColPrefix+includeCol)
+	}
+	quotedTargetCols := make([]string, 0, len(targetCols))
+	for _, col := range targetCols {
+		quotedTargetCols = append(quotedTargetCols, sqlquote.Ident(col))
+	}
+	sql += fmt.Sprintf("(%s) ", strings.Join(quotedTargetCols, ", "))
 
 	versql := fmt.Sprintf("SELECT CAST(%s as BIGINT) FROM %s WHERE `%s` = 'version'", catalog.SystemSI_IVFFLAT_TblCol_Metadata_val,
 		sqlquote.QualifiedIdent(w.info.DBName, w.meta_tbl), catalog.SystemSI_IVFFLAT_TblCol_Metadata_key)