Fix Row lifetime for IPC stream (#92)

* Fix Row lifetime for IPC stream

* Rename row variable for clarity on its purpose

Author: jasonlin45

go/adbc/driver/databricks/ipc_reader_adapter.go

@@ -20,6 +20,7 @@ package databricks
 import (
 	"bytes"
 	"context"
+	"database/sql/driver"
 	"fmt"
 	"io"
 	"sync/atomic"
@@ -38,12 +39,13 @@ type rowsWithIPCStream interface {
 
 // ipcReaderAdapter uses the new IPC stream interface for Arrow access
 type ipcReaderAdapter struct {
-	ipcIterator   dbsqlrows.ArrowIPCStreamIterator
-	currentReader *ipc.Reader
-	currentRecord arrow.RecordBatch
-	schema        *arrow.Schema
-	closed        bool
-	refCount      int64
+	dbsqlRowsHandle dbsqlrows.Rows // Row from which we obtain the iterator - their lifetimes are tied together
+	ipcIterator     dbsqlrows.ArrowIPCStreamIterator
+	currentReader   *ipc.Reader
+	currentRecord   arrow.RecordBatch
+	schema          *arrow.Schema
+	closed          bool
+	refCount        int64
 }
 
 // newIPCReaderAdapter creates a RecordReader using direct IPC stream access
@@ -86,9 +88,10 @@ func newIPCReaderAdapter(ctx context.Context, rows dbsqlrows.Rows) (array.Record
 	}
 
 	adapter := &ipcReaderAdapter{
-		refCount:    1,
-		ipcIterator: ipcIterator,
-		schema:      schema,
+		refCount:        1,
+		dbsqlRowsHandle: rows,
+		ipcIterator:     ipcIterator,
+		schema:          schema,
 	}
 
 	// Initialize the first reader
@@ -202,6 +205,11 @@ func (r *ipcReaderAdapter) Release() {
 		}
 
 		r.ipcIterator.Close()
+
+		if r.dbsqlRowsHandle != nil {
+			r.dbsqlRowsHandle.(driver.Rows).Close()
+			r.dbsqlRowsHandle = nil
+		}
 	}
 }
 

go/adbc/driver/databricks/statement.go

@@ -21,7 +21,6 @@ import (
 	"context"
 	"database/sql"
 	"database/sql/driver"
-	"errors"
 	"fmt"
 
 	"github.com/apache/arrow-adbc/go/adbc"
@@ -132,12 +131,6 @@ func (s *statementImpl) ExecuteQuery(ctx context.Context) (array.RecordReader, i
 		}
 	}
 
-	defer func() {
-		if closeErr := driverRows.Close(); closeErr != nil {
-			err = errors.Join(err, closeErr)
-		}
-	}()
-
 	// Convert to databricks rows interface to get Arrow batches
 	databricksRows, ok := driverRows.(dbsqlrows.Rows)
 	if !ok {