fix: preserve binary string execution semantics

pkg/common/bitmap/bitmap.go

@@ -534,6 +534,12 @@ func (n *Bitmap) RemapOrdered(sels []int64, negate bool) {
 // RemapMaskOrdered is RemapOrdered for an ordered bitmap selection. Selection
 // bitmap iteration is monotonic, so the rewrite uses no row-scaled scratch.
 func (n *Bitmap) RemapMaskOrdered(sels *Bitmap, negate bool) {
+	n.RemapMaskOrderedWithOffset(sels, negate, 0)
+}
+
+// RemapMaskOrderedWithOffset applies an ordered bitmap selection after adding
+// offset to every selected source row, without materializing an index slice.
+func (n *Bitmap) RemapMaskOrderedWithOffset(sels *Bitmap, negate bool, offset uint64) {
 	if n == nil || sels == nil {
 		return
 	}
@@ -570,19 +576,19 @@ func (n *Bitmap) RemapMaskOrdered(sels *Bitmap, negate bool) {
 	iterator := sels.Iterator()
 	if !negate {
 		for iterator.HasNext() {
-			source := int64(iterator.Next())
+			source := int64(iterator.Next() + offset)
 			writeDestination(output, readSource(source))
 			output++
 		}
 	} else {
 		var selected int64 = -1
 		if iterator.HasNext() {
-			selected = int64(iterator.Next())
+			selected = int64(iterator.Next() + offset)
 		}
 		for source := int64(0); source < oldLength; source++ {
 			if source == selected {
 				if iterator.HasNext() {
-					selected = int64(iterator.Next())
+					selected = int64(iterator.Next() + offset)
 				} else {
 					selected = -1
 				}

pkg/common/bitmap/bitmap_test.go

@@ -345,6 +345,28 @@ func TestBitmapRemapMaskOrdered(t *testing.T) {
 	require.True(t, value.Contains(125))
 }
 
+func TestBitmapRemapMaskOrderedWithOffset(t *testing.T) {
+	selection := newBm(129)
+	selection.AddMany([]uint64{0, 62, 128})
+
+	var value Bitmap
+	value.InitWithSize(130)
+	value.AddMany([]uint64{1, 63, 64, 127, 129})
+	value.RemapMaskOrderedWithOffset(selection, false, 1)
+	require.Equal(t, int64(3), value.Len())
+	require.Equal(t, 3, value.Count())
+	require.True(t, value.Contains(0))
+	require.True(t, value.Contains(1))
+	require.True(t, value.Contains(2))
+
+	value.InitWithSize(130)
+	value.AddMany([]uint64{1, 63, 64, 127, 129})
+	value.RemapMaskOrderedWithOffset(selection, true, 1)
+	require.Equal(t, 2, value.Count())
+	require.True(t, value.Contains(62))
+	require.True(t, value.Contains(125))
+}
+
 func TestBitmapExternalStorageUnmarshal(t *testing.T) {
 	source := newBm(128)
 	source.Add(1)

pkg/container/batch/allocation_account_test.go

@@ -299,6 +299,10 @@ func TestBatchGroupingCodecRoundTrip(t *testing.T) {
 	source := newBatchAllocationTestSource(t, mp, nil)
 	source.Vecs[0].GetGrouping().Add(1, 7, 31)
 	source.Vecs[1].GetGrouping().Add(2, 9)
+	require.NoError(t, source.Vecs[1].SetIsBinaryStringAt(0, true))
+	kinds := make([]vector.PrepareParamKind, source.RowCount())
+	kinds[0] = vector.PrepareParamInteger
+	require.NoError(t, source.Vecs[1].SetPrepareParamKindsWithMP(kinds, mp))
 	source.ExtraBuf = bytes.Repeat([]byte("x"), 1<<20)
 
 	var encoded bytes.Buffer
@@ -316,6 +320,9 @@ func TestBatchGroupingCodecRoundTrip(t *testing.T) {
 	for i := range source.Vecs {
 		require.True(t, decoded.Vecs[i].GetGrouping().IsSame(source.Vecs[i].GetGrouping()))
 	}
+	require.True(t, decoded.Vecs[1].GetIsBinaryStringAt(0))
+	require.False(t, decoded.Vecs[1].GetIsBinaryStringAt(1))
+	require.Equal(t, vector.PrepareParamInteger, decoded.Vecs[1].GetPrepareParamKindAt(0))
 	withoutGrouping := newBatchAllocationTestSource(t, mp, nil)
 	encoded.Reset()
 	require.NoError(t, withoutGrouping.MarshalBinaryWithGroupingTo(&encoded))
@@ -355,8 +362,12 @@ func TestBatchGroupingCodecRejectsStableMetadataBeforePayloadAllocation(t *testi
 			source := NewWithSize(0)
 			source.Attrs = test.attrs
 			source.ExtraBuf = test.extra
+			var payload bytes.Buffer
+			require.NoError(t, source.MarshalBinaryTo(&payload))
 			var encoded bytes.Buffer
-			require.NoError(t, source.MarshalBinaryTo(&encoded))
+			payloadSize := int64(payload.Len())
+			encoded.Write(types.EncodeInt64(&payloadSize))
+			encoded.Write(payload.Bytes())
 
 			decoded := NewOffHeapEmpty()
 			require.NoError(t, decoded.SetAllocationAccount(state.selection))