Merge pull request #655 from kuba--/close-iterators

Close iterators

Author: ajnavarro

sql/analyzer/pushdown.go

@@ -329,7 +329,10 @@ func (i *releaseIter) Next() (sql.Row, error) {
 	return row, nil
 }
 
-func (i *releaseIter) Close() error {
+func (i *releaseIter) Close() (err error) {
 	i.once.Do(i.release)
-	return nil
+	if i.child != nil {
+		err = i.child.Close()
+	}
+	return err
 }

sql/index/pilosa/driver.go

@@ -251,6 +251,7 @@ func (d *Driver) savePartition(
 		}
 
 		idx.mapping.close()
+		kviter.Close()
 	}()
 
 	for colID = offset; err == nil; colID++ {
@@ -344,6 +345,7 @@ func (d *Driver) Save(
 		return err
 	}
 
+	defer iter.Close()
 	pilosaIndex := idx.index
 	var rows uint64
 	for {

sql/plan/cross_join.go

@@ -155,17 +155,18 @@ func (i *crossJoinIterator) Next() (sql.Row, error) {
 	}
 }
 
-func (i *crossJoinIterator) Close() error {
-	if err := i.l.Close(); err != nil {
-		if i.r != nil {
-			_ = i.r.Close()
-		}
-		return err
+func (i *crossJoinIterator) Close() (err error) {
+	if i.l != nil {
+		err = i.l.Close()
 	}
 
 	if i.r != nil {
-		return i.r.Close()
+		if err == nil {
+			err = i.r.Close()
+		} else {
+			i.r.Close()
+		}
 	}
 
-	return nil
+	return err
 }

sql/plan/exchange.go

@@ -183,10 +183,6 @@ func (it *exchangeRowIter) iterPartitions(ch chan<- sql.Partition) {
 		}
 
 		close(ch)
-
-		if err := it.partitions.Close(); err != nil {
-			it.err <- err
-		}
 	}()
 
 	for {
@@ -278,13 +274,25 @@ func (it *exchangeRowIter) Next() (sql.Row, error) {
 	}
 }
 
-func (it *exchangeRowIter) Close() error {
-	if it.quit == nil {
-		return nil
+func (it *exchangeRowIter) Close() (err error) {
+	if it.quit != nil {
+		close(it.quit)
+		it.quit = nil
+	}
+
+	// TODO(kuba): in my opinion we should close err channel here,
+	// but becasue we use it in another go routine, I'll leave this block commented.
+	//
+	// if it.err != nil {
+	// 	close(it.err)
+	// 	it.err = nil
+	// }
+
+	if it.partitions != nil {
+		err = it.partitions.Close()
 	}
 
-	close(it.quit)
-	return nil
+	return err
 }
 
 type exchangePartition struct {

sql/plan/exchange_test.go

@@ -95,6 +95,7 @@ func TestExchangePanicRecover(t *testing.T) {
 	it := &partitionPanic{}
 	ex := newExchangeRowIter(ctx, 1, it, nil)
 	ex.start()
+	it.Close()
 
 	require.True(t, it.closed)
 }

sql/plan/innerjoin.go

@@ -355,19 +355,22 @@ func (i *innerJoinIter) Next() (sql.Row, error) {
 	}
 }
 
-func (i *innerJoinIter) Close() error {
-	if err := i.l.Close(); err != nil {
-		if i.r != nil {
-			_ = i.r.Close()
+func (i *innerJoinIter) Close() (err error) {
+	i.right = nil
+
+	if i.l != nil {
+		if err = i.l.Close(); err != nil {
+			if i.r != nil {
+				_ = i.r.Close()
+			}
+			return err
 		}
-		return err
+
 	}
 
 	if i.r != nil {
-		return i.r.Close()
+		err = i.r.Close()
 	}
 
-	i.right = nil
-
-	return nil
+	return err
 }

sql/plan/process.go

@@ -183,9 +183,12 @@ func (i *trackedIndexKeyValueIter) done() {
 	}
 }
 
-func (i *trackedIndexKeyValueIter) Close() error {
+func (i *trackedIndexKeyValueIter) Close() (err error) {
 	i.done()
-	return nil
+	if i.iter != nil {
+		err = i.iter.Close()
+	}
+	return err
 }
 
 func (i *trackedIndexKeyValueIter) Next() ([]interface{}, []byte, error) {