to 3.0: report the error when the subscription table fails. (#22406)

Report an error when the subscription table fails.

Approved by: @XuPeng-SH, @heni02, @aunjgr, @fengttt

Author: gouhongshen

pkg/fulltext/fixedbytepool.go

@@ -123,7 +123,7 @@ func (part *Partition) Id() uint64 {
 
 // memory allocation with mpool.MPool
 func (part *Partition) alloc(capacity uint64) (err error) {
-	part.data, err = part.proc.Mp().Alloc(int(capacity), false)
+	part.data, err = part.proc.Mp().Alloc(int(capacity), true)
 	if err != nil {
 		return err
 	}

pkg/objectio/injects.go

@@ -62,7 +62,8 @@ const (
 	FJ_CDCAddExecErr             = "fj/cdc/addexecerr"
 	FJ_CDCAddExecConsumeTruncate = "fj/cdc/addexecconsumetruncate"
 
-	FJ_CNFlushSmallObjs = "fj/cn/flush_small_objs"
+	FJ_CNFlushSmallObjs     = "fj/cn/flush_small_objs"
+	FJ_CNSubscribeTableFail = "fj/cn/subscribe_table_fail"
 )
 
 const (
@@ -186,6 +187,16 @@ func LogCNFlushSmallObjsInjected(args ...string) (bool, int) {
 	return ok, level
 }
 
+func LogCNSubscribeTableFailInjected(args ...string) (bool, int) {
+	iarg, sarg, injected := fault.TriggerFault(FJ_CNSubscribeTableFail)
+	if !injected {
+		return false, 0
+	}
+
+	ok, level := checkLoggingArgs(int(iarg), sarg, args...)
+	return ok, level
+}
+
 func InjectLogPartitionState(
 	databaseName string,
 	tableName string,

pkg/sql/colexec/anti/join.go

@@ -155,13 +155,13 @@ func (ctr *container) emptyProbe(ap *AntiJoin, inbat *batch.Batch, proc *process
 		if n > hashmap.UnitLimit {
 			n = hashmap.UnitLimit
 		}
-		for k := 0; k < n; k++ {
-			for j, pos := range ap.Result {
-				if err := ctr.rbat.Vecs[j].UnionOne(inbat.Vecs[pos], int64(i+k), proc.Mp()); err != nil {
-					return err
-				}
+
+		for j, pos := range ap.Result {
+			if err := ctr.rbat.Vecs[j].UnionBatch(inbat.Vecs[pos], int64(i), n, nil, proc.Mp()); err != nil {
+				return err
 			}
 		}
+
 		ctr.rbat.AddRowCount(n)
 	}
 

pkg/vm/engine/disttae/logtail_consumer.go

@@ -388,6 +388,12 @@ func (c *PushClient) toSubscribeTable(
 	dbName string,
 ) (ps *logtailreplay.PartitionState, err error) {
 
+	if faultInjected, _ := objectio.LogCNSubscribeTableFailInjected(
+		dbName, tableName,
+	); faultInjected {
+		return nil, moerr.NewInternalErrorNoCtx("injected subscribe table err")
+	}
+
 	var skip bool
 	if skip, ps = c.skipSubIfSubscribed(ctx, tableID, dbID); skip {
 		return ps, nil

pkg/vm/engine/disttae/txn_table.go

@@ -2042,13 +2042,23 @@ func (tbl *txnTable) getPartitionState(
 
 	// Subscribe a latest partition state
 	eng := tbl.eng.(*Engine)
-	ps, err = eng.PushClient().toSubscribeTable(
+	if ps, err = eng.PushClient().toSubscribeTable(
 		ctx,
 		tbl.tableId,
 		tbl.tableName,
 		tbl.db.databaseId,
-		tbl.db.databaseName)
-	if ps != nil && ps.CanServe(types.TimestampToTS(tbl.db.op.SnapshotTS())) {
+		tbl.db.databaseName,
+	); err != nil {
+		logutil.Error(
+			"Txn-Table-Subscribe-Failed",
+			zap.String("table-name", tbl.tableName),
+			zap.Uint64("table-id", tbl.tableId),
+			zap.String("txn", tbl.db.op.Txn().DebugString()),
+			zap.Bool("is-snapshot", tbl.db.op.IsSnapOp()),
+			zap.Error(err),
+		)
+		return nil, err
+	} else if ps != nil && ps.CanServe(types.TimestampToTS(tbl.db.op.SnapshotTS())) {
 		return
 	}
 

test/distributed/cases/snapshot/clone/table_clone.result

@@ -7,6 +7,17 @@ insert into t1(a,b) values(20,20),(21,21);
 insert into t1(b) values(20000),(20001);
 insert into t1(a,b) values(40000, 40000);
 create table t2 clone t1;
+select enable_fault_injection();
+enable_fault_injection()
+true
+select add_fault_point('fj/cn/subscribe_table_fail',':::','echo',40,'db1.t2');
+add_fault_point(fj/cn/subscribe_table_fail, :::, echo, 40, db1.t2)
+true
+select * from t2 order by a asc;
+internal error: injected subscribe table err
+select disable_fault_injection();
+disable_fault_injection()
+true
 select * from t2 order by a asc;
 a    b
 1    1

test/distributed/cases/snapshot/clone/table_clone.sql

@@ -9,6 +9,13 @@ insert into t1(b) values(20000),(20001);
 insert into t1(a,b) values(40000, 40000);
 
 create table t2 clone t1;
+
+select enable_fault_injection();
+-- @ignore:0
+select add_fault_point('fj/cn/subscribe_table_fail',':::','echo',40,'db1.t2');
+select * from t2 order by a asc;
+select disable_fault_injection();
+
 select * from t2 order by a asc;
 
 insert into t2(b) values(3),(4);