refactor filter related (matrixorigin#18972)

1. remove dummy code
2. reuse reader and filter related code for tae and disttae

Approved by: @triump2020, @LeftHandCold

pkg/vm/engine/disttae/change_handle.go

@@ -26,6 +26,7 @@ import (
 	plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/logtailreplay"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/engine_util"
 )
 
 func (tbl *txnTable) CollectChanges(ctx context.Context, from, to types.TS, mp *mpool.MPool) (engine.ChangesHandle, error) {
@@ -126,9 +127,8 @@ func (h *CheckpointChangesHandle) initReader(ctx context.Context) (err error) {
 	}
 
 	var blockList objectio.BlockInfoSlice
-	if _, err = TryFastFilterBlocks(
+	if _, err = engine_util.TryFastFilterBlocks(
 		ctx,
-		h.table,
 		h.end.ToTimestamp(),
 		tblDef,
 		nil,
@@ -137,7 +137,6 @@ func (h *CheckpointChangesHandle) initReader(ctx context.Context) (err error) {
 		nil,
 		&blockList,
 		h.fs,
-		h.table.proc.Load(),
 	); err != nil {
 		return
 	}

pkg/vm/engine/disttae/engine.go

@@ -684,7 +684,7 @@ func (e *Engine) BuildBlockReaders(
 			shard)
 		rd, err := NewReader(
 			ctx,
-			proc,
+			proc.Mp(),
 			e,
 			def,
 			ts,

pkg/vm/engine/disttae/reader.go

@@ -38,7 +38,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/engine_util"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/blockio"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
 // -----------------------------------------------------------------
@@ -233,7 +232,7 @@ func (r *mergeReader) Read(
 // -----------------------------------------------------------------
 func NewReader(
 	ctx context.Context,
-	proc *process.Process, //it comes from transaction if reader run in local,otherwise it comes from remote compile.
+	mp *mpool.MPool,
 	e *Engine,
 	tableDef *plan.TableDef,
 	ts timestamp.Timestamp,
@@ -245,7 +244,7 @@ func NewReader(
 	baseFilter, err := engine_util.ConstructBasePKFilter(
 		expr,
 		tableDef,
-		proc,
+		mp,
 	)
 	if err != nil {
 		return nil, err

pkg/vm/engine/disttae/tools.go

@@ -33,18 +33,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/txn/trace"
 )
 
-func newColumnExpr(pos int, typ plan.Type, name string) *plan.Expr {
-	return &plan.Expr{
-		Typ: typ,
-		Expr: &plan.Expr_Col{
-			Col: &plan.ColRef{
-				Name:   name,
-				ColPos: int32(pos),
-			},
-		},
-	}
-}
-
 func genWriteReqs(
 	ctx context.Context,
 	txnCommit *Transaction,

pkg/vm/engine/disttae/transfer.go

@@ -31,30 +31,14 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
-	plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan"
 	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/logtailreplay"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/engine_util"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/mergesort"
 )
 
-func ConstructInExpr(
-	ctx context.Context,
-	colName string,
-	colVec *vector.Vector,
-) *plan.Expr {
-	data, _ := colVec.MarshalBinary()
-	colExpr := newColumnExpr(0, plan2.MakePlan2Type(colVec.GetType()), colName)
-	return plan2.MakeInExpr(
-		ctx,
-		colExpr,
-		int32(colVec.Length()),
-		data,
-		false,
-	)
-}
-
 func TransferTombstones(
 	ctx context.Context,
 	table *txnTable,
@@ -345,12 +329,11 @@ func doTransferRowids(
 		v2.BatchTransferTombstonesDurationHistogram.Observe(duration.Seconds())
 	}()
 	pkColumName := table.GetTableDef(ctx).Pkey.PkeyColName
-	expr := ConstructInExpr(ctx, pkColumName, searchPKColumn)
+	expr := engine_util.ConstructInExpr(ctx, pkColumName, searchPKColumn)
 
 	var blockList objectio.BlockInfoSlice
-	if _, err = TryFastFilterBlocks(
+	if _, err = engine_util.TryFastFilterBlocks(
 		ctx,
-		table,
 		table.db.op.SnapshotTS(),
 		table.GetTableDef(ctx),
 		[]*plan.Expr{expr},
@@ -359,7 +342,6 @@ func doTransferRowids(
 		nil,
 		&blockList,
 		fs,
-		table.proc.Load(),
 	); err != nil {
 		return
 	}

pkg/vm/engine/disttae/txn_table.go

@@ -702,9 +702,8 @@ func (tbl *txnTable) rangesOnePart(
 ) (err error) {
 	var done bool
 
-	if done, err = TryFastFilterBlocks(
+	if done, err = engine_util.TryFastFilterBlocks(
 		ctx,
-		tbl,
 		tbl.db.op.SnapshotTS(),
 		tbl.tableDef,
 		exprs,
@@ -713,7 +712,6 @@ func (tbl *txnTable) rangesOnePart(
 		uncommittedObjects,
 		outBlocks,
 		tbl.getTxn().engine.fs,
-		tbl.proc.Load(),
 	); err != nil {
 		return err
 	} else if done {
@@ -1769,7 +1767,7 @@ func (tbl *txnTable) BuildReaders(
 		}
 		rd, err := NewReader(
 			ctx,
-			proc,
+			proc.Mp(),
 			tbl.getTxn().engine,
 			def,
 			tbl.db.op.SnapshotTS(),
@@ -1945,15 +1943,15 @@ func (tbl *txnTable) PKPersistedBetween(
 
 	keys.InplaceSort()
 	bytes, _ := keys.MarshalBinary()
-	colExpr := newColumnExpr(0, plan2.MakePlan2Type(keys.GetType()), tbl.tableDef.Pkey.PkeyColName)
+	colExpr := engine_util.NewColumnExpr(0, plan2.MakePlan2Type(keys.GetType()), tbl.tableDef.Pkey.PkeyColName)
 	inExpr := plan2.MakeInExpr(
 		tbl.proc.Load().Ctx,
 		colExpr,
 		int32(keys.Length()),
 		bytes,
 		false)
 
-	basePKFilter, err := engine_util.ConstructBasePKFilter(inExpr, tbl.tableDef, tbl.proc.Load())
+	basePKFilter, err := engine_util.ConstructBasePKFilter(inExpr, tbl.tableDef, tbl.proc.Load().Mp())
 	if err != nil {
 		return false, err
 	}

pkg/vm/engine/disttae/txn_table_sharding.go

@@ -716,7 +716,7 @@ func (tbl *txnTableDelegate) BuildShardingReaders(
 			}
 			lrd, err := NewReader(
 				ctx,
-				proc,
+				proc.Mp(),
 				tbl.origin.getTxn().engine,
 				tbl.origin.GetTableDef(ctx),
 				tbl.origin.db.op.SnapshotTS(),

pkg/vm/engine/disttae/txn_table_sharding_handle.go

@@ -276,7 +276,7 @@ func HandleShardingReadBuildReader(
 
 	rd, err := NewReader(
 		ctx,
-		tbl.proc.Load(),
+		tbl.proc.Load().Mp(),
 		e.(*Engine),
 		tbl.tableDef,
 		tbl.db.op.SnapshotTS(),