Remove speculative Arrow overlap trimming

Keep the JDBC-parity row count cap for CloudFetch payloads and avoid adding overlap handling that is not reproducible.

Signed-off-by: Madhavendra Rathore <madhavendra.rathore@databricks.com>

Author: msrathore-db

internal/rows/arrowbased/arrowRecordIterator.go

@@ -38,67 +38,29 @@ type arrowRecordIterator struct {
 	isFinished         bool
 	arrowSchemaBytes   []byte
 	arrowSchema        *arrow.Schema
-	nextRowNumber      int64
-	hasNextRowNumber   bool
 }
 
 var _ rows.ArrowBatchIterator = (*arrowRecordIterator)(nil)
 
 // Retrieve the next arrow record
 func (ri *arrowRecordIterator) Next() (arrow.Record, error) {
-	for {
-		if !ri.HasNext() {
-			// returning EOF indicates that there are no more records to iterate
-			return nil, io.EOF
-		}
-
-		// make sure we have the current batch
-		err := ri.getCurrentBatch()
-		if err != nil {
-			return nil, err
-		}
-
-		// return next record in current batch
-		r, err := ri.currentBatch.Next()
-		if err != nil {
-			ri.checkFinished()
-			return nil, err
-		}
-
-		r2 := ri.skipReturnedRows(r)
-		ri.checkFinished()
-		if r2 == nil {
-			continue
-		}
-
-		return r2, nil
+	if !ri.HasNext() {
+		// returning EOF indicates that there are no more records to iterate
+		return nil, io.EOF
 	}
-}
 
-func (ri *arrowRecordIterator) skipReturnedRows(r SparkArrowRecord) arrow.Record {
-	if !ri.hasNextRowNumber {
-		ri.nextRowNumber = r.Start()
-		ri.hasNextRowNumber = true
+	// make sure we have the current batch
+	err := ri.getCurrentBatch()
+	if err != nil {
+		return nil, err
 	}
 
-	if r.End() < ri.nextRowNumber {
-		r.Release()
-		return nil
-	}
+	// return next record in current batch
+	r, err := ri.currentBatch.Next()
 
-	if r.Start() < ri.nextRowNumber {
-		start := ri.nextRowNumber - r.Start()
-		sliced := r.NewSlice(start, r.NumRows())
-		r.Release()
-		if sliced == nil {
-			return nil
-		}
-		ri.nextRowNumber += sliced.NumRows()
-		return sliced
-	}
+	ri.checkFinished()
 
-	ri.nextRowNumber = r.End() + 1
-	return r
+	return r, err
 }
 
 // Indicate whether there are any more records available

internal/rows/arrowbased/arrowRecordIterator_test.go

@@ -194,69 +194,6 @@ func TestArrowRecordIterator(t *testing.T) {
 	})
 }
 
-func TestArrowRecordIterator_SkipsOverlappingReturnedRows(t *testing.T) {
-	var releasedOverlappingRecord bool
-	var slicedStart int64
-	var slicedEnd int64
-
-	overlappingRecord := fakeRecord{
-		fnNumRows: func() int64 { return 10 },
-		fnRelease: func() {
-			releasedOverlappingRecord = true
-		},
-		fnNewSlice: func(i, j int64) arrow.Record {
-			slicedStart = i
-			slicedEnd = j
-			return fakeRecord{fnNumRows: func() int64 { return j - i }}
-		},
-	}
-
-	rs := &arrowRecordIterator{
-		batchIterator: &fakeBatchIterator{
-			index: -1,
-			batches: []SparkArrowBatch{
-				&sparkArrowBatch{
-					Delimiter: rowscanner.NewDelimiter(0, 5),
-					arrowRecords: []SparkArrowRecord{
-						&sparkArrowRecord{
-							Delimiter: rowscanner.NewDelimiter(0, 5),
-							Record:    fakeRecord{fnNumRows: func() int64 { return 5 }},
-						},
-					},
-				},
-				&sparkArrowBatch{
-					Delimiter: rowscanner.NewDelimiter(0, 10),
-					arrowRecords: []SparkArrowRecord{
-						&sparkArrowRecord{
-							Delimiter: rowscanner.NewDelimiter(0, 10),
-							Record:    overlappingRecord,
-						},
-					},
-				},
-			},
-		},
-	}
-	defer rs.Close()
-
-	r1, err := rs.Next()
-	assert.NoError(t, err)
-	assert.Equal(t, int64(5), r1.NumRows())
-	r1.Release()
-
-	r2, err := rs.Next()
-	assert.NoError(t, err)
-	assert.Equal(t, int64(5), r2.NumRows())
-	r2.Release()
-
-	assert.True(t, releasedOverlappingRecord)
-	assert.Equal(t, int64(5), slicedStart)
-	assert.Equal(t, int64(10), slicedEnd)
-
-	r3, err := rs.Next()
-	assert.Nil(t, r3)
-	assert.ErrorIs(t, err, io.EOF)
-}
-
 func TestLimitArrowRecordsUsesExpectedRowCount(t *testing.T) {
 	var releasedOriginal bool
 	var releasedExtra bool