Classify CH ErrTooManyParts (#3915)

Author: ilidemi

flow/alerting/classifier.go

@@ -50,6 +50,7 @@ var (
 	)
 	// ID(a14c2a1c-edcd-5fcb-73be-bd04e09fccb7) not found in user directories
 	ClickHouseNotFoundInUserDirsRe    = regexp.MustCompile("ID\\([a-z0-9-]+\\) not found in `?user directories`?")
+	ClickHouseTooManyPartsTableRe     = regexp.MustCompile(`in table '(.+)'\.`)
 	PostgresPublicationDoesNotExistRe = regexp.MustCompile(`publication ".*?" does not exist`)
 	PostgresSnapshotDoesNotExistRe    = regexp.MustCompile(`snapshot ".*?" does not exist`)
 	PostgresWalSegmentRemovedRe       = regexp.MustCompile(`requested WAL segment \w+ has already been removed`)
@@ -214,6 +215,9 @@ var (
 	ErrorNotifyClickHouseSupportIsDisabledError = ErrorClass{
 		Class: "NOTIFY_CLICKHOUSE_SUPPORT_IS_DISABLED_ERROR", action: NotifyUser,
 	}
+	ErrorNotifyTooManyPartsError = ErrorClass{
+		Class: "NOTIFY_TOO_MANY_PARTS", action: NotifyUser,
+	}
 	// Catch-all for unclassified errors
 	ErrorOther = ErrorClass{
 		// These are unclassified and should not be exposed
@@ -884,6 +888,18 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 			}
 		case chproto.ErrSupportIsDisabled:
 			return ErrorNotifyClickHouseSupportIsDisabledError, chErrorInfo
+		case chproto.ErrTooManyParts:
+			var additionalAttributes map[AdditionalErrorAttributeKey]string
+			if matches := ClickHouseTooManyPartsTableRe.FindStringSubmatch(chException.Message); len(matches) > 1 {
+				additionalAttributes = map[AdditionalErrorAttributeKey]string{
+					ErrorAttributeKeyTable: matches[1],
+				}
+			}
+			return ErrorNotifyTooManyPartsError, ErrorInfo{
+				Source:               chErrorInfo.Source,
+				Code:                 chErrorInfo.Code,
+				AdditionalAttributes: additionalAttributes,
+			}
 		}
 		var normalizationErr *exceptions.NormalizationError
 		if isClickHouseMvError(chException) {
@@ -995,7 +1011,7 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 			Code:   "UNSUPPORTED_SCHEMA_CHANGE",
 			AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
 				ErrorAttributeKeyTable:  postgresReplicaIdentityIndexError.Table,
-				ErrorAttributeKeyColumn: "n\a",
+				ErrorAttributeKeyColumn: "n/a",
 			},
 		}
 	}

flow/alerting/classifier_test.go

@@ -728,6 +728,39 @@ func TestMySQLStreamingTLSHandshakeErrorShouldBeRecoverable(t *testing.T) {
 	}, errInfo, "Unexpected error info")
 }
 
+func TestClickHouseTooManyPartsWithTableName(t *testing.T) {
+	err := &clickhouse.Exception{
+		Code: int32(chproto.ErrTooManyParts),
+		//nolint:lll
+		Message: "Too many parts (3025 with average size of 65.51 MiB) in table 'ss_replica.posts_resync (db2b0f62-f577-4116-8b5d-e0f760a42bee)'. Merges are processing significantly slower than inserts",
+	}
+	errorClass, errInfo := GetErrorClass(t.Context(),
+		exceptions.NewQRepSyncError(fmt.Errorf("QRepSync Error: %w", err), "", ""))
+	assert.Equal(t, ErrorNotifyTooManyPartsError, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceClickHouse,
+		Code:   strconv.Itoa(int(chproto.ErrTooManyParts)),
+		AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+			ErrorAttributeKeyTable: "ss_replica.posts_resync (db2b0f62-f577-4116-8b5d-e0f760a42bee)",
+		},
+	}, errInfo)
+}
+
+func TestClickHouseTooManyPartsWithoutTableName(t *testing.T) {
+	err := &clickhouse.Exception{
+		Code: int32(chproto.ErrTooManyParts),
+		//nolint:lll
+		Message: "Too many partitions for single INSERT block (more than 9999). The limit is controlled by 'max_partitions_per_insert_block' setting.",
+	}
+	errorClass, errInfo := GetErrorClass(t.Context(),
+		exceptions.NewQRepSyncError(fmt.Errorf("QRepSync Error: %w", err), "", ""))
+	assert.Equal(t, ErrorNotifyTooManyPartsError, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceClickHouse,
+		Code:   strconv.Itoa(int(chproto.ErrTooManyParts)),
+	}, errInfo)
+}
+
 func TestMySQLUnsupportedDDLShouldNotifyUser(t *testing.T) {
 	err := exceptions.NewMySQLUnsupportedDDLError("test_db.test_table")
 	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("mysql error: %w", err))