fix: skip insert payload fields absent from schema on querynode consumption (#51117) (#51137)

## What

Filter out insert payload columns whose field is absent from the current
collection schema when converting an insert message on the QueryNode
consumption path (`TransferInsertMsgToInsertRecord`), instead of
forwarding them into segcore where `SegmentGrowingImpl::Insert`
hard-asserts on unknown fields and panics the whole StreamingNode.

## Why (root cause, verified from Loki logs of the failing run)

1. A growing segment was created under schema v104 and received one
insert whose payload — written at WAL-append time under v104 — contains
field 158 (a field generation that was later dropped; by v112 it no
longer exists in the schema).
2. The StreamingNode died once for an unrelated reason. On every
subsequent restart, `WatchDmChannels` seeds the collection with the
**latest** schema (v112, without field 158 — coord meta is updated right
after the DDL's WAL append ack, so it always runs ahead of any replaying
consumer), and WAL replay starts from the channel checkpoint, which
predates the segment's data.
3. The replayed v104 insert still carries field 158; the growing segment
rebuilt with v112 has no such column and can never gain it. Replayed
schema events (v1→v103) are correctly skipped by the schema-version gate
(#50577), so the mismatch is structural, not a race.
4. `SegmentGrowingImpl::Insert` tolerates only the inverse staleness
(segment schema newer than payload → fill empty columns). An unknown
payload field hits `AssertInfo(insert_record_.is_data_exist(field_id),
...)` → SegcoreError → `delegator.ProcessInsert` panics → the checkpoint
never advances past the message → deterministic CrashLoopBackOff.

Dropped-field data is unqueryable by definition, so skipping those
columns is semantically lossless. On the live path the WAL-append
schema-version gate plus segment fencing on schema change guarantee the
payload is a subset of the current schema, making this filter a no-op
outside replay.

A WARN log records the skipped field IDs for observability.

## Notes

- Related latent issues intentionally out of scope (to be filed
separately): compaction can leave a flushed segment out of the watch
request's flushed/dropped exclusion lists (duplicate-data risk on
replay); `Reopen`/`FillAbsentFields` never create function-output
columns on pre-existing growing segments.

issue: #51117

---------

Signed-off-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

internal/querynodev2/pipeline/insert_node.go

@@ -46,12 +46,30 @@ type insertNode struct {
 }
 
 func (iNode *insertNode) addInsertData(insertDatas map[UniqueID]*delegator.InsertData, msg *InsertMsg, collection *Collection) {
-	insertRecord, err := storage.TransferInsertMsgToInsertRecord(collection.Schema(), msg)
+	schema := collection.Schema()
+	insertRecord, skippedFields, err := storage.TransferInsertMsgToInsertRecord(schema, msg)
 	if err != nil {
 		err = merr.Wrap(err, "failed to get primary keys")
 		mlog.Error(context.TODO(), err.Error(), mlog.Int64("collectionID", iNode.collectionID), mlog.String("channel", iNode.channel))
 		panic(err)
 	}
+	if len(skippedFields) > 0 {
+		// Attributing the skip to dropped fields is safe only because a SchemaChange
+		// message never shares a pack with inserts: the WAL adaptor emits every V1/V2
+		// message as its own pack, msgdispatcher never merges packs, and the DML
+		// micro-batcher refuses to merge non-Insert/Delete messages. If that
+		// pack-granularity invariant is ever relaxed, filtering against the current
+		// schema could silently drop fields that exist in a newer schema.
+		mlog.Warn(context.TODO(), "skip insert payload fields absent from current schema, fields are dropped since the message was written",
+			mlog.FieldCollectionID(iNode.collectionID),
+			mlog.FieldSegmentID(msg.SegmentID),
+			mlog.String("channel", iNode.channel),
+			mlog.Int32("schemaVersion", schema.GetVersion()),
+			mlog.Int64s("skippedFieldIDs", skippedFields))
+		metrics.QueryNodeSkippedInsertFieldCount.
+			WithLabelValues(paramtable.GetStringNodeID(), fmt.Sprint(iNode.collectionID)).
+			Add(float64(len(skippedFields)))
+	}
 	iData, ok := insertDatas[msg.SegmentID]
 	if !ok {
 		iData = &delegator.InsertData{
@@ -72,12 +90,12 @@ func (iNode *insertNode) addInsertData(insertDatas map[UniqueID]*delegator.Inser
 		iData.InsertRecord.NumRows += insertRecord.NumRows
 	}
 
-	if err := iNode.appendBM25Stats(iData, msg, collection.Schema()); err != nil {
+	if err := iNode.appendBM25Stats(iData, msg, schema); err != nil {
 		mlog.Error(context.TODO(), "failed to append BM25 stats from insert message", mlog.String("channel", iNode.channel), mlog.Err(err))
 		panic(err)
 	}
 
-	pks, err := segments.GetPrimaryKeys(msg, collection.Schema())
+	pks, err := segments.GetPrimaryKeys(msg, schema)
 	if err != nil {
 		mlog.Error(context.TODO(), "failed to get primary keys from insert message", mlog.Err(err))
 		panic(err)

internal/querynodev2/segments/retrieve_test.go

@@ -173,7 +173,7 @@ func (suite *RetrieveSuite) SetupTest() {
 
 	insertMsg, err := mock_segcore.GenInsertMsg(suite.collection.GetCCollection(), suite.partitionID, suite.growing.ID(), msgLength)
 	suite.Require().NoError(err)
-	insertRecord, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
+	insertRecord, _, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
 	suite.Require().NoError(err)
 	err = suite.growing.Insert(suite.ctx, insertMsg.RowIDs, insertMsg.Timestamps, insertRecord)
 	suite.Require().NoError(err)
@@ -366,7 +366,7 @@ func (suite *RetrieveSuite) TestRetrieveStreamWithFilterDoesNotPruneGrowing() {
 
 		insertMsg, err := mock_segcore.GenInsertMsg(suite.collection.GetCCollection(), suite.partitionID, segID, msgLen)
 		suite.Require().NoError(err)
-		insertRecord, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
+		insertRecord, _, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
 		suite.Require().NoError(err)
 		err = seg.Insert(ctx, insertMsg.RowIDs, insertMsg.Timestamps, insertRecord)
 		suite.Require().NoError(err)
@@ -455,7 +455,7 @@ func (suite *RetrieveSuite) TestRetrieveWithFilterDoesNotPruneGrowing() {
 
 		insertMsg, err := mock_segcore.GenInsertMsg(suite.collection.GetCCollection(), suite.partitionID, segID, msgLen)
 		suite.Require().NoError(err)
-		insertRecord, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
+		insertRecord, _, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
 		suite.Require().NoError(err)
 		err = seg.Insert(ctx, insertMsg.RowIDs, insertMsg.Timestamps, insertRecord)
 		suite.Require().NoError(err)

internal/querynodev2/segments/search_test.go

@@ -127,7 +127,7 @@ func (suite *SearchSuite) SetupTest() {
 
 	insertMsg, err := mock_segcore.GenInsertMsg(suite.collection.GetCCollection(), suite.partitionID, suite.growing.ID(), msgLength)
 	suite.Require().NoError(err)
-	insertRecord, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
+	insertRecord, _, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
 	suite.Require().NoError(err)
 	suite.growing.Insert(ctx, insertMsg.RowIDs, insertMsg.Timestamps, insertRecord)
 
@@ -289,7 +289,7 @@ func (suite *SearchSuite) TestSearchStreamingWithFilterDoesNotPruneGrowing() {
 
 		insertMsg, err := mock_segcore.GenInsertMsg(suite.collection.GetCCollection(), suite.partitionID, segID, msgLen)
 		suite.Require().NoError(err)
-		insertRecord, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
+		insertRecord, _, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
 		suite.Require().NoError(err)
 		err = seg.Insert(ctx, insertMsg.RowIDs, insertMsg.Timestamps, insertRecord)
 		suite.Require().NoError(err)

internal/querynodev2/segments/segment_test.go

@@ -137,7 +137,7 @@ func (suite *SegmentSuite) SetupTest() {
 
 	insertMsg, err := mock_segcore.GenInsertMsg(suite.collection.GetCCollection(), suite.partitionID, suite.growing.ID(), msgLength)
 	suite.Require().NoError(err)
-	insertRecord, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
+	insertRecord, _, err := storage.TransferInsertMsgToInsertRecord(suite.collection.Schema(), insertMsg)
 	suite.Require().NoError(err)
 	err = suite.growing.Insert(ctx, insertMsg.RowIDs, insertMsg.Timestamps, insertRecord)
 	suite.Require().NoError(err)

internal/storage/utils.go

@@ -1693,28 +1693,52 @@ func TransferInsertDataToInsertRecord(insertData *InsertData) (*segcorepb.Insert
 	return insertRecord, nil
 }
 
-func TransferInsertMsgToInsertRecord(schema *schemapb.CollectionSchema, msg *msgstream.InsertMsg) (*segcorepb.InsertRecord, error) {
+// TransferInsertMsgToInsertRecord converts an insert message into a segcore
+// insert record under the given schema. Payload columns of fields absent from
+// the schema are skipped and their field IDs returned, so the caller decides
+// how to surface the drop (log/metric) with its own context. Only the
+// column-based branch applies this filtering: the legacy row-based format
+// predates schema change and cannot carry since-dropped fields.
+func TransferInsertMsgToInsertRecord(schema *schemapb.CollectionSchema, msg *msgstream.InsertMsg) (*segcorepb.InsertRecord, []int64, error) {
 	if msg.IsRowBased() {
 		insertData, err := RowBasedInsertMsgToInsertData(msg, schema, false)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
-		return TransferInsertDataToInsertRecord(insertData)
+		insertRecord, err := TransferInsertDataToInsertRecord(insertData)
+		return insertRecord, nil, err
 	}
 
 	// column base insert msg
 	if err := validateColumnBasedInsertMsgNullableVectors(schema, msg); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	insertRecord := &segcorepb.InsertRecord{
 		NumRows:    int64(msg.NumRows),
-		FieldsData: make([]*schemapb.FieldData, 0),
+		FieldsData: make([]*schemapb.FieldData, 0, len(msg.FieldsData)),
 	}
 
-	insertRecord.FieldsData = append(insertRecord.FieldsData, msg.FieldsData...)
+	// The payload was written under the schema version at WAL-append time. On WAL
+	// replay after schema changes, it may still carry columns of since-dropped
+	// fields, while the growing segment is rebuilt with the current schema and can
+	// never own such columns (segcore hard-asserts on unknown fields). Dropped-field
+	// data is unqueryable by definition, so skip those columns instead of passing
+	// them through.
+	knownFields := typeutil.NewSet[int64]()
+	for _, field := range typeutil.GetAllFieldSchemas(schema) {
+		knownFields.Insert(field.GetFieldID())
+	}
+	var skippedFields []int64
+	for _, fieldData := range msg.FieldsData {
+		if !knownFields.Contain(fieldData.GetFieldId()) {
+			skippedFields = append(skippedFields, fieldData.GetFieldId())
+			continue
+		}
+		insertRecord.FieldsData = append(insertRecord.FieldsData, fieldData)
+	}
 
-	return insertRecord, nil
+	return insertRecord, skippedFields, nil
 }
 
 func Min(a, b int64) int64 {

internal/storage/utils_test.go

@@ -1125,8 +1125,44 @@ func TestRowBasedTransferInsertMsgToInsertRecord(t *testing.T) {
 	schema, _, _ := genAllFieldsSchema(dim, false)
 	msg, _, _ := genRowBasedInsertMsg(numRows, dim)
 
-	_, err := TransferInsertMsgToInsertRecord(schema, msg)
+	_, skippedFields, err := TransferInsertMsgToInsertRecord(schema, msg)
 	assert.NoError(t, err)
+	assert.Nil(t, skippedFields)
+}
+
+func TestColumnBasedTransferInsertMsgToInsertRecordSkipDroppedField(t *testing.T) {
+	// Simulate WAL replay after drop-field: the payload still carries the column of
+	// a since-dropped field, which must be skipped instead of forwarded to segcore.
+	schema := &schemapb.CollectionSchema{
+		Name: "test_skip_dropped_field",
+		Fields: []*schemapb.FieldSchema{
+			{FieldID: 100, Name: "pk", IsPrimaryKey: true, DataType: schemapb.DataType_Int64},
+			{FieldID: 101, Name: "age", DataType: schemapb.DataType_Int64},
+		},
+	}
+	numRows := 4
+	msg, _, _ := genColumnBasedInsertMsg(schema, numRows, 8)
+	msg.FieldsData = append(msg.FieldsData, &schemapb.FieldData{
+		Type:      schemapb.DataType_VarChar,
+		FieldName: "dropped",
+		FieldId:   158,
+		Field: &schemapb.FieldData_Scalars{
+			Scalars: &schemapb.ScalarField{
+				Data: &schemapb.ScalarField_StringData{
+					StringData: &schemapb.StringArray{Data: []string{"a", "b", "c", "d"}},
+				},
+			},
+		},
+	})
+
+	record, skippedFields, err := TransferInsertMsgToInsertRecord(schema, msg)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(numRows), record.GetNumRows())
+	fieldIDs := lo.Map(record.GetFieldsData(), func(fd *schemapb.FieldData, _ int) int64 {
+		return fd.GetFieldId()
+	})
+	assert.ElementsMatch(t, []int64{100, 101}, fieldIDs)
+	assert.Equal(t, []int64{158}, skippedFields)
 }
 
 func TestRowBasedInsertMsgToInsertFloat16VectorDataError(t *testing.T) {
@@ -1550,7 +1586,7 @@ func TestColumnBasedInsertMsgToInsertDataRejectsNullableVectorNonCompactData(t *
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), "compact")
 
-			_, err = TransferInsertMsgToInsertRecord(schema, msg)
+			_, _, err = TransferInsertMsgToInsertRecord(schema, msg)
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), "compact")
 		})
@@ -1658,7 +1694,7 @@ func TestColumnBasedInsertMsgToInsertDataRejectsNullableVectorPartialRowData(t *
 			_, err := ColumnBasedInsertMsgToInsertData(msg, schema)
 			require.Error(t, err)
 
-			_, err = TransferInsertMsgToInsertRecord(schema, msg)
+			_, _, err = TransferInsertMsgToInsertRecord(schema, msg)
 			require.Error(t, err)
 		})
 	}

pkg/metrics/querynode_metrics.go

@@ -107,6 +107,17 @@ var (
 			collectionIDLabelName,
 		})
 
+	QueryNodeSkippedInsertFieldCount = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: milvusNamespace,
+			Subsystem: typeutil.QueryNodeRole,
+			Name:      "skipped_insert_field_count",
+			Help:      "count of insert payload field columns skipped because the field is absent from the current schema, e.g. dropped fields carried by messages replayed from WAL",
+		}, []string{
+			nodeIDLabelName,
+			collectionIDLabelName,
+		})
+
 	QueryNodeNumPartitions = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: milvusNamespace,
@@ -991,6 +1002,7 @@ func RegisterQueryNode(registry *prometheus.Registry) {
 	registry.MustRegister(QueryNodeConsumeCounter)
 	registry.MustRegister(QueryNodeExecuteCounter)
 	registry.MustRegister(QueryNodeConsumerMsgCount)
+	registry.MustRegister(QueryNodeSkippedInsertFieldCount)
 	registry.MustRegister(QueryNodeConsumeTimeTickLag)
 	registry.MustRegister(QueryNodeMsgDispatcherTtLag)
 	registry.MustRegister(QueryNodeSegmentSearchLatencyPerVector)
@@ -1050,6 +1062,7 @@ func CleanupQueryNodeCollectionMetrics(nodeID int64, collectionID int64) {
 		collectionIDLabelName: fmt.Sprint(collectionID),
 	}
 	QueryNodeConsumerMsgCount.DeletePartialMatch(labels)
+	QueryNodeSkippedInsertFieldCount.DeletePartialMatch(labels)
 	QueryNodeConsumeTimeTickLag.DeletePartialMatch(labels)
 	QueryNodeNumEntities.DeletePartialMatch(labels)
 	QueryNodeEntitiesSize.DeletePartialMatch(labels)