Repository: matrixorigin/matrixone
Pull request #4581 (2022-08-19 00:50:47)

Implement general Mark Join
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #4153

## What this PR does / why we need it:

nnsgmsone: Realization of the existence of some problems.... Needs to be modified.
nnsgmsone: Please handle multiple columns
JackTan25: > Test, test test.
> 
> All new operators, should have a bvt test that triggers this operator. explain and confirm this operator is activated.

ok, I got it
JackTan25: > Test, test test.
> 
> All new operators, should have a bvt test that triggers this operator. explain and confirm this operator is activated.

I have add mark join, please wait Jensen to add multi col feature.
JackTan25: > Test, test test.
> 
> All new operators, should have a bvt test that triggers this operator. explain and confirm this operator is activated.

I‘ve added the mark-join bvt test. Now please wait jensen to finish the multiple columns feature.
JackTan25: > Can you add a explain call on bvt test case and grep the result to see if it is really a mark join ...

For now plan, if join result is marked as null, we don't output.

Patch:
diff --git a/pkg/container/batch/batch.go b/pkg/container/batch/batch.go
index 7d87a388e..a7e842262 100644
--- a/pkg/container/batch/batch.go
+++ b/pkg/container/batch/batch.go
@@ -180,6 +180,7 @@ func (bat *Batch) Clean(m *mheap.Mheap) {
 	}
 	if len(bat.Zs) != 0 {
 		m.PutSels(bat.Zs)
+		bat.Zs = nil
 	}
 	bat.Vecs = nil
 }
