make table_scan as normal instruction

pkg/sql/compile/compile.go

@@ -2218,6 +2218,13 @@ func (c *Compile) compileTableScanWithNode(n *plan.Node, node engine.Node) (*Sco
 	s.DataSource = &Source{
 		node: n,
 	}
+
+	s.appendInstruction(vm.Instruction{
+		Op:      vm.TableScan,
+		Idx:     c.anal.curr,
+		IsFirst: c.anal.isFirst,
+		Arg:     constructTableScan(),
+	})
 	s.Proc = process.NewWithAnalyze(c.proc, c.ctx, 0, c.anal.Nodes())
 	return s, nil
 }

pkg/sql/compile/operator.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/productl2"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/table_scan"
 
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/shufflebuild"
 
@@ -496,6 +497,10 @@ func dupInstruction(sourceIns *vm.Instruction, regMap map[*process.WaitRegister]
 		arg.PkTyp = t.PkTyp
 		arg.BuildIdx = t.BuildIdx
 		res.Arg = arg
+	case vm.TableScan:
+		// t := sourceIns.Arg.(*table_scan.Argument)
+		arg := table_scan.NewArgument()
+		res.Arg = arg
 	default:
 		panic(fmt.Sprintf("unexpected instruction type '%d' to dup", sourceIns.Op))
 	}
@@ -1912,6 +1917,10 @@ func constructJoinCondition(expr *plan.Expr, proc *process.Process) (*plan.Expr,
 	return e.F.Args[0], e.F.Args[1]
 }
 
+func constructTableScan() *table_scan.Argument {
+	return table_scan.NewArgument()
+}
+
 func extraJoinConditions(exprs []*plan.Expr) (*plan.Expr, []*plan.Expr) {
 	exprs = colexec.SplitAndExprs(exprs)
 	eqConds := make([]*plan.Expr, 0, len(exprs))

pkg/sql/compile/scope.go

@@ -31,6 +31,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/right"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/rightanti"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/rightsemi"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/table_scan"
 	"github.com/matrixorigin/matrixone/pkg/sql/util"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
@@ -786,7 +787,11 @@ func (s *Scope) handleRuntimeFilter(c *Compile) error {
 		isFilterOnPK := s.DataSource.TableDef.Pkey != nil && col.Name == s.DataSource.TableDef.Pkey.PkeyColName
 		if !isFilterOnPK {
 			// put expr in filter instruction
-			ins := s.Instructions[0]
+			idx := 0
+			if _, ok := s.Instructions[0].Arg.(*table_scan.Argument); ok {
+				idx = 1
+			}
+			ins := s.Instructions[idx]
 			arg, ok := ins.Arg.(*filter.Argument)
 			if !ok {
 				panic("missing instruction for runtime filter!")

pkg/sql/compile/scopeRemoteRun.go

@@ -22,6 +22,7 @@ import (
 	"unsafe"
 
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/productl2"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/table_scan"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
@@ -1045,6 +1046,8 @@ func convertToPipelineInstruction(opr *vm.Instruction, ctx *scopeContext, ctxId
 			Limit:  t.Limit,
 			Offset: t.Offset,
 		}
+	case *table_scan.Argument:
+		in.TableScan = &pipeline.TableScan{}
 	default:
 		return -1, nil, moerr.NewInternalErrorNoCtx(fmt.Sprintf("unexpected operator: %v", opr.Op))
 	}
@@ -1471,6 +1474,9 @@ func convertToVmInstruction(opr *pipeline.Instruction, ctx *scopeContext, eng en
 		arg.Limit = t.Limit
 		arg.Offset = t.Offset
 		v.Arg = arg
+	case vm.TableScan:
+		arg := table_scan.NewArgument()
+		v.Arg = arg
 	default:
 		return v, moerr.NewInternalErrorNoCtx(fmt.Sprintf("unexpected operator: %v", opr.Op))
 	}

pkg/vm/pipeline/pipeline.go

@@ -66,21 +66,12 @@ func (p *Pipeline) Run(r engine.Reader, topValueMsgTag int32, proc *process.Proc
 		}
 	}
 
-	tableScanOperator := table_scan.Argument{
-		Reader:         r,
-		TopValueMsgTag: topValueMsgTag,
-		Attrs:          p.attrs,
-		TableID:        p.tableID,
+	if tableScanOperator, ok := p.instructions[0].Arg.(*table_scan.Argument); ok {
+		tableScanOperator.Reader = r
+		tableScanOperator.TopValueMsgTag = topValueMsgTag
+		tableScanOperator.Attrs = p.attrs
+		tableScanOperator.TableID = p.tableID
 	}
-	p.instructions = append([]vm.Instruction{
-		{
-			Op:      vm.TableScan,
-			Idx:     -1,
-			Arg:     &tableScanOperator,
-			IsFirst: true,
-			IsLast:  false,
-		},
-	}, p.instructions...)
 
 	if err = vm.Prepare(p.instructions, proc); err != nil {
 		return false, err

proto/pipeline.proto

@@ -363,6 +363,10 @@ message StreamScan {
   int64 limit = 3;
 }
 
+message TableScan {
+
+}
+
 message SampleFunc {
   enum SampleType {
     Rows = 0;
@@ -425,6 +429,7 @@ message Instruction{
   int32 max_parallel = 42;
   IndexJoin index_join = 43;
   ProductL2 product_l2 = 44;
+  TableScan table_scan = 45;
 }
 
 message AnalysisList {