Transfer setup & teardown copy statement aux functions to executor (#999)

* Transfer setup & teardown copy statement aux functions to executor

* Transfer ProcQuery logic to Executor

* fix

Author: reshke

router/frontend/frontend.go

@@ -192,7 +192,7 @@ func Frontend(qr qrouter.QueryRouter, cl client.RouterClient, cmngr poolmgr.Pool
 				// ok
 			default:
 				spqrlog.Zero.Error().
-					Uint("client", rst.Client().ID()).Int("tx-status", int(rst.TxStatus())).Err(err).
+					Uint("client", rst.Client().ID()).Int("tx-status", int(rst.QueryExecutor().TxStatus())).Err(err).
 					Msg("client iteration done with error")
 				if err := rst.UnRouteWithError(rst.ActiveShards(), fmt.Errorf("client processing error: %v, tx status %s", err, rst.TxStatus().String())); err != nil {
 					return err

router/frontend/frontend_test.go

@@ -59,13 +59,17 @@ func TestFrontendSimple(t *testing.T) {
 	qr := mockqr.NewMockQueryRouter(ctrl)
 	cmngr := mockcmgr.NewMockPoolMgr(ctrl)
 
+	mmgr := mockmgr.NewMockEntityMgr(ctrl)
+
 	frrule := &config.FrontendRule{
 		DB:  "db1",
 		Usr: "user1",
 	}
 
 	beRule := &config.BackendRule{}
 
+	qr.EXPECT().Mgr().Return(mmgr).AnyTimes()
+
 	srv.EXPECT().Datashards().AnyTimes().Return([]shard.Shard{})
 	srv.EXPECT().Name().AnyTimes().Return("serv1")
 
@@ -167,6 +171,8 @@ func TestFrontendXProto(t *testing.T) {
 	qr := mockqr.NewMockQueryRouter(ctrl)
 	cmngr := mockcmgr.NewMockPoolMgr(ctrl)
 
+	mmgr := mockmgr.NewMockEntityMgr(ctrl)
+
 	frrule := &config.FrontendRule{
 		DB:       "db1",
 		Usr:      "user1",
@@ -175,6 +181,8 @@ func TestFrontendXProto(t *testing.T) {
 
 	beRule := &config.BackendRule{}
 
+	qr.EXPECT().Mgr().Return(mmgr).AnyTimes()
+
 	sh.EXPECT().ID().AnyTimes()
 	sh.EXPECT().Send(gomock.Any()).AnyTimes()
 	sh.EXPECT().Receive().AnyTimes()

router/qrouter/local.go

@@ -6,6 +6,7 @@ import (
 	"github.com/juju/errors"
 	"github.com/pg-sharding/lyx/lyx"
 	"github.com/pg-sharding/spqr/pkg/config"
+	"github.com/pg-sharding/spqr/pkg/meta"
 	"github.com/pg-sharding/spqr/pkg/models/kr"
 	"github.com/pg-sharding/spqr/pkg/models/topology"
 	"github.com/pg-sharding/spqr/pkg/session"
@@ -88,3 +89,7 @@ func (l *LocalQrouter) DataShardsRoutes() []*kr.ShardKey {
 		},
 	}
 }
+
+func (l *LocalQrouter) Mgr() meta.EntityMgr {
+	return nil
+}

router/relay/phandler.go

@@ -1,19 +1,35 @@
 package relay
 
 import (
+	"context"
+
+	"github.com/jackc/pgx/v5/pgproto3"
+	"github.com/pg-sharding/lyx/lyx"
+	"github.com/pg-sharding/spqr/pkg/meta"
 	"github.com/pg-sharding/spqr/pkg/txstatus"
+	"github.com/pg-sharding/spqr/router/client"
 	"github.com/pg-sharding/spqr/router/parser"
+	"github.com/pg-sharding/spqr/router/pgcopy"
 )
 
 // Execute requered command via
 // some protoc-specific logic
 type QueryStateExecutor interface {
 	txstatus.TxStatusMgr
 
+	Client() client.RouterClient
+
 	ExecBegin(rst RelayStateMgr, query string, st *parser.ParseStateTXBegin) error
 	ExecCommit(rst RelayStateMgr, query string) error
 	ExecRollback(rst RelayStateMgr, query string) error
 
+	/* Copy execution */
+	ProcCopyPrepare(ctx context.Context, mgr meta.EntityMgr, stmt *lyx.Copy) (*pgcopy.CopyState, error)
+	ProcCopy(ctx context.Context, data *pgproto3.CopyData, cps *pgcopy.CopyState) ([]byte, error)
+	ProcCopyComplete(query pgproto3.FrontendMessage) error
+
+	ProcQuery(query pgproto3.FrontendMessage, stmt lyx.Node, mgr meta.EntityMgr, waitForResp bool, replyCl bool) ([]pgproto3.BackendMessage, bool, error)
+
 	ExecSet(rst RelayStateMgr, query, name, value string) error
 	ExecSetLocal(rst RelayStateMgr, query, name, value string) error
 	ExecReset(rst RelayStateMgr, query, name string) error