fix: Split large sels into multiple batches in CompactBatchs.Union - hotfix (#22841)

### **User description**
## What type of PR is this?

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #22825

## What this PR does / why we need it:

This PR fixes a bug in `CompactBatchs.Union` method where large
selection arrays (`sels`) exceeding `batchMaxRow` were not properly
split into multiple batches when the batch collection was empty
(`bats.Length() == 0`).

**Problem:**
In the `Union` method, when `bats.Length() == 0` and `sels` length
exceeds `batchMaxRow`, the original implementation would create a single
batch containing all selected rows without checking the batch size
limit. This violates the `batchMaxRow` constraint that each batch should
respect, potentially causing:
- Batches exceeding the maximum row limit
- Inconsistent behavior compared to the case when `bats.Length() != 0`
(which already handles large `sels` correctly by splitting them)

**Solution:**
Modified the `Union` method to handle large `sels` arrays when
`bats.Length() == 0` by splitting them into multiple batches, ensuring
each batch respects the `batchMaxRow` limit. The fix iteratively
processes `sels` in chunks of `batchMaxRow` size, creating multiple
batches as needed.

**Changes:**
- Updated `pkg/container/batch/compact_batchs.go`: Added logic in the
`bats.Length() == 0` branch to split large `sels` into multiple batches,
making it consistent with the existing logic for `bats.Length() != 0`
case
- Added comprehensive test cases in
`pkg/container/batch/compact_batchs_test.go`:
`TestCompactBatchsUnionLargeSels` covers various scenarios:
  - Union with `selsLen > batchMaxRow` when `bats.Length() == 0`
  - Union with `selsLen == batchMaxRow` when `bats.Length() == 0`
  - Union with large `sels` when last batch is already full
  - Union with large `sels` when last batch has some rows

This fix ensures consistent batch size handling across all code paths in
the `Union` method, preventing potential issues with oversized batches.


___

### **PR Type**
Bug fix, Tests


___

### **Description**
- Split large selection arrays into multiple batches when empty

- Ensures each batch respects batchMaxRow limit constraint

- Added comprehensive test coverage for large selections

- Fixes inconsistent behavior between empty and non-empty batch
collections


___

### Diagram Walkthrough


```mermaid
flowchart LR
  A["Union method called<br/>with large sels"] --> B{"bats.Length() == 0?"}
  B -->|Yes| C["Split sels into<br/>batchMaxRow chunks"]
  C --> D["Create multiple<br/>batches iteratively"]
  D --> E["Each batch respects<br/>batchMaxRow limit"]
  B -->|No| F["Existing logic<br/>handles splitting"]
  E --> G["Consistent behavior<br/>across all paths"]
  F --> G
```



<details> <summary><h3> File Walkthrough</h3></summary>

<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Bug
fix</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>compact_batchs.go</strong><dd><code>Implement batch
splitting for large selections</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

pkg/container/batch/compact_batchs.go

<ul><li>Wrapped large selection handling in a loop to process selections
in <br>chunks<br> <li> Each chunk is limited to <code>batchMaxRow</code>
size<br> <li> Creates multiple batches iteratively instead of single
oversized batch<br> <li> Maintains consistency with existing logic for
non-empty batch <br>collections</ul>


</details>


  </td>
<td><a
href="https://github.com/matrixorigin/matrixone/pull/22841/files#diff-13a01ffd2c00c5e8faff10e905d8067f400bafe3d940f6da252774850e0bfc85">+18/-8</a>&nbsp;
&nbsp; </td>

</tr>
</table></td></tr><tr><td><strong>Tests</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>compact_batchs_test.go</strong><dd><code>Add
comprehensive tests for large selection splitting</code>&nbsp; &nbsp;
&nbsp; &nbsp; </dd></summary>
<hr>

pkg/container/batch/compact_batchs_test.go

<ul><li>Added <code>TestCompactBatchsUnionLargeSels</code> test function
with four <br>comprehensive test cases<br> <li> Test case 1: Validates
splitting when selsLen > batchMaxRow on empty <br>collection<br> <li>
Test case 2: Validates single batch creation when selsLen ==
<br>batchMaxRow<br> <li> Test case 3: Validates new batch creation when
last batch is full<br> <li> Test case 4: Validates batch filling and
overflow when last batch has <br>partial rows</ul>


</details>


  </td>
<td><a
href="https://github.com/matrixorigin/matrixone/pull/22841/files#diff-b568feda7c727d7a72c52ec6ebd3a44ed0676127e33093f4d66ddc1c784b2471">+116/-0</a>&nbsp;
</td>

</tr>
</table></td></tr></tr></tbody></table>

</details>

___

Author: aptend

pkg/container/batch/compact_batchs.go

@@ -166,16 +166,26 @@ func (bats *CompactBatchs) Union(mpool *mpool.MPool, inBatch *Batch, sels []int3
 	}
 
 	if bats.Length() == 0 {
-		tmpBat := NewWithSize(len(inBatch.Vecs))
-		for i := range tmpBat.Vecs {
-			tmpBat.Vecs[i] = vector.NewVec(*inBatch.Vecs[i].GetType())
-			err := tmpBat.Vecs[i].UnionInt32(inBatch.Vecs[i], sels, mpool)
-			if err != nil {
-				return err
+		// Handle large sels: split into multiple batches if selsLen > batchMaxRow
+		remainingSels := sels
+		for len(remainingSels) > 0 {
+			tmpSize := len(remainingSels)
+			if tmpSize > bats.batchMaxRow {
+				tmpSize = bats.batchMaxRow
 			}
+			tmpSels := remainingSels[:tmpSize]
+			tmpBat := NewWithSize(len(inBatch.Vecs))
+			for i := range tmpBat.Vecs {
+				tmpBat.Vecs[i] = vector.NewVec(*inBatch.Vecs[i].GetType())
+				err := tmpBat.Vecs[i].UnionInt32(inBatch.Vecs[i], tmpSels, mpool)
+				if err != nil {
+					return err
+				}
+			}
+			tmpBat.rowCount = tmpBat.Vecs[0].Length()
+			bats.batchs = append(bats.batchs, tmpBat)
+			remainingSels = remainingSels[tmpSize:]
 		}
-		tmpBat.rowCount = tmpBat.Vecs[0].Length()
-		bats.batchs = append(bats.batchs, tmpBat)
 		return nil
 	}
 

pkg/container/batch/compact_batchs_test.go

@@ -232,6 +232,122 @@ func TestCompactBatchsUnion(t *testing.T) {
 	bats.Clean(mp)
 }
 
+func TestCompactBatchsUnionLargeSels(t *testing.T) {
+	mp := mpool.MustNewZero()
+	bats := NewCompactBatchs(TestBatchMaxRow)
+
+	// Test case 1: Union with selsLen > batchMaxRow when bats.Length() == 0
+	bat1 := makeTestBatch(20000, mp) // Create a large batch
+	var sels []int32
+	for i := 0; i < 15000; i++ {
+		sels = append(sels, int32(i))
+	}
+	err := bats.Union(mp, bat1, sels)
+	require.NoError(t, err)
+
+	// Should create multiple batches since selsLen (15000) > batchMaxRow (8192)
+	expectedBatches1 := (15000 + TestBatchMaxRow - 1) / TestBatchMaxRow // Ceiling division
+	require.Equal(t, expectedBatches1, bats.Length())
+	require.Equal(t, 15000, bats.RowCount())
+
+	// Verify each batch doesn't exceed batchMaxRow
+	for i := 0; i < bats.Length(); i++ {
+		bat := bats.Get(i)
+		require.NotNil(t, bat)
+		require.LessOrEqual(t, bat.rowCount, TestBatchMaxRow)
+		if i < bats.Length()-1 {
+			// All batches except the last should be exactly batchMaxRow
+			require.Equal(t, TestBatchMaxRow, bat.rowCount)
+		}
+	}
+
+	bat1.Clean(mp)
+	bats.Clean(mp)
+	require.Equal(t, int64(0), mp.CurrNB())
+
+	// Test case 2: Union with selsLen == batchMaxRow when bats.Length() == 0
+	bats = NewCompactBatchs(TestBatchMaxRow)
+	bat2 := makeTestBatch(TestBatchMaxRow, mp)
+	sels = nil
+	for i := 0; i < TestBatchMaxRow; i++ {
+		sels = append(sels, int32(i))
+	}
+	err = bats.Union(mp, bat2, sels)
+	require.NoError(t, err)
+	require.Equal(t, 1, bats.Length())
+	require.Equal(t, TestBatchMaxRow, bats.RowCount())
+	require.Equal(t, TestBatchMaxRow, bats.Get(0).rowCount)
+
+	bat2.Clean(mp)
+	bats.Clean(mp)
+	require.Equal(t, int64(0), mp.CurrNB())
+
+	// Test case 3: Union with selsLen > batchMaxRow when lastBat is already full
+	bats = NewCompactBatchs(TestBatchMaxRow)
+	bat3 := makeTestBatch(TestBatchMaxRow, mp)
+	sels = nil
+	for i := 0; i < TestBatchMaxRow; i++ {
+		sels = append(sels, int32(i))
+	}
+	// Fill first batch to full
+	err = bats.Union(mp, bat3, sels)
+	require.NoError(t, err)
+	require.Equal(t, 1, bats.Length())
+	require.Equal(t, TestBatchMaxRow, bats.Get(0).rowCount)
+
+	// Now Union with large sels again
+	bat4 := makeTestBatch(20000, mp)
+	sels = nil
+	for i := 0; i < 10000; i++ {
+		sels = append(sels, int32(i))
+	}
+	err = bats.Union(mp, bat4, sels)
+	require.NoError(t, err)
+	// Should create new batches since lastBat is full
+	// First batch is full (8192), remaining 10000 sels should create 2 batches: 8192 + 1808
+	expectedBatches3 := 1 + (10000+TestBatchMaxRow-1)/TestBatchMaxRow
+	require.Equal(t, expectedBatches3, bats.Length())
+	require.Equal(t, TestBatchMaxRow+10000, bats.RowCount())
+	require.Equal(t, TestBatchMaxRow, bats.Get(0).rowCount)
+	require.Equal(t, TestBatchMaxRow, bats.Get(1).rowCount)
+	require.Equal(t, 10000-TestBatchMaxRow, bats.Get(2).rowCount) // 10000 - 8192 = 1808
+
+	bat3.Clean(mp)
+	bat4.Clean(mp)
+	bats.Clean(mp)
+	require.Equal(t, int64(0), mp.CurrNB())
+
+	// Test case 4: Union with selsLen > batchMaxRow when lastBat has some rows
+	bats = NewCompactBatchs(TestBatchMaxRow)
+	bat5 := makeTestBatch(10000, mp)
+	sels = nil
+	for i := 0; i < 5000; i++ {
+		sels = append(sels, int32(i))
+	}
+	// First Union: creates batch with 5000 rows
+	err = bats.Union(mp, bat5, sels)
+	require.NoError(t, err)
+	require.Equal(t, 1, bats.Length())
+	require.Equal(t, 5000, bats.Get(0).rowCount)
+
+	// Second Union: with 10000 sels, should fill first batch to 8192, then create new batch
+	sels = nil
+	for i := 0; i < 10000; i++ {
+		sels = append(sels, int32(i))
+	}
+	err = bats.Union(mp, bat5, sels)
+	require.NoError(t, err)
+	// First batch should be filled to 8192 (5000 + 3192), second batch has remaining 6808
+	require.Equal(t, 2, bats.Length())
+	require.Equal(t, 5000+10000, bats.RowCount())
+	require.Equal(t, TestBatchMaxRow, bats.Get(0).rowCount)
+	require.Equal(t, 10000-3192, bats.Get(1).rowCount) // 10000 - (8192-5000) = 6808
+
+	bat5.Clean(mp)
+	bats.Clean(mp)
+	require.Equal(t, int64(0), mp.CurrNB())
+}
+
 func makeTestBatch(max int, mp *mpool.MPool) *Batch {
 	bat := NewWithSize(2)
 	bat.Vecs[0] = makeTestVec(max, mp)