to 3.0: Try snapshot read if the subscription does not exist. (#22412)

Try snapshot read if the subscription does not exist.

Approved by: @XuPeng-SH

Author: gouhongshen

pkg/vm/engine/disttae/logtail_consumer.go

@@ -388,18 +388,24 @@ func (c *PushClient) toSubscribeTable(
 	dbName string,
 ) (ps *logtailreplay.PartitionState, err error) {
 
-	if faultInjected, _ := objectio.LogCNSubscribeTableFailInjected(
+	var (
+		skip     bool
+		state    SubscribeState
+		injected bool
+	)
+
+	if injected, _ = objectio.LogCNSubscribeTableFailInjected(
 		dbName, tableName,
-	); faultInjected {
-		return nil, moerr.NewInternalErrorNoCtx("injected subscribe table err")
+	); injected {
+		return nil,
+			moerr.NewInternalErrorNoCtx("injected subscribe table err")
 	}
 
-	var skip bool
 	if skip, ps = c.skipSubIfSubscribed(ctx, tableID, dbID); skip {
 		return ps, nil
 	}
 
-	state, err := c.toSubIfUnsubscribed(ctx, dbID, tableID)
+	state, err = c.toSubIfUnsubscribed(ctx, dbID, tableID)
 	if err != nil {
 		return nil, err
 	}
@@ -423,12 +429,7 @@ func (c *PushClient) toSubscribeTable(
 			}
 		case SubRspTableNotExist:
 			c.subscribed.clearTable(dbID, tableID)
-			return nil, moerr.NewInternalErrorf(
-				ctx,
-				"%s to subcribe tbl[%d-%s] failed since table is not exist",
-				logTag,
-				tableID,
-				tableName)
+			return nil, moerr.NewNoSuchTable(ctx, fmt.Sprintf("%s(%d)", dbName, dbID), fmt.Sprintf("%s(%d)", tableName, tableID))
 		case Unsubscribing:
 			//need to wait for unsubscribe succeed for making the subscribe and unsubscribe execute in order,
 			// otherwise the partition state will leak log tails.

pkg/vm/engine/disttae/txn_table.go

@@ -2025,7 +2025,12 @@ func (tbl *txnTable) getPartitionState(
 		}
 	}()
 
-	createdInTxn, err := tbl.isCreatedInTxn(ctx)
+	var (
+		eng          = tbl.eng.(*Engine)
+		createdInTxn bool
+	)
+
+	createdInTxn, err = tbl.isCreatedInTxn(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -2041,7 +2046,6 @@ func (tbl *txnTable) getPartitionState(
 	}
 
 	// Subscribe a latest partition state
-	eng := tbl.eng.(*Engine)
 	if ps, err = eng.PushClient().toSubscribeTable(
 		ctx,
 		tbl.tableId,
@@ -2050,14 +2054,21 @@ func (tbl *txnTable) getPartitionState(
 		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()),
+			"Txn-Table-ToSubscribeTable-Failed",
+			zap.String("db-name", tbl.db.databaseName),
+			zap.Uint64("db-id", tbl.db.databaseId),
+			zap.String("tbl-name", tbl.tableName),
+			zap.Uint64("tbl-id", tbl.tableId),
+			zap.String("txn-info", tbl.db.op.Txn().DebugString()),
+			zap.Bool("is-snapshot-op", tbl.db.op.IsSnapOp()),
 			zap.Error(err),
 		)
-		return nil, err
+
+		// if the table not exists, try snapshot read
+		if !moerr.IsMoErrCode(err, moerr.ErrNoSuchTable) {
+			return nil, err
+		}
+
 	} else if ps != nil && ps.CanServe(types.TimestampToTS(tbl.db.op.SnapshotTS())) {
 		return
 	}