fix(infinity): keep one row per chunk id on insert (#21043)

internal/engine/infinity/chunk.go

@@ -443,6 +443,11 @@ func (e *Engine) InsertChunks(ctx context.Context, chunks []map[string]interface
 		return nil, err
 	}
 
+	// Chunk ids are content-addressed, so one batch can repeat an id. Elasticsearch
+	// keeps a single row per _id; Infinity has no id uniqueness, so collapse to one row
+	// per id, last wins. Repeats across batches need nothing: the delete below covers them.
+	insertChunks = dedupeByChunkID(insertChunks)
+
 	// Delete existing rows with matching IDs
 	if len(insertChunks) > 0 {
 		idList := make([]string, len(insertChunks))
@@ -486,6 +491,37 @@ func (e *Engine) InsertChunks(ctx context.Context, chunks []map[string]interface
 // embedding size; no mapping file declares them.
 var vectorColumnPattern = regexp.MustCompile(`^q_\d+_vec$`)
 
+// dedupeByChunkID keeps the last row of each chunk id, mirroring Elasticsearch, where
+// a repeated _id overwrites the earlier document. Rows without an id are all kept.
+func dedupeByChunkID(rows []map[string]interface{}) []map[string]interface{} {
+	if len(rows) < 2 {
+		return rows
+	}
+	lastAt := make(map[string]int, len(rows))
+	dupes := 0
+	for i, row := range rows {
+		id, ok := row["id"].(string)
+		if !ok || id == "" {
+			continue
+		}
+		if _, seen := lastAt[id]; seen {
+			dupes++
+		}
+		lastAt[id] = i
+	}
+	if dupes == 0 {
+		return rows
+	}
+	out := make([]map[string]interface{}, 0, len(rows)-dupes)
+	for i, row := range rows {
+		if id, ok := row["id"].(string); ok && id != "" && lastAt[id] != i {
+			continue
+		}
+		out = append(out, row)
+	}
+	return out
+}
+
 // dropUndeclaredFields removes each row key that is not a column of the table
 // and that no mapping file declares, such as parser bookkeeping keys
 // (slide_number, __outline__). Infinity fails the whole batch on one unknown

internal/engine/infinity/chunk_transform_test.go

@@ -919,3 +919,37 @@ func TestDropUndeclaredFieldsFailsWhenMappingIsMissing(t *testing.T) {
 		t.Fatal("a failed mapping read must leave the row unchanged")
 	}
 }
+
+// TestDedupeByChunkIDKeepsLast pins that a repeated id keeps its last row.
+func TestDedupeByChunkIDKeepsLast(t *testing.T) {
+	rows := []map[string]interface{}{
+		{"id": "a", "chunk_order_int": int(0)},
+		{"id": "b", "chunk_order_int": int(1)},
+		{"id": "a", "chunk_order_int": int(9)},
+		{"id": "c", "chunk_order_int": int(2)},
+	}
+
+	got := dedupeByChunkID(rows)
+	if len(got) != 3 {
+		t.Fatalf("rows = %d, want 3: %#v", len(got), got)
+	}
+	if got[0]["id"] != "b" || got[1]["id"] != "a" || got[2]["id"] != "c" {
+		t.Fatalf("order = %#v, want b, a(last), c", got)
+	}
+	if got[1]["chunk_order_int"] != int(9) {
+		t.Fatalf("kept row = %#v, want the last occurrence of a", got[1])
+	}
+}
+
+// TestDedupeByChunkIDKeepsRowsWithoutID pins that rows the engine cannot key stay.
+func TestDedupeByChunkIDKeepsRowsWithoutID(t *testing.T) {
+	rows := []map[string]interface{}{
+		{"chunk_data": "x"},
+		{"id": ""},
+		{"id": "a"},
+		{"id": "a"},
+	}
+	if got := dedupeByChunkID(rows); len(got) != 3 {
+		t.Fatalf("rows = %d, want 3 (only the duplicate id collapses): %#v", len(got), got)
+	}
+}