colexecdisk: propagate DiskFull errors as expected

We just saw a sentry report that was issued due to InternalError raised
after Dequeue'ing from a disk queue. It's not clear what the error was
(since it was redacted), but it might have been a DiskFull error. We
already have special handling for it on the Enqueue path, but the
Dequeue path can also trigger this error (on the first call to Dequeue
after some Enqueue calls - in order to flush the buffered batches), so
this commit audits all disk queue methods to use the helper for error
propagation.

The only place where we do disk usage accounting is
`diskQueue.writeFooterAndFlush`, so I traced which methods could end up
calling it (both Enqueue and Dequeue, but also Close) and their call
sites - this is how the affected places were chosen. Additionally,
I didn't want to introduce the error propagation via panics if it wasn't
there already, so one spot wasn't modified.

Release note: None

Author: yuzefovich

pkg/sql/colexec/colexecdisk/external_sort.go

@@ -462,10 +462,10 @@ func (s *externalSorter) Next() coldata.Batch {
 			// resources to be properly released in CloseInactiveReadPartitions
 			// call below.
 			if err := s.partitioner.CloseAllOpenReadFileDescriptors(); err != nil {
-				colexecerror.InternalError(err)
+				colexecutils.HandleErrorFromDiskQueue(err)
 			}
 			if err := s.partitioner.CloseInactiveReadPartitions(s.Ctx); err != nil {
-				colexecerror.InternalError(err)
+				colexecutils.HandleErrorFromDiskQueue(err)
 			}
 			s.state = externalSorterNewPartition
 

pkg/sql/colexec/colexecdisk/hash_based_partitioner.go

@@ -442,7 +442,7 @@ StateChanged:
 				// FDs to the semaphore.
 				for i := range op.inputs {
 					if err := op.partitioners[i].CloseAllOpenWriteFileDescriptors(op.Ctx); err != nil {
-						colexecerror.InternalError(err)
+						colexecutils.HandleErrorFromDiskQueue(err)
 					}
 				}
 				op.inMemMainOp.Init(op.Ctx)
@@ -487,7 +487,7 @@ StateChanged:
 					for {
 						op.unlimitedAllocator.PerformOperation(batch.ColVecs(), func() {
 							if err := partitioner.Dequeue(op.Ctx, parentPartitionIdx, batch); err != nil {
-								colexecerror.InternalError(err)
+								colexecutils.HandleErrorFromDiskQueue(err)
 							}
 						})
 						if batch.Length() == 0 {
@@ -498,7 +498,7 @@ StateChanged:
 					// We're done reading from this partition, and it will never
 					// be read from again, so we can close it.
 					if err := partitioner.CloseInactiveReadPartitions(op.Ctx); err != nil {
-						colexecerror.InternalError(err)
+						colexecutils.HandleErrorFromDiskQueue(err)
 					}
 					// We're done writing to the newly created partitions.
 					// TODO(yuzefovich): we should not release the descriptors
@@ -512,7 +512,7 @@ StateChanged:
 					// want. This will allow us to remove the call to
 					// CloseAllOpen... in the first state as well.
 					if err := partitioner.CloseAllOpenWriteFileDescriptors(op.Ctx); err != nil {
-						colexecerror.InternalError(err)
+						colexecutils.HandleErrorFromDiskQueue(err)
 					}
 				}
 				for idx := 0; idx < op.numBuckets; idx++ {
@@ -594,7 +594,7 @@ StateChanged:
 				// transition to processing new ones.
 				for i := range op.inputs {
 					if err := op.partitioners[i].CloseInactiveReadPartitions(op.Ctx); err != nil {
-						colexecerror.InternalError(err)
+						colexecutils.HandleErrorFromDiskQueue(err)
 					}
 				}
 				op.state = hbpProcessNewPartitionUsingMain
@@ -625,7 +625,7 @@ StateChanged:
 				// transition to processing new ones.
 				for i := range op.inputs {
 					if err := op.partitioners[i].CloseInactiveReadPartitions(op.Ctx); err != nil {
-						colexecerror.InternalError(err)
+						colexecutils.HandleErrorFromDiskQueue(err)
 					}
 				}
 				op.state = hbpProcessNewPartitionUsingFallback

pkg/sql/colexec/colexecdisk/utils.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/col/coldata"
 	"github.com/cockroachdb/cockroach/pkg/sql/colcontainer"
-	"github.com/cockroachdb/cockroach/pkg/sql/colexecerror"
+	"github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecop"
 	"github.com/cockroachdb/cockroach/pkg/sql/colmem"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
@@ -71,7 +71,7 @@ func (p *partitionerToOperator) Next() coldata.Batch {
 		err = p.partitioner.Dequeue(p.Ctx, p.partitionIdx, p.batch)
 	})
 	if err != nil {
-		colexecerror.InternalError(err)
+		colexecutils.HandleErrorFromDiskQueue(err)
 	}
 	return p.batch
 }

pkg/sql/colexec/colexecutils/spilling_buffer.go

@@ -286,7 +286,7 @@ func (b *SpillingBuffer) GetVecWithTuple(
 		var ok bool
 		b.numDequeued += b.dequeueScratch.Length()
 		if ok, err = b.diskQueue.Dequeue(ctx, b.dequeueScratch); err != nil {
-			colexecerror.InternalError(err)
+			HandleErrorFromDiskQueue(err)
 		}
 		if !ok || b.dequeueScratch.Length() == 0 {
 			colexecerror.InternalError(
@@ -303,7 +303,7 @@ func (b *SpillingBuffer) Length() int {
 func (b *SpillingBuffer) closeSpillingQueue(ctx context.Context) {
 	if b.diskQueue != nil {
 		if err := b.diskQueue.Close(ctx); err != nil {
-			colexecerror.InternalError(err)
+			HandleErrorFromDiskQueue(err)
 		}
 		if b.fdSemaphore != nil {
 			b.fdSemaphore.Release(numSpillingBufferFDs)

pkg/sql/colexec/colexecutils/spilling_queue.go

@@ -367,6 +367,9 @@ func (q *SpillingQueue) Dequeue(ctx context.Context) (coldata.Batch, error) {
 		}
 		ok, err := q.diskQueue.Dequeue(ctx, q.dequeueScratch)
 		if err != nil {
+			// TODO(yuzefovich): err here could be DiskFull, but we don't want
+			// to use HandleErrorFromDiskQueue helper since it panics and we do
+			// explicit error propagation from this method.
 			return nil, err
 		}
 		if !ok {