diff --git a/pkg/sql/colexec/mark/join.go b/pkg/sql/colexec/mark/join.go
new file mode 100644
index 000000000..c3f2587f8
--- /dev/null
+++ b/pkg/sql/colexec/mark/join.go
@@ -0,0 +1,472 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mark
+
+import (
+	"bytes"
+	"unsafe"
+
+	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/nulls"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
+	"github.com/matrixorigin/matrixone/pkg/sql/plan"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+)
+
+func String(_ any, buf *bytes.Buffer) {
+	buf.WriteString(" mark join ")
+}
+
+func Prepare(proc *process.Process, arg any) error {
+	ap := arg.(*Argument)
+	ap.ctr = new(container)
+	ap.ctr.inBuckets = make([]uint8, hashmap.UnitLimit)
+	ap.ctr.evecs = make([]evalVector, len(ap.Conditions[0]))
+	ap.ctr.vecs = make([]*vector.Vector, len(ap.Conditions[0]))
+	ap.ctr.buildEqVec = make([]*vector.Vector, len(ap.Conditions[1]))
+	ap.ctr.buildEqEvecs = make([]evalVector, len(ap.Conditions[1]))
+	return nil
+}
+
+// Note: before mark join, right table has been used in hashbuild operator to build JoinMap, which only contains those tuples without null
+// the idx of tuples contains null is stored in nullSels
+
+// 1. for each tuple in left table, join with tuple(s) in right table based on Three-valued logic. Conditions may contain equal conditions and non-equal conditions
+// logic state for same row is Three-valued AND, for different rows is Three-valued OR
+
+// 2.1 if a probe tuple has null(i.e. zvals[k] == 0)
+//       scan whole right table directly and join with each tuple to determine state
+
+// 2.2 if a probe tuple has no null. then scan JoinMap firstly to check equal condtions.(condEq)
+//	    2.2.1 if condEq is condtrue in JoinMap(i.e. vals[k] > 0)
+//	 		    further check non-eq condtions in those tupe IN JoinMap
+//				2.2.1.1 if condNonEq is condTrue
+//						   mark as condTrue
+//	 	        2.2.1.2 if condNonEq is condUnkown
+//						   mark as condUnkown
+//	 	        2.2.1.3 if condNonEq is condFalse in JoinMap
+//						   further check eq and non-eq conds IN nullSels
+//                         (probe state could still be unknown BUT NOT FALSE as long as one unknown state exists, so have to scan the whole right table)
+
+//	    2.2.2 if condEq is condFalse in JoinMap
+//				check eq and non-eq conds in nullSels to determine condState. (same as 2.2.1.3)
+
+func Call(idx int, proc *process.Process, arg any) (bool, error) {
+	anal := proc.GetAnalyze(idx)
+	anal.Start()
+	defer anal.Stop()
+	ap := arg.(*Argument)
+	ctr := ap.ctr
+	for {
+		switch ctr.state {
+		case Build:
+			if err := ctr.build(ap, proc, anal); err != nil {
+				ctr.state = End
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
+				return true, err
+			}
+			ctr.state = Probe
+		case Probe:
+			bat := <-proc.Reg.MergeReceivers[0].Ch
+			if bat == nil {
+				ctr.state = End
+				ctr.freeBuildEqVec(proc)
+				if ctr.nullWithBatch != nil {
+					ctr.nullWithBatch.Clean(proc.GetMheap())
+				}
+				if ctr.mp != nil {
+					ctr.mp.Free()
+				}
+				if ctr.bat != nil {
+					ctr.bat.Clean(proc.GetMheap())
+				}
+				continue
+			}
+			if bat.Length() == 0 {
+				continue
+			}
+			if ctr.bat == nil || ctr.bat.Length() == 0 {
+				if err := ctr.emptyProbe(bat, ap, proc, anal); err != nil {
+					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
+					proc.SetInputBatch(nil)
+					return true, err
+				}
+			} else {
+				if err := ctr.probe(bat, ap, proc, anal); err != nil {
+					ctr.freeBuildEqVec(proc)
+					if ctr.nullWithBatch != nil {
+						ctr.nullWithBatch.Clean(proc.GetMheap())
+					}
+					ctr.state = End
+					if ctr.mp != nil {
+						ctr.mp.Free()
+					}
+					if ctr.bat != nil {
+						ctr.bat.Clean(proc.GetMheap())
+					}
+					proc.SetInputBatch(nil)
+					return true, err
+				}
+			}
+			return false, nil
+		default:
+			proc.SetInputBatch(nil)
+			return true, nil
+		}
+	}
+}
+
+func (ctr *container) build(ap *Argument, proc *process.Process, anal process.Analyze) error {
+	bat := <-proc.Reg.MergeReceivers[1].Ch
+	if bat != nil {
+		joinMap := bat.Ht.(*hashmap.JoinMap)
+		ctr.evalNullSels(bat)
+		ctr.nullWithBatch = DumpBatch(bat, proc, ctr.nullSels)
+		if err := ctr.evalJoinBuildCondition(bat, ap.Conditions[1], proc); err != nil {
+			return err
+		}
+		ctr.rewriteCond = colexec.RewriteFilterExprList(ap.OnList)
+		ctr.bat = bat
+		ctr.mp = joinMap.Dup()
+		ctr.hasNull = ctr.mp.HasNull()
+	}
+	return nil
+}
+
+func (ctr *container) emptyProbe(bat *batch.Batch, ap *Argument, proc *process.Process, anal process.Analyze) error {
+	defer bat.Clean(proc.GetMheap())
+	anal.Input(bat)
+	rbat := batch.NewWithSize(len(ap.Result) + 1)
+	rbat.Zs = proc.GetMheap().GetSels()
+	for i, pos := range ap.Result {
+		rbat.Vecs[i] = vector.New(bat.Vecs[pos].Typ)
+	}
+	rbat.Vecs[len(ap.Result)] = vector.New(types.T_bool.ToType())
+	ctr.joinFlags = make([]bool, bat.Length())
+	if ap.OutputMark {
+		ctr.Nsp = nulls.NewWithSize(bat.Length())
+		data := unsafe.Slice((*byte)(unsafe.Pointer(&ctr.joinFlags[0])), cap(ctr.joinFlags))[:len(ctr.joinFlags)]
+		// add mark flag, the initial
+		rbat.Vecs[len(ap.Result)] = vector.NewWithData(types.T_bool.ToType(), data, ctr.joinFlags, ctr.Nsp)
+	}
+	count := bat.Length()
+	for i := 0; i < count; i += hashmap.UnitLimit {
+		n := count - i
+		if n > hashmap.UnitLimit {
+			n = hashmap.UnitLimit
+		}
+		for k := 0; k < n; k++ {
+			if ap.MarkMeaning == ctr.joinFlags[i+k] {
+				for j, pos := range ap.Result {
+					if err := vector.UnionOne(rbat.Vecs[j], bat.Vecs[pos], int64(i+k), proc.GetMheap()); err != nil {
+						rbat.Clean(proc.GetMheap())
+						return err
+					}
+				}
+				rbat.Zs = append(rbat.Zs, bat.Zs[i+k])
+			}
+		}
+	}
+	if !ap.OutputMark {
+		rbat.Vecs = rbat.Vecs[:len(rbat.Vecs)-1]
+	}
+	rbat.ExpandNulls()
+	anal.Output(rbat)
+	proc.SetInputBatch(rbat)
+	return nil
+}
+
+func (ctr *container) probe(bat *batch.Batch, ap *Argument, proc *process.Process, anal process.Analyze) error {
+	defer bat.Clean(proc.Mp)
+	anal.Input(bat)
+	rbat := batch.NewWithSize(len(ap.Result) + 1)
+	// vector.UnionBatch()
+	rbat.Zs = proc.GetMheap().GetSels()
+	for i, pos := range ap.Result {
+		rbat.Vecs[i] = vector.New(bat.Vecs[pos].Typ)
+	}
+	lastIndex := len(rbat.Vecs) - 1
+	rbat.Vecs[lastIndex] = vector.New(types.T_bool.ToType())
+	ctr.joinFlags = make([]bool, bat.Length())
+	ctr.Nsp = nulls.NewWithSize(bat.Length())
+	if err := ctr.evalJoinProbeCondition(bat, ap.Conditions[0], proc); err != nil {
+		return err
+	}
+	defer ctr.freeProbeEqVec(proc)
+	count := bat.Length()
+	itr := ctr.mp.Map().NewIterator()
+	mSels := ctr.mp.Sels()
+	for i := 0; i < count; i += hashmap.UnitLimit {
+		n := count - i
+		if n > hashmap.UnitLimit {
+			n = hashmap.UnitLimit
+		}
+		copy(ctr.inBuckets, hashmap.OneUInt8s)
+		vals, zvals := itr.Find(i, n, ctr.vecs, ctr.inBuckets)
+		var condState resultType
+		// var condNonEq resultType
+		// var condEq resultType
+		var err error
+		for k := 0; k < n; k++ {
+			if ctr.inBuckets[k] == 0 {
+				continue
+			}
+			if zvals[k] == 0 { // 2.1 : probe tuple has null
+				condState, err := ctr.EvalEntire(bat, ctr.bat, i+k, proc, ctr.rewriteCond)
+				if err != nil {
+					return err
+				}
+				ctr.handleResultType(i+k, condState)
+			} else if vals[k] > 0 { // 2.2.1 : condEq is condTrue in JoinMap
+				condState, err = ctr.nonEqJoinInMap(ap, mSels, vals, k, i, proc, bat)
+				if err != nil {
+					rbat.Clean(proc.GetMheap())
+					return err
+				}
+				if condState == condTrue { // 2.2.1.1 : condNonEq is condTrue in JoinMap
+					ctr.joinFlags[i+k] = true
+				} else if condState == condUnkown { // 2.2.1.2 : condNonEq is condUnkown in JoinMap
+					ctr.Nsp.Np.Add(uint64(i + k))
+				} else { // 2.2.1.3 : condNonEq is condFalse in JoinMap, further check in nullSels
+					if len(ctr.nullSels) == 0 {
+						ctr.handleResultType(i+k, condFalse)
+						continue
+					}
+					condState, err = ctr.EvalEntire(bat, ctr.nullWithBatch, i+k, proc, ctr.rewriteCond)
+					if err != nil {
+						rbat.Clean(proc.GetMheap())
+						return err
+					}
+					ctr.handleResultType(i+k, condState)
+				}
+			} else { // 2.2.2 : condEq in condFalse in JoinMap, further check in nullSels
+				if len(ctr.nullSels) == 0 {
+					ctr.handleResultType(i+k, condFalse)
+					continue
+				}
+				condState, err = ctr.EvalEntire(bat, ctr.nullWithBatch, i+k, proc, ctr.rewriteCond)
+				if err != nil {
+					rbat.Clean(proc.GetMheap())
+					return err
+				}
+				ctr.handleResultType(i+k, condState)
+			}
+		}
+		data := unsafe.Slice((*byte)(unsafe.Pointer(&ctr.joinFlags[0])), cap(ctr.joinFlags))[:len(ctr.joinFlags)]
+		// add mark flag, the initial
+		rbat.Vecs[len(ap.Result)] = vector.NewWithData(types.T_bool.ToType(), data, ctr.joinFlags, ctr.Nsp)
+		markVec := vector.NewWithData(types.T_bool.ToType(), data, ctr.joinFlags, ctr.Nsp)
+		for k := 0; k < n; k++ {
+			if ap.OutputAnyway || (ctr.Nsp.Np.Contains(uint64(i+k)) && ap.OutputNull || !ctr.Nsp.Np.Contains(uint64(i+k)) && ctr.joinFlags[i+k] == ap.MarkMeaning) {
+				for j, pos := range ap.Result {
+					if err := vector.UnionOne(rbat.Vecs[j], bat.Vecs[pos], int64(i+k), proc.GetMheap()); err != nil {
+						rbat.Clean(proc.GetMheap())
+						return err
+					}
+				}
+				if ap.OutputMark {
+					if err := vector.UnionOne(rbat.Vecs[lastIndex], markVec, int64(i+k), proc.GetMheap()); err != nil {
+						rbat.Clean(proc.GetMheap())
+						return err
+					}
+				}
+				rbat.Zs = append(rbat.Zs, bat.Zs[i+k])
+			}
+		}
+	}
+	if !ap.OutputMark {
+		rbat.Vecs = rbat.Vecs[:len(rbat.Vecs)-1]
+	}
+	rbat.ExpandNulls()
+	anal.Output(rbat)
+	proc.SetInputBatch(rbat)
+	return nil
+}
+
+// store the results of the calculation on the probe side of the equation condition
+func (ctr *container) evalJoinProbeCondition(bat *batch.Batch, conds []*plan.Expr, proc *process.Process) error {
+	for i, cond := range conds {
+		vec, err := colexec.EvalExpr(bat, proc, cond)
+		if err != nil || vec.ConstExpand(proc.GetMheap()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.GetMheap())
+				}
+			}
+			return err
+		}
+		ctr.vecs[i] = vec
+		ctr.evecs[i].vec = vec
+		ctr.evecs[i].needFree = true
+		for j := range bat.Vecs {
+			if bat.Vecs[j] == vec {
+				ctr.evecs[i].needFree = false
+				break
+			}
+		}
+	}
+	return nil
+}
+
+// store the results of the calculation on the build side of the equation condition
+func (ctr *container) evalJoinBuildCondition(bat *batch.Batch, conds []*plan.Expr, proc *process.Process) error {
+	for i, cond := range conds {
+		vec, err := colexec.EvalExpr(bat, proc, cond)
+		if err != nil || vec.ConstExpand(proc.GetMheap()) == nil {
+			for j := 0; j < i; j++ {
+				if ctr.evecs[j].needFree {
+					vector.Clean(ctr.evecs[j].vec, proc.GetMheap())
+				}
+			}
+			return err
+		}
+		ctr.buildEqVec[i] = vec
+		ctr.buildEqEvecs[i].vec = vec
+		ctr.buildEqEvecs[i].needFree = true
+		for j := range bat.Vecs {
+			if bat.Vecs[j] == vec {
+				ctr.buildEqEvecs[i].needFree = false
+				break
+			}
+		}
+	}
+	return nil
+}
+
+func (ctr *container) freeProbeEqVec(proc *process.Process) {
+	for i := range ctr.evecs {
+		if ctr.evecs[i].needFree {
+			ctr.evecs[i].vec.Free(proc.GetMheap())
+		}
+	}
+}
+
+func (ctr *container) freeBuildEqVec(proc *process.Process) {
+	for i := range ctr.buildEqEvecs {
+		if ctr.buildEqEvecs[i].needFree {
+			ctr.buildEqEvecs[i].vec.Free(proc.GetMheap())
+		}
+	}
+}
+
+// calculate the state of non-equal conditions for those tuples in JoinMap
+func (ctr *container) nonEqJoinInMap(ap *Argument, mSels [][]int64, vals []uint64, k int, i int, proc *process.Process, bat *batch.Batch) (resultType, error) {
+	if ap.Cond != nil {
+		condState := condFalse
+		sels := mSels[vals[k]-1]
+		for _, sel := range sels {
+			vec, err := colexec.JoinFilterEvalExprInBucket(bat, ctr.bat, i+k, int(sel), proc, ap.Cond)
+			if err != nil {
+				return condUnkown, err
+			}
+			if vec.Nsp.Contains(0) {
+				condState = condUnkown
+			}
+			bs := vec.Col.([]bool)
+			if bs[0] {
+				condState = condTrue
+				vec.Free(proc.Mp)
+				break
+			}
+			vec.Free(proc.Mp)
+		}
+		return condState, nil
+	} else {
+		return condTrue, nil
+	}
+}
+
+func (ctr *container) EvalEntire(pbat, bat *batch.Batch, idx int, proc *process.Process, cond *plan.Expr) (resultType, error) {
+	if cond == nil {
+		return condTrue, nil
+	}
+	vec, err := colexec.JoinFilterEvalExpr(pbat, bat, idx, proc, cond)
+	defer vec.Free(proc.GetMheap())
+	if err != nil {
+		return condUnkown, err
+	}
+	bs := vec.Col.([]bool)
+	for _, b := range bs {
+		if b {
+			return condTrue, nil
+		}
+	}
+	if nulls.Any(vec.Nsp) {
+		return condUnkown, nil
+	}
+	return condFalse, nil
+}
+
+// collect the idx of tuple which contains null values
+func (ctr *container) evalNullSels(bat *batch.Batch) {
+	joinMap := bat.Ht.(*hashmap.JoinMap)
+	jmSels := joinMap.Sels()
+	selsMap := make(map[int64]bool)
+	for _, sel := range jmSels {
+		for _, i := range sel {
+			selsMap[i] = true
+		}
+	}
+	var nullSels []int64
+	for i := 0; i < bat.Length(); i++ {
+		if selsMap[int64(i)] {
+			ctr.sels = append(ctr.sels, int64(i))
+			continue
+		}
+		nullSels = append(nullSels, int64(i))
+	}
+	ctr.nullSels = nullSels
+}
+
+// mark probe tuple state
+func (ctr *container) handleResultType(idx int, r resultType) {
+	switch r {
+	case condTrue:
+		ctr.joinFlags[idx] = true
+	case condFalse:
+		ctr.joinFlags[idx] = false
+	case condUnkown:
+		ctr.Nsp.Np.Add(uint64(idx))
+	}
+}
+
+func DumpBatch(originBatch *batch.Batch, proc *process.Process, sels []int64) *batch.Batch {
+	length := originBatch.Length()
+	flags := make([]uint8, length)
+	for _, sel := range sels {
+		flags[sel] = 1
+	}
+	bat := batch.NewWithSize(len(originBatch.Vecs))
+	for i, vec := range originBatch.Vecs {
+		bat.Vecs[i] = vector.New(vec.GetType())
+	}
+	if len(sels) == 0 {
+		return bat
+	}
+	for i, vec := range originBatch.Vecs {
+		vector.UnionBatch(bat.Vecs[i], vec, 0, length, flags, proc.GetMheap())
+	}
+
+	bat.Zs = append(bat.Zs, originBatch.Zs...)
+	return bat
+}
diff --git a/pkg/sql/colexec/mark/types.go b/pkg/sql/colexec/mark/types.go
new file mode 100644
index 000000000..5b8f3682f
--- /dev/null
+++ b/pkg/sql/colexec/mark/types.go
@@ -0,0 +1,162 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mark
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/common/hashmap"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/nulls"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/sql/plan"
+)
+
+const (
+	Build = iota
+	Probe
+	End
+)
+
+type resultType int
+
+const (
+	condFalse resultType = iota
+	condTrue
+	condUnkown
+)
+
+type evalVector struct {
+	// In MO,for example, select a from t1 where a+1 > 1, col a will be transformed as a vector,
+	// when it comes to be accepted by the operator,will firstly calculate the a+1, so it's up to
+	// the eval implementor whether he store the a+1 result in the old col a vector or create a
+	// new vector to store it.
+	needFree bool
+	vec      *vector.Vector
+}
+
+// note that, different from other joins,the result vector is like below:
+// Result[0],Result[1],......,Result[n-1],bool
+// we will give more one bool type vector as a marker col
+// so if you use mark join result, remember to get the last vector,that's what you want
+type container struct {
+	// here, we will have three states:
+	// Build：we will use the right table to build a hashtable
+	// Probe: we will use the left table data to probe the hashtable
+	// End: Join working is over
+	state int
+
+	// // sels[i] records the table used to build hash table has which row numbers in the i-th group
+	// sels [][]int64
+
+	// in the probe stage, when we invoke func find to find rows in the hashtable,it will modify the
+	// inBuckets Slice, inBuckets[i] means the i-th row is whether in the bucket
+	// 0 means no, 1 means yes
+	inBuckets []uint8
+
+	// store the all batch from the build table
+	bat *batch.Batch
+	// records the eval result of the batch from the probe table
+	evecs []evalVector
+	// vecs is same as the evecs.vec's union, we need this because
+	// when we use the Insert func to build the hashtable, we need
+	// vecs not evecs
+	vecs []*vector.Vector
+
+	// record those tuple that contain null value in build table
+	nullSels []int64
+	// record those tuple that contain normal value in build table
+	sels []int64
+	// the result of eval join condtion for conds[1] and cons[0], those two vectors is used to
+	// check equal condition when zval == 0 or condState is False from JoinMap
+	buildEqVec   []*vector.Vector
+	buildEqEvecs []evalVector
+
+	// a tag means wheretr the build table has null
+	hasNull bool
+
+	Nsp       *nulls.Nulls
+	joinFlags []bool
+
+	mp *hashmap.JoinMap
+
+	nullWithBatch *batch.Batch
+
+	rewriteCond *plan.Expr
+}
+
+// // for join operator, it's a two-ary operator, we will reference to two table
+// // so we need this info to determine the columns to output
+// // (rel,pos) gives which table and which column
+// type ResultPos struct {
+// 	Rel int32
+// 	Pos int32
+// }
+
+// remember that we may use partition stragey, for example, if the origin table has data squence
+// like 1,2,3,4. If we use the hash method, after using hash function,assume that we get 13,14,15,16.
+// and we divide them into 3 buckets. so 13%3 = 1,so 3 is in the 1-th bucket and so on like this
+type Argument struct {
+	// container means the local parameters defined by the operator constructor
+	ctr *container
+	// the five attributes below are passed by the outside
+
+	// Ibucket determines the data partition this operator need to deal with
+	Ibucket uint64
+	// Nbucket means how many partitions there are
+	Nbucket uint64
+
+	// // the input batch's columns' type
+	// Typs []types.Type
+
+	// records the result cols' position that the build table needs to return
+	Result []int32
+	// because we have two tables here,so it's a slice
+	// Conditions[i] stands for the table_i's expression
+	// we ned this to eval first, if there is a expression
+	// like  t.a+1, we will eval it first and then to build
+	// hashtable and probe to get the result. note that,though
+	// we have two tables, but len(Conditions[0]) == len(Conditions[1])
+	// for example, select * from t1 join t2 on t1.a = t2.d+t2.e+t2.f;
+	// We will get Condition below:
+	// for t1: Expr_Col --> t1.a
+	// for t2: Expr_F(arg0,arg1)
+	// and the arg0 is Expr_F(t2.d,t2.e)
+	// and the arg1 is Expr_Col --> t2.f
+	// so from the view of above, the len of two Conditions is the same
+	// then you will see I just make the evals with len(Conditions[0]),but
+	// I will use evalJoinConditions with parameters Conditions[0] or Conditions[1]
+	// they are both ok
+	Conditions [][]*plan.Expr
+
+	// tell whether the tuple marked with NULL should be outputed
+	OutputNull bool
+
+	// tell whether mark col should be outputed
+	OutputMark bool
+
+	Typs []types.Type
+
+	// markMeaning means,if MarkMeaning is true and a tuple is marked as true, output it,if is marked with false,
+	// don't output. The same way that if MarkMeaning is false and a tuple is marked as false, output it,if is
+	// marked with true, don't output.
+	MarkMeaning bool
+
+	// if OutputAnyway is true,we will output all tuples from left table
+	OutputAnyway bool
+
+	Cond *plan.Expr
+
+	OnList []*plan.Expr
+}
diff --git a/pkg/sql/compile/compile.go b/pkg/sql/compile/compile.go
index 5a0208eeb..f6156ca31 100644
--- a/pkg/sql/compile/compile.go
+++ b/pkg/sql/compile/compile.go
@@ -702,12 +702,20 @@ func (c *Compile) compileJoin(n, right *plan.Node, ss []*Scope, children []*Scop
 	case plan.Node_ANTI:
 		_, conds := extraJoinConditions(n.OnList)
 		for i := range rs {
-			if isEq && len(conds) == 1 {
-				rs[i].appendInstruction(vm.Instruction{
-					Op:  vm.Anti,
-					Idx: c.anal.curr,
-					Arg: constructAnti(n, typs, c.proc),
-				})
+			if isEq {
+				if len(conds) > 1 {
+					rs[i].appendInstruction(vm.Instruction{
+						Op:  vm.Mark,
+						Idx: c.anal.curr,
+						Arg: constructMark(n, typs, c.proc, n.OnList),
+					})
+				} else {
+					rs[i].appendInstruction(vm.Instruction{
+						Op:  vm.Anti,
+						Idx: c.anal.curr,
+						Arg: constructAnti(n, typs, c.proc),
+					})
+				}
 			} else {
 				rs[i].appendInstruction(vm.Instruction{
 					Op:  vm.LoopAnti,
@@ -1076,6 +1084,8 @@ func joinType(n *plan.Node, ns []*plan.Node) (bool, plan.Node_JoinFlag) {
 		return true, plan.Node_LEFT
 	case plan.Node_SINGLE:
 		return false, plan.Node_SINGLE
+	case plan.Node_MARK:
+		return false, plan.Node_MARK
 	default:
 		panic(errors.New(errno.SyntaxErrororAccessRuleViolation, fmt.Sprintf("join typ '%v' not support now", n.JoinType)))
 	}
diff --git a/pkg/sql/compile/operator.go b/pkg/sql/compile/operator.go
index 1dd5b02e4..c699843e2 100644
--- a/pkg/sql/compile/operator.go
+++ b/pkg/sql/compile/operator.go
@@ -17,6 +17,7 @@ package compile
 import (
 	"context"
 	"fmt"
+
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/intersectall"
 
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/anti"
@@ -24,6 +25,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/hashbuild"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/intersect"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/loopanti"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/mark"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/minus"
 
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/loopjoin"
@@ -138,6 +140,20 @@ func dupInstruction(in vm.Instruction) vm.Instruction {
 			Result:     arg.Result,
 			Conditions: arg.Conditions,
 		}
+	case *mark.Argument:
+		{
+			rin.Arg = &mark.Argument{
+				Typs:         arg.Typs,
+				Cond:         arg.Cond,
+				Result:       arg.Result,
+				Conditions:   arg.Conditions,
+				OutputNull:   arg.OutputNull,
+				OutputMark:   arg.OutputMark,
+				MarkMeaning:  arg.MarkMeaning,
+				OutputAnyway: arg.OutputAnyway,
+				OnList:       arg.OnList,
+			}
+		}
 	case *offset.Argument:
 		rin.Arg = &offset.Argument{
 			Offset: arg.Offset,
@@ -427,6 +443,29 @@ func constructAnti(n *plan.Node, typs []types.Type, proc *process.Process) *anti
 	}
 }
 
+func constructMark(n *plan.Node, typs []types.Type, proc *process.Process, onList []*plan.Expr) *mark.Argument {
+	result := make([]int32, len(n.ProjectList))
+	for i, expr := range n.ProjectList {
+		rel, pos := constructJoinResult(expr)
+		if rel != 0 {
+			panic(errors.New(errno.SyntaxErrororAccessRuleViolation, fmt.Sprintf("mark result '%s' not support now", expr)))
+		}
+		result[i] = pos
+	}
+	cond, conds := extraJoinConditions(n.OnList)
+	return &mark.Argument{
+		Typs:         typs,
+		Result:       result,
+		Cond:         cond,
+		Conditions:   constructJoinConditions(conds),
+		OutputMark:   false,
+		OutputNull:   false,
+		MarkMeaning:  false,
+		OutputAnyway: false,
+		OnList:       onList,
+	}
+}
+
 func constructOrder(n *plan.Node, proc *process.Process) *order.Argument {
 	fs := make([]order.Field, len(n.OrderBy))
 	for i, e := range n.OrderBy {
@@ -661,6 +700,13 @@ func constructHashBuild(in vm.Instruction) *hashbuild.Argument {
 			Typs:        arg.Typs,
 			Conditions:  arg.Conditions[1],
 		}
+	case vm.Mark:
+		arg := in.Arg.(*mark.Argument)
+		return &hashbuild.Argument{
+			NeedHashMap: true,
+			Typs:        arg.Typs,
+			Conditions:  arg.Conditions[1],
+		}
 	case vm.Join:
 		arg := in.Arg.(*join.Argument)
 		return &hashbuild.Argument{
@@ -725,6 +771,7 @@ func constructHashBuild(in vm.Instruction) *hashbuild.Argument {
 			NeedHashMap: false,
 			Typs:        arg.Typs,
 		}
+
 	default:
 		panic(errors.New(errno.SyntaxErrororAccessRuleViolation, fmt.Sprintf("unsupport join type '%v'", in.Op)))
 	}
diff --git a/pkg/vm/overload.go b/pkg/vm/overload.go
index b588794fa..a4f15edb3 100644
--- a/pkg/vm/overload.go
+++ b/pkg/vm/overload.go
@@ -23,6 +23,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/hashbuild"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/intersect"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/loopanti"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/mark"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/minus"
 
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/loopsingle"
@@ -78,6 +79,7 @@ var stringFunc = [...]func(any, *bytes.Buffer){
 	Connector:  connector.String,
 	Projection: projection.String,
 	Anti:       anti.String,
+	Mark:       mark.String,
 
 	LoopJoin:   loopjoin.String,
 	LoopLeft:   loopleft.String,
@@ -121,6 +123,7 @@ var prepareFunc = [...]func(*process.Process, any) error{
 	Connector:  connector.Prepare,
 	Projection: projection.Prepare,
 	Anti:       anti.Prepare,
+	Mark:       mark.Prepare,
 
 	LoopJoin:   loopjoin.Prepare,
 	LoopLeft:   loopleft.Prepare,
@@ -164,7 +167,7 @@ var execFunc = [...]func(int, *process.Process, any) (bool, error){
 	Connector:  connector.Call,
 	Projection: projection.Call,
 	Anti:       anti.Call,
-
+	Mark:       mark.Call,
 	LoopJoin:   loopjoin.Call,
 	LoopLeft:   loopleft.Call,
 	LoopSingle: loopsingle.Call,
diff --git a/pkg/vm/types.go b/pkg/vm/types.go
index 209842b9f..051201819 100644
--- a/pkg/vm/types.go
+++ b/pkg/vm/types.go
@@ -32,6 +32,7 @@ const (
 	Projection
 	Anti
 	Single
+	Mark
 
 	LoopJoin
 	LoopLeft


Test patch:
diff --git a/pkg/sql/colexec/mark/join_test.go b/pkg/sql/colexec/mark/join_test.go
new file mode 100644
index 000000000..3b08c5e22
--- /dev/null
+++ b/pkg/sql/colexec/mark/join_test.go
@@ -0,0 +1,280 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mark
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/nulls"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/hashbuild"
+	"github.com/matrixorigin/matrixone/pkg/sql/plan/function"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/vm/mheap"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	Rows          = 10     // default rows
+	BenchmarkRows = 100000 // default rows for benchmark
+)
+
+// add unit tests for cases
+type markTestCase struct {
+	arg    *Argument
+	flgs   []bool // flgs[i] == true: nullable
+	types  []types.Type
+	proc   *process.Process
+	cancel context.CancelFunc
+	barg   *hashbuild.Argument
+}
+
+var (
+	tcs []markTestCase
+)
+
+func init() {
+	tcs = []markTestCase{
+		newTestCase(testutil.NewMheap(), []bool{false}, []types.Type{{Oid: types.T_int8}}, []int32{0},
+			[][]*plan.Expr{
+				{
+					newExpr(0, types.Type{Oid: types.T_int8}),
+				},
+				{
+					newExpr(0, types.Type{Oid: types.T_int8}),
+				},
+			}),
+		newTestCase(testutil.NewMheap(), []bool{true}, []types.Type{{Oid: types.T_int8}}, []int32{0},
+			[][]*plan.Expr{
+				{
+					newExpr(0, types.Type{Oid: types.T_int8}),
+				},
+				{
+					newExpr(0, types.Type{Oid: types.T_int8}),
+				},
+			}),
+	}
+}
+
+func TestString(t *testing.T) {
+	buf := new(bytes.Buffer)
+	for _, tc := range tcs {
+		String(tc.arg, buf)
+	}
+}
+
+func TestMark(t *testing.T) {
+	for _, tc := range tcs {
+		bat := hashBuild(t, tc)
+		err := Prepare(tc.proc, tc.arg)
+		require.NoError(t, err)
+		batWithNull := newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		batWithNull.Vecs[0].Nsp.Np.Add(0)
+		tc.proc.Reg.MergeReceivers[0].Ch <- batWithNull
+		tc.proc.Reg.MergeReceivers[0].Ch <- &batch.Batch{}
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- nil
+		tc.proc.Reg.MergeReceivers[1].Ch <- bat
+		for {
+			if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
+				break
+			}
+			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
+		}
+		require.Equal(t, int64(0), mheap.Size(tc.proc.Mp))
+	}
+	for _, tc := range tcs {
+		err := Prepare(tc.proc, tc.arg)
+		require.NoError(t, err)
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- &batch.Batch{}
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+		tc.proc.Reg.MergeReceivers[0].Ch <- nil
+		tc.proc.Reg.MergeReceivers[1].Ch <- nil
+		for {
+			if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
+				break
+			}
+			tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
+		}
+		require.Equal(t, int64(0), mheap.Size(tc.proc.Mp))
+	}
+}
+
+func TestHandleResultType(t *testing.T) {
+	ctr := new(container)
+	ctr.joinFlags = make([]bool, 3)
+	ctr.Nsp = nulls.NewWithSize(3)
+	ctr.handleResultType(0, condTrue)
+	ctr.handleResultType(1, condFalse)
+	ctr.handleResultType(2, condUnkown)
+}
+
+func BenchmarkMark(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		tcs = []markTestCase{
+			newTestCase(testutil.NewMheap(), []bool{false}, []types.Type{{Oid: types.T_int8}}, []int32{0},
+				[][]*plan.Expr{
+					{
+						newExpr(0, types.Type{Oid: types.T_int8}),
+					},
+					{
+						newExpr(0, types.Type{Oid: types.T_int8}),
+					},
+				}),
+			newTestCase(testutil.NewMheap(), []bool{true}, []types.Type{{Oid: types.T_int8}}, []int32{0},
+				[][]*plan.Expr{
+					{
+						newExpr(0, types.Type{Oid: types.T_int8}),
+					},
+					{
+						newExpr(0, types.Type{Oid: types.T_int8}),
+					},
+				}),
+		}
+		t := new(testing.T)
+		for _, tc := range tcs {
+			bat := hashBuild(t, tc)
+			err := Prepare(tc.proc, tc.arg)
+			require.NoError(t, err)
+			tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+			tc.proc.Reg.MergeReceivers[0].Ch <- &batch.Batch{}
+			tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+			tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+			tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+			tc.proc.Reg.MergeReceivers[0].Ch <- nil
+			tc.proc.Reg.MergeReceivers[1].Ch <- bat
+			for {
+				if ok, err := Call(0, tc.proc, tc.arg); ok || err != nil {
+					break
+				}
+				tc.proc.Reg.InputBatch.Clean(tc.proc.Mp)
+			}
+		}
+	}
+}
+
+func newExpr(pos int32, typ types.Type) *plan.Expr {
+	return &plan.Expr{
+		Typ: &plan.Type{
+			Size:  typ.Size,
+			Scale: typ.Scale,
+			Width: typ.Width,
+			Id:    int32(typ.Oid),
+		},
+		Expr: &plan.Expr_Col{
+			Col: &plan.ColRef{
+				ColPos: pos,
+			},
+		},
+	}
+}
+
+func newTestCase(m *mheap.Mheap, flgs []bool, ts []types.Type, rp []int32, cs [][]*plan.Expr) markTestCase {
+	proc := process.New(m)
+	proc.Reg.MergeReceivers = make([]*process.WaitRegister, 2)
+	ctx, cancel := context.WithCancel(context.Background())
+	proc.Reg.MergeReceivers[0] = &process.WaitRegister{
+		Ctx: ctx,
+		Ch:  make(chan *batch.Batch, 10),
+	}
+	proc.Reg.MergeReceivers[1] = &process.WaitRegister{
+		Ctx: ctx,
+		Ch:  make(chan *batch.Batch, 3),
+	}
+	fid := function.EncodeOverloadID(function.EQUAL, 4)
+	args := make([]*plan.Expr, 0, 2)
+	args = append(args, &plan.Expr{
+		Typ: &plan.Type{
+			Size: ts[0].Size,
+			Id:   int32(ts[0].Oid),
+		},
+		Expr: &plan.Expr_Col{
+			Col: &plan.ColRef{
+				RelPos: 0,
+				ColPos: 0,
+			},
+		},
+	})
+	args = append(args, &plan.Expr{
+		Typ: &plan.Type{
+			Size: ts[0].Size,
+			Id:   int32(ts[0].Oid),
+		},
+		Expr: &plan.Expr_Col{
+			Col: &plan.ColRef{
+				RelPos: 1,
+				ColPos: 0,
+			},
+		},
+	})
+	cond := &plan.Expr{
+		Typ: &plan.Type{
+			Size: 1,
+			Id:   int32(types.T_bool),
+		},
+		Expr: &plan.Expr_F{
+			F: &plan.Function{
+				Args: args,
+				Func: &plan.ObjectRef{Obj: fid},
+			},
+		},
+	}
+
+	c := markTestCase{
+		types:  ts,
+		flgs:   flgs,
+		proc:   proc,
+		cancel: cancel,
+		arg: &Argument{
+			Typs:       ts,
+			Result:     rp,
+			Conditions: cs,
+			Cond:       cond,
+			OnList:     []*plan.Expr{cond},
+		},
+		barg: &hashbuild.Argument{
+			Typs:        ts,
+			NeedHashMap: true,
+			Conditions:  cs[1],
+		},
+	}
+	return c
+}
+
+func hashBuild(t *testing.T, tc markTestCase) *batch.Batch {
+	err := hashbuild.Prepare(tc.proc, tc.barg)
+	require.NoError(t, err)
+	tc.proc.Reg.MergeReceivers[0].Ch <- newBatch(t, tc.flgs, tc.types, tc.proc, Rows)
+	tc.proc.Reg.MergeReceivers[0].Ch <- nil
+	ok, err := hashbuild.Call(0, tc.proc, tc.barg)
+	require.NoError(t, err)
+	require.Equal(t, true, ok)
+	return tc.proc.Reg.InputBatch
+}
+
+// create a new block based on the type information, flgs[i] == ture: has null
+func newBatch(t *testing.T, flgs []bool, ts []types.Type, proc *process.Process, rows int64) *batch.Batch {
+	return testutil.NewBatch(ts, true, int(rows), proc.Mp)
+}
diff --git a/test/cases/join/markjoin.test b/test/cases/join/markjoin.test
new file mode 100644
index 000000000..e6abe6554
--- /dev/null
+++ b/test/cases/join/markjoin.test
@@ -0,0 +1,16 @@
+-- @bvt:issue#4153
+drop table if exists t1;
+drop table if exists t2;
+Create table t1(sid int, sname varchar(20), mid int);
+Create table t2(mid int,mname varchar(20));
+insert into t1 values(1,'Seven',1),(2,'Ultraman',2),(3,'Gaia',NULL);
+insert into t2 values(1,'James'),(3,'Tiga');
+select t1.sname from t1 where not exists (select t2.mname from t2 where t2.mid = t1.mid);
+drop table if exists t1;
+drop table if exists t2;
+Create table t1(a int,b int,c int);
+Create table t2(d int,e int,f int);
+insert into t1 values(1,2,4),(3,4,7),(2,2,5),(5,2,3),(3,NULL,2),(NULL,2,1);
+insert into t2 values(2,NULL,1),(3,3,4),(5,5,1),(NULL,NULL,7),(NULL,4,5);
+select t1.c from t1 where a not in (select d from t2 where t1.b = e);
+-- @bvt:issue
diff --git a/test/result/join/markjoin.result b/test/result/join/markjoin.result
new file mode 100644
index 000000000..83b30fa89
--- /dev/null
+++ b/test/result/join/markjoin.result
@@ -0,0 +1,23 @@
+drop table if exists t1;
+drop table if exists t2;
+Create table t1(sid int, sname varchar(20), mid int);
+Create table t2(mid int,mname varchar(20));
+insert into t1 values(1,'Seven',1),(2,'Ultraman',2),(3,'Gaia',NULL);
+insert into t2 values(1,'James'),(3,'Tiga');
+select t1.sname from t1 where not exists (select t2.mname from t2 where t2.mid = t1.mid);
+sname
+Ultraman
+Gaia
+drop table if exists t1;
+drop table if exists t2;
+Create table t1(a int,b int,c int);
+Create table t2(d int,e int,f int);
+insert into t1 values(1,2,4),(3,4,7),(2,2,5),(5,2,3),(3,NULL,2),(NULL,2,1);
+insert into t2 values(2,NULL,1),(3,3,4),(5,5,1),(NULL,NULL,7),(NULL,4,5);
+select t1.c from t1 where a not in (select d from t2 where t1.b = e);
+c
+4
+5
+3
+2
+1