miguelpragier commented on code in PR #40090:
URL: https://github.com/apache/arrow/pull/40090#discussion_r1492340710
##########
go/arrow/flight/flightsql/driver/driver.go:
##########
@@ -37,34 +38,50 @@ import (
)
type Rows struct {
- schema *arrow.Schema
- records []arrow.Record
- currentRecord int
- currentRow int
+ schema *arrow.Schema
+ recordChan chan arrow.Record
+ currentRecord arrow.Record
+ currentRecordMux sync.Mutex
+ currentRow uint64
+ initializedChan chan bool
+ streamError error
+}
+
+func newRows() *Rows {
+ return &Rows{
+ recordChan: make(chan arrow.Record, 1),
+ initializedChan: make(chan bool),
+ }
}
// Columns returns the names of the columns.
func (r *Rows) Columns() []string {
- if len(r.records) == 0 {
+ if r.schema == nil {
return nil
}
// All records have the same columns
- var cols []string
- for _, c := range r.schema.Fields() {
- cols = append(cols, c.Name)
+ cols := make([]string, len(r.schema.Fields()))
+ for i, c := range r.schema.Fields() {
+ cols[i] = c.Name
}
return cols
}
+func (r *Rows) releaseRecord() {
+ r.currentRecordMux.Lock()
+ defer r.currentRecordMux.Unlock()
+
+ if r.currentRecord != nil {
+ r.currentRecord.Release()
+ r.currentRecord = nil
+ }
+}
+
// Close closes the rows iterator.
func (r *Rows) Close() error {
Review Comment:
with these adjustments, Close() now properly cancel() the streaming.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscr...@arrow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
