statistics: do not ignore DDL event handing errors (pingcap#58980) (p…

…ingcap#59367)

close pingcap#58545

pkg/ddl/notifier/subscribe.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	goerr "errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -160,6 +161,12 @@ func (n *DDLNotifier) start() {
 			return
 		case <-ticker.C:
 			if err := n.processEvents(ctx); err != nil {
+				intest.Assert(
+					errors.ErrorEqual(err, context.Canceled) ||
+						strings.Contains(err.Error(), "mock handleTaskOnce error") ||
+						strings.Contains(err.Error(), "session pool closed"),
+					fmt.Sprintf("error processing events: %v", err),
+				)
 				logutil.Logger(ctx).Error("Error processing events", zap.Error(err))
 			}
 		}

pkg/statistics/handle/ddl/BUILD.bazel

@@ -31,7 +31,7 @@ go_test(
     timeout = "short",
     srcs = ["ddl_test.go"],
     flaky = True,
-    shard_count = 21,
+    shard_count = 22,
     deps = [
         ":ddl",
         "//pkg/ddl/notifier",

pkg/statistics/handle/ddl/ddl.go

@@ -16,16 +16,17 @@ package ddl
 
 import (
 	"context"
+	"fmt"
+	"strings"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/ddl/notifier"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/statistics/handle/lockstats"
-	statslogutil "github.com/pingcap/tidb/pkg/statistics/handle/logutil"
 	"github.com/pingcap/tidb/pkg/statistics/handle/storage"
 	"github.com/pingcap/tidb/pkg/statistics/handle/types"
 	"github.com/pingcap/tidb/pkg/statistics/handle/util"
-	"go.uber.org/zap"
+	"github.com/pingcap/tidb/pkg/util/intest"
 )
 
 type ddlHandlerImpl struct {
@@ -50,16 +51,17 @@ func NewDDLHandler(
 
 // HandleDDLEvent begins to process a ddl task.
 func (h *ddlHandlerImpl) HandleDDLEvent(ctx context.Context, sctx sessionctx.Context, s *notifier.SchemaChangeEvent) error {
-	// Ideally, we shouldn't allow any errors to be ignored, but for now, some queries can fail.
-	// Temporarily ignore the error and we need to check all queries to ensure they are correct.
-	if err := h.sub.handle(ctx, sctx, s); err != nil {
-		statslogutil.StatsLogger().Warn(
-			"failed to handle DDL event",
-			zap.String("event", s.String()),
-			zap.Error(err),
+	err := h.sub.handle(ctx, sctx, s)
+	if err != nil {
+		intest.Assert(
+			errors.ErrorEqual(err, context.Canceled) ||
+				strings.Contains(err.Error(), "mock handleTaskOnce error") ||
+				strings.Contains(err.Error(), "session pool closed"),
+			fmt.Sprintf("handle ddl event failed, err: %v", err),
 		)
 	}
-	return nil
+
+	return err
 }
 
 // DDLEventCh returns ddl events channel in handle.

pkg/statistics/handle/ddl/ddl_test.go

@@ -1390,8 +1390,34 @@ func TestDumpStatsDeltaBeforeHandleDDLEvent(t *testing.T) {
 	tk.MustExec("insert into t values (1), (2), (3)")
 	h := dom.StatsHandle()
 	require.NoError(t, h.DumpStatsDeltaToKV(true))
+	// Also manually insert a histogram record.
+	is := dom.InfoSchema()
+	tbl, err := is.TableByName(context.Background(), pmodel.NewCIStr("test"), pmodel.NewCIStr("t"))
+	require.NoError(t, err)
+	_, err = tk.Exec("insert into mysql.stats_histograms (table_id, is_index, hist_id, distinct_count, version) values (?, 0, ?, 0, ?)", tbl.Meta().ID, 1, 1)
+	require.NoError(t, err)
 	// Find the DDL event.
 	event := findEvent(h.DDLEventCh(), model.ActionCreateTable)
-	err := statstestutil.HandleDDLEventWithTxn(h, event)
+	err = statstestutil.HandleDDLEventWithTxn(h, event)
+	require.NoError(t, err)
+}
+
+func TestDumpStatsDeltaBeforeHandleAddColumnEvent(t *testing.T) {
+	store, do := testkit.CreateMockStoreAndDomain(t)
+	testKit := testkit.NewTestKit(t, store)
+	testKit.MustExec("use test")
+	testKit.MustExec("create table t (c1 int, c2 int, index idx(c1, c2))")
+	// Insert some data.
+	testKit.MustExec("insert into t values (1, 2), (2, 3), (3, 4)")
+	testKit.MustExec("analyze table t")
+	// Add column.
+	testKit.MustExec("alter table t add column c10 int")
+	// Insert some data.
+	testKit.MustExec("insert into t values (4, 5, 6)")
+	// Analyze table to force create the histogram meta record.
+	testKit.MustExec("analyze table t")
+	// Find the add column event.
+	event := findEvent(do.StatsHandle().DDLEventCh(), model.ActionAddColumn)
+	err := statstestutil.HandleDDLEventWithTxn(do.StatsHandle(), event)
 	require.NoError(t, err)
 }

pkg/statistics/handle/storage/save.go

@@ -478,7 +478,7 @@ func InsertColStats2KV(
 			continue
 		}
 
-		// If this stats exists, we insert histogram meta first, the distinct_count will always be one.
+		// If this stats doest not exist, we insert histogram meta first, the distinct_count will always be one.
 		if _, err = util.ExecWithCtx(
 			ctx, sctx,
 			`insert into mysql.stats_histograms
@@ -521,15 +521,15 @@ func InsertTableStats2KV(
 	}
 	if _, err = util.ExecWithCtx(
 		ctx, sctx,
-		"insert into mysql.stats_meta (version, table_id) values(%?, %?)",
+		"insert ignore into mysql.stats_meta (version, table_id) values(%?, %?)",
 		startTS, physicalID,
 	); err != nil {
 		return 0, errors.Trace(err)
 	}
 	for _, col := range info.Columns {
 		if _, err = util.ExecWithCtx(
 			ctx, sctx,
-			`insert into mysql.stats_histograms
+			`insert ignore into mysql.stats_histograms
 				(table_id, is_index, hist_id, distinct_count, version)
 			values (%?, 0, %?, 0, %?)`,
 			physicalID, col.ID, startTS,
@@ -540,7 +540,7 @@ func InsertTableStats2KV(
 	for _, idx := range info.Indices {
 		if _, err = util.ExecWithCtx(
 			ctx, sctx,
-			`insert into mysql.stats_histograms
+			`insert ignore into mysql.stats_histograms
 				(table_id, is_index, hist_id, distinct_count, version)
 			values(%?, 1, %?, 0, %?)`,
 			physicalID, idx.ID, startTS,