cleanup 3

Author: andreimatei

pkg/sql/conn_executor.go

@@ -61,10 +61,7 @@ const (
 // (the buffer is not involved in this).
 // The buffer internally maintains a cursor representing the reader's position.
 // The reader has to manually move the cursor using advanceOne(),
-// seekToNextBatch() and rewind(). To support querying the buffer to ask if the
-// consumer is idle (through the ConsumerIdle() method), the contract is that
-// the consumer only advances the cursor after it is done producing results for
-// the previous command.
+// seekToNextBatch() and rewind().
 // In practice, the writer is a module responsible for communicating with a SQL
 // client (i.e. pgwire.conn) and the reader is a connExecutor.
 //
@@ -331,7 +328,7 @@ func (buf *StmtBuf) Push(ctx context.Context, cmd Command) error {
 //
 // If the buffer has previously been Close()d, or is closed while this is
 // blocked, io.EOF is returned.
-func (buf *StmtBuf) curCmd(ctx context.Context) (Command, CmdPos, error) {
+func (buf *StmtBuf) curCmd() (Command, CmdPos, error) {
 	buf.mu.Lock()
 	defer buf.mu.Unlock()
 	for {
@@ -357,14 +354,6 @@ func (buf *StmtBuf) curCmd(ctx context.Context) (Command, CmdPos, error) {
 	}
 }
 
-// ConsumerIdle returns true if the consumer is currently blocked waiting for a
-// command to be pushed into the buffer. See top comments about contract.
-func (buf *StmtBuf) ConsumerIdle() bool {
-	buf.mu.Lock()
-	defer buf.mu.Unlock()
-	return buf.mu.readerBlocked
-}
-
 // translatePosLocked translates an absolute position of a command (counting
 // from the connection start) to the index of the respective command in the
 // buffer (so, it returns an index relative to the start of the buffer).
@@ -400,14 +389,19 @@ func (buf *StmtBuf) ltrim(ctx context.Context, pos CmdPos) {
 		if buf.mu.startPos == pos {
 			break
 		}
+		buf.mu.data[0] = nil
 		buf.mu.data = buf.mu.data[1:]
 		buf.mu.startPos++
 	}
+	// nil out the whole buffer if
+	if len(buf.mu.data) == 0 {
+		buf.mu.data = nil
+	}
 }
 
 // advanceOne advances the cursor one Command over. The command over which the
 // cursor will be positioned when this returns may not be in the buffer yet.
-func (buf *StmtBuf) advanceOne(ctx context.Context) {
+func (buf *StmtBuf) advanceOne() {
 	buf.mu.Lock()
 	buf.mu.curPos++
 	buf.mu.Unlock()
@@ -424,7 +418,7 @@ func (buf *StmtBuf) advanceOne(ctx context.Context) {
 //
 // It is an error to start seeking when the cursor is positioned on an empty
 // slot.
-func (buf *StmtBuf) seekToNextBatch(ctx context.Context) error {
+func (buf *StmtBuf) seekToNextBatch() error {
 	buf.mu.Lock()
 	curPos := buf.mu.curPos
 	cmdIdx, err := buf.translatePosLocked(curPos)
@@ -440,8 +434,8 @@ func (buf *StmtBuf) seekToNextBatch(ctx context.Context) error {
 
 	var foundSync bool
 	for !foundSync {
-		buf.advanceOne(ctx)
-		_, pos, err := buf.curCmd(ctx)
+		buf.advanceOne()
+		_, pos, err := buf.curCmd()
 		if err != nil {
 			return err
 		}
@@ -537,29 +531,33 @@ type ClientComm interface {
 }
 
 // CommandResult represents the result of a statement. It which needs to be
-// ultimately delivered to the client. The pgwire module implements this.
+// ultimately delivered to the client. pgwire.conn implements this.
 type CommandResult interface {
 	RestrictedCommandResult
+	CommandResultClose
 
 	// SetLimit is used when executing a portal to set a limit on the number of
 	// rows to be returned. We don't currently properly support this feature of
 	// the Postgres protocol; instead, we'll return an error if the number of rows
 	// produced is larger than this limit.
 	SetLimit(n int)
+}
 
-	// Close - see ResultBase.
-	Close(TransactionStatusIndicator)
-
-	// CloseWithErr - see ResultBase.
-	CloseWithErr(error)
+type CommandResultErrBase interface {
+	// SetError accumulates an execution error that needs to be reported to the
+	// client. No further calls other than Close() and Discard() are allowed. In
+	// particular, CloseWithErr() is not allowed.
+	SetError(error)
 
-	// Discard - see ResultBase.
-	Discard()
+	// Err returns the error previously set with SetError(), if any.
+	Err() error
 }
 
 // RestrictedCommandResult is a subset of CommandResult meant to make it clear
 // that its clients don't close the CommandResult.
 type RestrictedCommandResult interface {
+	CommandResultErrBase
+
 	// SetColumns informs the client about the schema of the result. The columns
 	// can be nil.
 	//
@@ -585,14 +583,6 @@ type RestrictedCommandResult interface {
 	// value. The callback can be nil, in which case nothing will be called.
 	SetFinishedCallback(callback func())
 
-	// SetError accumulates an execution error that needs to be reported to the
-	// client. No further calls other that Err() and Close()/Discard() are allowed
-	// - in particular, CloseWithErr() is not allowed.
-	SetError(error)
-
-	// Err returns the error previously set with SetError(), if any.
-	Err() error
-
 	// IncrementRowsAffected increments a counter by n. This is used for all
 	// result types other than tree.Rows.
 	IncrementRowsAffected(n int)
@@ -662,14 +652,11 @@ type CopyInResult interface {
 // ResultBase is the common interface implemented by all the different command
 // results.
 type ResultBase interface {
-	// SetError accumulates an execution error that needs to be reported to the
-	// client. No further calls other than Close() and Discard() are allowed. In
-	// particular, CloseWithErr() is not allowed.
-	SetError(error)
-
-	// Err returns the error previously set with SetError(), if any.
-	Err() error
+	CommandResultErrBase
+	CommandResultClose
+}
 
+type CommandResultClose interface {
 	// Close marks a result as complete. All results must be eventually closed
 	// through Close()/CloseWithErr()/Discard. No further uses of the CommandResult are
 	// allowed.

pkg/sql/conn_io.go

@@ -81,7 +81,7 @@ func TestStmtBuf(t *testing.T) {
 	// same statement.
 	expPos := CmdPos(0)
 	for i := 0; i < 2; i++ {
-		cmd, pos, err := buf.curCmd(ctx)
+		cmd, pos, err := buf.curCmd()
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -91,9 +91,9 @@ func TestStmtBuf(t *testing.T) {
 		assertStmt(t, cmd, "SELECT 1")
 	}
 
-	buf.advanceOne(ctx)
+	buf.advanceOne()
 	expPos++
-	cmd, pos, err := buf.curCmd(ctx)
+	cmd, pos, err := buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -102,9 +102,9 @@ func TestStmtBuf(t *testing.T) {
 	}
 	assertStmt(t, cmd, "SELECT 2")
 
-	buf.advanceOne(ctx)
+	buf.advanceOne()
 	expPos++
-	cmd, pos, err = buf.curCmd(ctx)
+	cmd, pos, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -113,9 +113,9 @@ func TestStmtBuf(t *testing.T) {
 	}
 	assertStmt(t, cmd, "SELECT 3")
 
-	buf.advanceOne(ctx)
+	buf.advanceOne()
 	expPos++
-	cmd, pos, err = buf.curCmd(ctx)
+	cmd, pos, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -127,7 +127,7 @@ func TestStmtBuf(t *testing.T) {
 	// Now rewind.
 	expPos = 1
 	buf.rewind(ctx, expPos)
-	cmd, pos, err = buf.curCmd(ctx)
+	cmd, pos, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -153,7 +153,7 @@ func TestStmtBufSignal(t *testing.T) {
 	}()
 
 	expPos := CmdPos(0)
-	cmd, pos, err := buf.curCmd(ctx)
+	cmd, pos, err := buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -177,8 +177,8 @@ func TestStmtBufLtrim(t *testing.T) {
 		mustPush(ctx, t, buf, ExecStmt{Stmt: stmt})
 	}
 	// Advance the cursor so that we can trim.
-	buf.advanceOne(ctx)
-	buf.advanceOne(ctx)
+	buf.advanceOne()
+	buf.advanceOne()
 	trimPos := CmdPos(2)
 	buf.ltrim(ctx, trimPos)
 	if l := len(buf.mu.data); l != 3 {
@@ -203,7 +203,7 @@ func TestStmtBufClose(t *testing.T) {
 	mustPush(ctx, t, buf, ExecStmt{Stmt: stmt})
 	buf.Close()
 
-	_, _, err = buf.curCmd(ctx)
+	_, _, err = buf.curCmd()
 	if err != io.EOF {
 		t.Fatalf("expected EOF, got: %v", err)
 	}
@@ -213,14 +213,13 @@ func TestStmtBufClose(t *testing.T) {
 func TestStmtBufCloseUnblocksReader(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
-	ctx := context.TODO()
 	buf := NewStmtBuf()
 
 	go func() {
 		buf.Close()
 	}()
 
-	_, _, err := buf.curCmd(ctx)
+	_, _, err := buf.curCmd()
 	if err != io.EOF {
 		t.Fatalf("expected EOF, got: %v", err)
 	}
@@ -242,29 +241,29 @@ func TestStmtBufPreparedStmt(t *testing.T) {
 	mustPush(ctx, t, buf, PrepareStmt{Name: "p1"})
 	mustPush(ctx, t, buf, PrepareStmt{Name: "p2"})
 
-	cmd, _, err := buf.curCmd(ctx)
+	cmd, _, err := buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
 	assertStmt(t, cmd, "SELECT 1")
 
-	buf.advanceOne(ctx)
-	cmd, _, err = buf.curCmd(ctx)
+	buf.advanceOne()
+	cmd, _, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
 	assertPrepareStmt(t, cmd, "p1")
 
-	buf.advanceOne(ctx)
-	cmd, _, err = buf.curCmd(ctx)
+	buf.advanceOne()
+	cmd, _, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
 	assertPrepareStmt(t, cmd, "p2")
 
 	// Rewind to the first prepared stmt.
 	buf.rewind(ctx, CmdPos(1))
-	cmd, _, err = buf.curCmd(ctx)
+	cmd, _, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -301,10 +300,10 @@ func TestStmtBufBatching(t *testing.T) {
 	mustPush(ctx, t, buf, ExecStmt{Stmt: s1})
 
 	// Go to 2nd batch.
-	if err := buf.seekToNextBatch(ctx); err != nil {
+	if err := buf.seekToNextBatch(); err != nil {
 		t.Fatal(err)
 	}
-	_, pos, err := buf.curCmd(ctx)
+	_, pos, err := buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -313,10 +312,10 @@ func TestStmtBufBatching(t *testing.T) {
 	}
 
 	// Go to 3rd batch.
-	if err := buf.seekToNextBatch(ctx); err != nil {
+	if err := buf.seekToNextBatch(); err != nil {
 		t.Fatal(err)
 	}
-	_, pos, err = buf.curCmd(ctx)
+	_, pos, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -331,10 +330,10 @@ func TestStmtBufBatching(t *testing.T) {
 	}()
 
 	// Go to 4th batch.
-	if err := buf.seekToNextBatch(ctx); err != nil {
+	if err := buf.seekToNextBatch(); err != nil {
 		t.Fatal(err)
 	}
-	_, pos, err = buf.curCmd(ctx)
+	_, pos, err = buf.curCmd()
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/sql/conn_io_test.go

@@ -2262,6 +2262,7 @@ func (e *Executor) execStmtInParallel(
 	return cols, nil
 }
 
+// Cfg returns a reference to the ExecutorConfig.
 func (e *Executor) Cfg() *ExecutorConfig {
 	return &e.cfg
 }