fix: close rows iterator

internal/infra/pull/datasource_ws.go

@@ -289,3 +289,7 @@ func (rs *ResultStream) Value() pull.Row {
 func (rs *ResultStream) Error() error {
 	return rs.err
 }
+
+func (rs *ResultStream) Close() error {
+	return nil
+}

internal/infra/pull/http_datasource.go

@@ -63,6 +63,8 @@ func (ds *HTTPDataSource) Read(source pull.Table, filter pull.Filter) (pull.RowS
 		return nil, err
 	}
 
+	defer reader.Close()
+
 	result := pull.RowSet{}
 	for reader.Next() {
 		result = append(result, reader.Value())

internal/infra/pull/rowreader_json.go

@@ -68,3 +68,7 @@ func (jrr *JSONRowReader) Value() pull.Row {
 func (jrr *JSONRowReader) Error() error {
 	return jrr.err
 }
+
+func (jrr *JSONRowReader) Close() error {
+	return nil
+}

internal/infra/pull/sql_datasource.go

@@ -99,6 +99,8 @@ func (ds *SQLDataSource) Read(source pull.Table, filter pull.Filter) (pull.RowSe
 		return nil, err
 	}
 
+	defer reader.Close()
+
 	result := pull.RowSet{}
 	for reader.Next() {
 		result = append(result, reader.Value())
@@ -250,3 +252,8 @@ func (di *SQLDataIterator) Value() pull.Row {
 func (di *SQLDataIterator) Error() error {
 	return di.err
 }
+
+// Close returns the iterator
+func (di *SQLDataIterator) Close() error {
+	return di.rows.Close()
+}

pkg/pull/driven.go

@@ -40,6 +40,7 @@ type RowReader interface {
 	Next() bool
 	Value() Row
 	Error() error
+	Close() error
 }
 
 // TraceListener receives diagnostic trace.

pkg/pull/driven_DatasourceInMemory.go

@@ -40,6 +40,8 @@ func (ds DataSourceInMemory) Read(source Table, filter Filter) (RowSet, error) {
 		return nil, err
 	}
 
+	defer reader.Close()
+
 	result := RowSet{}
 	for reader.Next() {
 		result = append(result, reader.Value())
@@ -104,3 +106,4 @@ func (rr *RowReaderInMemory) Value() Row {
 	return row
 }
 func (rr *RowReaderInMemory) Error() error { return nil }
+func (rr *RowReaderInMemory) Close() error { return nil }

pkg/pull/driven_OneEmptyRowReader.go

@@ -38,3 +38,6 @@ func (r OneEmptyRowReader) Value() Row { return Row{} }
 
 // Error return always nil
 func (r OneEmptyRowReader) Error() error { return nil }
+
+// Close return always nil
+func (r OneEmptyRowReader) Close() error { return nil }

pkg/pull/driver.go

@@ -110,6 +110,8 @@ func (p *puller) Pull(start Table, filter Filter, filterCohort RowReader, exclud
 			return fmt.Errorf("%w", err)
 		}
 
+		defer reader.Close()
+
 		for reader.Next() {
 			IncLinesPerStepCount(string(start.Name))
 			row := start.export(reader.Value())

pkg/pull/driver_parallel.go

@@ -121,6 +121,8 @@ func (p *pullerParallel) Pull(start Table, filter Filter, filterCohort RowReader
 			return fmt.Errorf("%w", err)
 		}
 
+		defer reader.Close()
+
 		for reader.Next() {
 			IncLinesPerStepCount(string(start.Name))
 			p.inChan <- reader.Value()