fix(iceberg): reject position delete row ordinal overflow

pkg/sql/colexec/icebergdelete/delete_apply.go

@@ -245,6 +245,9 @@ func (s *ApplyState) ApplyPositionMask(ctx context.Context, dataFile string, sta
 	if rowCount <= 0 {
 		return nil, nil
 	}
+	if err := validatePositionRowOrdinalRange(startRowOrdinal, rowCount); err != nil {
+		return nil, api.ToMOErr(ctx, err)
+	}
 	if err := s.CheckMemory(ctx); err != nil {
 		return nil, err
 	}
@@ -278,15 +281,30 @@ func (s *ApplyState) ApplyEqualityMask(ctx context.Context, rows [][]any) ([]boo
 	return keep, nil
 }
 
-func RowOrdinals(startRowOrdinal int64, rowCount int) []int64 {
+// validatePositionRowOrdinalRange rejects corrupted metadata before a wrapped
+// ordinal silently misses a position delete. rowCount must be positive.
+func validatePositionRowOrdinalRange(startRowOrdinal int64, rowCount int) error {
+	if startRowOrdinal < 0 || startRowOrdinal > math.MaxInt64-int64(rowCount-1) {
+		return api.NewError(api.ErrMetadataInvalid, "Iceberg position row ordinal range is invalid", map[string]string{
+			"start_row_ordinal": strconv.FormatInt(startRowOrdinal, 10),
+			"row_count":         strconv.Itoa(rowCount),
+		})
+	}
+	return nil
+}
+
+func RowOrdinals(startRowOrdinal int64, rowCount int) ([]int64, error) {
 	if rowCount <= 0 {
-		return nil
+		return nil, nil
+	}
+	if err := validatePositionRowOrdinalRange(startRowOrdinal, rowCount); err != nil {
+		return nil, err
 	}
 	out := make([]int64, rowCount)
 	for i := range out {
 		out[i] = startRowOrdinal + int64(i)
 	}
-	return out
+	return out, nil
 }
 
 func equalityKey(values []any) string {

pkg/sql/colexec/icebergdelete/delete_apply_test.go

@@ -51,8 +51,8 @@ func TestPositionDeleteMaskUsesGlobalRowOrdinal(t *testing.T) {
 	if state.Profile.PositionRowsFiltered != 2 {
 		t.Fatalf("unexpected profile: %+v", state.Profile)
 	}
-	if got := RowOrdinals(10, 3); len(got) != 3 || got[0] != 10 || got[2] != 12 {
-		t.Fatalf("unexpected row ordinals: %v", got)
+	if got, err := RowOrdinals(10, 3); err != nil || len(got) != 3 || got[0] != 10 || got[2] != 12 {
+		t.Fatalf("unexpected row ordinals: %v, err %v", got, err)
 	}
 }
 
@@ -181,3 +181,39 @@ func TestDeleteApplySpillEnabledFailsFastUntilImplemented(t *testing.T) {
 		t.Fatalf("expected profile to record spill intent: %+v", state.Profile)
 	}
 }
+
+func TestPositionDeleteRejectsOverflowedRowOrdinals(t *testing.T) {
+	state := NewApplyState(Options{MaxMemoryBytes: 1024})
+	state.Position.Add("data.parquet", math.MaxInt64-1)
+	state.Position.Add("data.parquet", math.MaxInt64)
+
+	for _, tc := range []struct {
+		name  string
+		start int64
+		count int
+	}{
+		{name: "overflow past max int64", start: math.MaxInt64 - 1, count: 3},
+		{name: "negative ordinal base", start: -1, count: 1},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ordinals, ordinalErr := RowOrdinals(tc.start, tc.count)
+			if ordinalErr == nil || !strings.Contains(ordinalErr.Error(), "ICEBERG_METADATA_INVALID") || ordinals != nil {
+				t.Fatalf("RowOrdinals(%d,%d) returned %v, %v", tc.start, tc.count, ordinals, ordinalErr)
+			}
+			keep, err := state.ApplyPositionMask(context.Background(), "data.parquet", tc.start, tc.count)
+			if err == nil || !strings.Contains(err.Error(), "ICEBERG_METADATA_INVALID") || keep != nil {
+				t.Fatalf("ApplyPositionMask(%d,%d) returned %v, %v", tc.start, tc.count, keep, err)
+			}
+		})
+	}
+
+	// A range ending exactly at MaxInt64 must still be valid.
+	ordinals, err := RowOrdinals(math.MaxInt64-1, 2)
+	if err != nil || len(ordinals) != 2 || ordinals[1] != math.MaxInt64 {
+		t.Fatalf("valid boundary ordinals=%v err=%v", ordinals, err)
+	}
+	mask, err := state.ApplyPositionMask(context.Background(), "data.parquet", math.MaxInt64-1, 2)
+	if err != nil || len(mask) != 2 || mask[0] || mask[1] {
+		t.Fatalf("valid boundary delete mask=%v err=%v", mask, err)
+	}
+}