fix: aggregate unsummable metrics with max instead of avg

(cherry picked from commit d88c48f75eb18c4669b68415c08da7d916ce5c24)

Author: zhuofeng

server/ingester/pkg/ckwriter/ckwriter.go

@@ -247,6 +247,18 @@ func initTable(conn clickhouse.Conn, timeZone string, t *ckdb.Table, orgID uint1
 			if err := ExecSQL(conn, t.MakeAggrTableRenameSQL1H(orgID)); err != nil {
 				log.Warningf("drop 1h agg table failed: %s", err)
 			}
+		} else if t.IsAggrMaxColumnWrong(aggrTableCreateSQL) {
+			// the aggregation of unsummable(_max) columns changed from avg to max;
+			// alter the columns in place to keep the 1h aggregated history. the mv is
+			// dropped first so it stops writing avgState and is recreated with maxState below.
+			if err := ExecSQL(conn, t.MakeAggrMvTableDropSQL1H(orgID)); err != nil {
+				log.Warningf("drop 1h mv table failed: %s", err)
+			}
+			for _, sql := range t.MakeAggrMaxColumnAlterSQL1H(orgID) {
+				if err := ExecSQL(conn, sql); err != nil {
+					log.Warningf("alter 1h agg max column failed: %s", err)
+				}
+			}
 		}
 
 		if err := ExecSQL(conn, t.MakeAggrTableCreateSQL1H(orgID)); err != nil {
@@ -270,6 +282,18 @@ func initTable(conn clickhouse.Conn, timeZone string, t *ckdb.Table, orgID uint1
 			if err := ExecSQL(conn, t.MakeAggrTableRenameSQL1D(orgID)); err != nil {
 				log.Warningf("drop 1d agg table failed: %s", err)
 			}
+		} else if t.IsAggrMaxColumnWrong(aggrTableCreateSQL) {
+			// the aggregation of unsummable(_max) columns changed from avg to max;
+			// alter the columns in place to keep the 1d aggregated history. the mv is
+			// dropped first so it stops writing avgState and is recreated with maxState below.
+			if err := ExecSQL(conn, t.MakeAggrMvTableDropSQL1D(orgID)); err != nil {
+				log.Warningf("drop 1d mv table failed: %s", err)
+			}
+			for _, sql := range t.MakeAggrMaxColumnAlterSQL1D(orgID) {
+				if err := ExecSQL(conn, sql); err != nil {
+					log.Warningf("alter 1d agg max column failed: %s", err)
+				}
+			}
 		}
 		if err := ExecSQL(conn, t.MakeAggrTableCreateSQL1D(orgID)); err != nil {
 			log.Warningf("create 1d agg table failed: %s", err)

server/libs/ckdb/table.go

@@ -361,7 +361,7 @@ func isUnsummable(column *Column) bool {
 
 func getAggr(column *Column) string {
 	if isUnsummable(column) {
-		return "avg"
+		return "max"
 	}
 	return "sum"
 }
@@ -389,6 +389,51 @@ func (t *Table) IsAggrTableWrong(createTableSQL string) bool {
 	return len(orderKeys) != t.OrderKeysCount()
 }
 
+// IsAggrMaxColumnWrong reports whether an existing aggregate table still
+// aggregates unsummable(`_max`) columns with `avg` instead of `max`. Such
+// tables only need their aggregate columns altered in place (see
+// MakeAggrMaxColumnAlterSQLs), not a full rebuild, so historical 1h/1d data
+// is preserved.
+func (t *Table) IsAggrMaxColumnWrong(createTableSQL string) bool {
+	return strings.Contains(createTableSQL, "_max__agg` AggregateFunction(avg")
+}
+
+// MakeAggrMaxColumnAlterSQLs returns the ALTER statements that migrate the
+// unsummable aggregate columns of an existing aggregate table from the old
+// `avg` aggregation to `max`. Each column is dropped and re-added as a
+// `max` AggregateFunction column; only the data of these specific columns is
+// lost, all other columns and rows are kept.
+func (t *Table) MakeAggrMaxColumnAlterSQLs(orgID uint16, aggrInterval AggregationInterval) []string {
+	tableAgg := t.AggrTable(orgID, aggrInterval)
+	sqls := []string{}
+	for _, c := range t.Columns {
+		// ignore fields starting with '_', such as _tid, _id
+		if strings.HasPrefix(c.Name, "_") || c.IgnoredInAggrTable || c.GroupBy {
+			continue
+		}
+		// only columns that fall into the default branch of
+		// MakeAggrTableCreateSQL are created via getAggr(); those are the
+		// ones whose aggregation switched from avg to max.
+		if c.Aggr != AggrNone || !isUnsummable(c) {
+			continue
+		}
+		aggrColumn := fmt.Sprintf("%s__agg", c.Name)
+		sqls = append(sqls,
+			fmt.Sprintf("ALTER TABLE %s DROP COLUMN IF EXISTS `%s`", tableAgg, aggrColumn),
+			fmt.Sprintf("ALTER TABLE %s ADD COLUMN IF NOT EXISTS `%s` AggregateFunction(max, %s)", tableAgg, aggrColumn, c.Type.String()),
+		)
+	}
+	return sqls
+}
+
+func (t *Table) MakeAggrMaxColumnAlterSQL1H(orgID uint16) []string {
+	return t.MakeAggrMaxColumnAlterSQLs(orgID, AggregationHour)
+}
+
+func (t *Table) MakeAggrMaxColumnAlterSQL1D(orgID uint16) []string {
+	return t.MakeAggrMaxColumnAlterSQLs(orgID, AggregationDay)
+}
+
 func (t *Table) IsLocalTableWrong(createTableSql string) bool {
 	return strings.Contains(createTableSql, "mem-inuse")
 }
@@ -397,6 +442,10 @@ func (t *Table) AggrTable(orgID uint16, aggrInterval AggregationInterval) string
 	return fmt.Sprintf("%s.`%s.%s_agg`", t.OrgDatabase(orgID), t.tableAggrPrefix(), aggrInterval.String())
 }
 
+func (t *Table) MvTable(orgID uint16, aggrInterval AggregationInterval) string {
+	return fmt.Sprintf("%s.`%s.%s_mv`", t.OrgDatabase(orgID), t.tableAggrPrefix(), aggrInterval.String())
+}
+
 func (t *Table) AggrTable1S(orgID uint16) string {
 	return t.AggrTable(orgID, AggregationSecond)
 }
@@ -434,6 +483,14 @@ func (t *Table) AggrTable1H(orgID uint16) string {
 	return t.AggrTable(orgID, AggregationHour)
 }
 
+func (t *Table) MvTable1H(orgID uint16) string {
+	return t.MvTable(orgID, AggregationHour)
+}
+
+func (t *Table) MakeAggrMvTableDropSQL1H(orgID uint16) string {
+	return fmt.Sprintf("DROP TABLE IF EXISTS %s", t.MvTable1H(orgID))
+}
+
 func (t *Table) MakeAggrTableRenameSQL1H(orgID uint16) string {
 	table := t.AggrTable1H(orgID)
 	return fmt.Sprintf("RENAME TABLE %s to %s_%d`", table, table[:len(table)-1], time.Now().Unix())
@@ -459,6 +516,14 @@ func (t *Table) AggrTable1D(orgID uint16) string {
 	return t.AggrTable(orgID, AggregationDay)
 }
 
+func (t *Table) MvTable1D(orgID uint16) string {
+	return t.MvTable(orgID, AggregationDay)
+}
+
+func (t *Table) MakeAggrMvTableDropSQL1D(orgID uint16) string {
+	return fmt.Sprintf("DROP TABLE IF EXISTS %s", t.MvTable1D(orgID))
+}
+
 func (t *Table) MakeAggrTableRenameSQL1D(orgID uint16) string {
 	table := t.AggrTable1D(orgID)
 	return fmt.Sprintf("RENAME TABLE %s to %s_%d`", table, table[:len(table)-1], time.Now().Unix())