fix: speed up business portrait statistics

2026-11-05 17:51:01 +08:00
parent aaee077253
commit f8e26c5baf
2 changed files with 200 additions and 50 deletions
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"strconv"

+	"golang.org/x/sync/errgroup"
 	"gorm.io/gorm"
 	"kra/internal/biz/business"
 )
@@ -161,15 +162,13 @@ func (r *businessPortraitRepo) Statistics(ctx context.Context, scope business.Po
 	if scope.Range.Hourly {
 		tables = append(tables, "tab_user", "tab_spend")
 	}
-	for _, table := range tables {
-		available, err := businessTableAvailable(db, table)
-		if err != nil {
-			return nil, err
-		}
-		if !available {
-			result.Available = false
-			return result, nil
-		}
+	available, err := businessTablesAvailable(db, tables...)
+	if err != nil {
+		return nil, err
+	}
+	if !available {
+		result.Available = false
+		return result, nil
 	}
 	mode := 0
 	if scope.IDCardDedup {
@@ -177,49 +176,77 @@ func (r *businessPortraitRepo) Statistics(ctx context.Context, scope business.Po
 	}
 	begin, end := scope.Range.Begin.Format("2006-01-02"), scope.Range.End.Format("2006-01-02")
 	previousBegin, previousEnd := scope.Range.PreviousBegin.Format("2006-01-02"), scope.Range.PreviousEnd.Format("2006-01-02")
-	periods := []struct {
-		begin, end string
-		counts     *business.PortraitCounts
-		pay        *string
-	}{{begin, end, &result.Current, &result.CurrentPay}}
-	if !scope.NoCompare {
-		periods = append(periods, struct {
-			begin, end string
-			counts     *business.PortraitCounts
-			pay        *string
-		}{previousBegin, previousEnd, &result.Previous, &result.PreviousPay})
+	// 合并本期和对比期的扫描；本月同期之间的日期空隙不能混入统计。
+	group, queryCtx := errgroup.WithContext(ctx)
+	group.SetLimit(2)
+	queryDB := db.WithContext(queryCtx)
+	summaryQuery := func(table string, dedup int) *gorm.DB {
+		if scope.NoCompare {
+			return portraitReportQuery(queryDB, scope, table, begin, end, dedup)
+		}
+		return portraitReportQuery(queryDB, scope, table, min(begin, previousBegin), max(end, previousEnd), dedup).
+			Where("(r.stat_date >= ? AND r.stat_date <= ?) OR (r.stat_date >= ? AND r.stat_date <= ?)", begin, end, previousBegin, previousEnd)
 	}
-	for _, period := range periods {
-		var rows []portraitCountPO
-		q := portraitReportQuery(db, scope, "tab_sw_report_day_key", period.begin, period.end, mode)
-		if err := q.Select("r.metric,COUNT(DISTINCT " + portraitIdentity(db, scope.IDCardDedup) + ") AS count").Group("r.metric").Scan(&rows).Error; err != nil {
-			return nil, business.ErrStorage
+	group.Go(func() error {
+		var rows []struct {
+			Metric            int
+			Current, Previous int64
 		}
-		*period.counts = toPortraitCounts(rows)
-		if err := portraitReportQuery(db, scope, "tab_sw_report_day", period.begin, period.end, 0).Select("COALESCE(SUM(r.total_pay),0)").Scan(period.pay).Error; err != nil {
-			return nil, business.ErrStorage
+		identity := portraitIdentity(queryDB, scope.IDCardDedup)
+		selectSQL := "r.metric,COUNT(DISTINCT CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN " + identity + " END) AS current"
+		args := []any{begin, end}
+		if !scope.NoCompare {
+			selectSQL += ",COUNT(DISTINCT CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN " + identity + " END) AS previous"
+			args = append(args, previousBegin, previousEnd)
 		}
-	}
-	if scope.Range.Hourly {
-		if err := portraitHourly(db, scope, result); err != nil {
-			return nil, err
+		if err := summaryQuery("tab_sw_report_day_key", mode).Select(selectSQL, args...).Group("r.metric").Scan(&rows).Error; err != nil {
+			return business.ErrStorage
 		}
-	} else {
-		if err := portraitDaily(db, scope, mode, begin, end, result); err != nil {
-			return nil, err
+		current, previous := make([]portraitCountPO, 0, len(rows)), make([]portraitCountPO, 0, len(rows))
+		for _, row := range rows {
+			current = append(current, portraitCountPO{Metric: row.Metric, Count: row.Current})
+			previous = append(previous, portraitCountPO{Metric: row.Metric, Count: row.Previous})
 		}
-	}
+		result.Current, result.Previous = toPortraitCounts(current), toPortraitCounts(previous)
+		return nil
+	})
+	group.Go(func() error {
+		var amounts struct{ Current, Previous string }
+		selectSQL := "COALESCE(SUM(CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN r.total_pay ELSE 0 END),0) AS current"
+		args := []any{begin, end}
+		if !scope.NoCompare {
+			selectSQL += ",COALESCE(SUM(CASE WHEN r.stat_date >= ? AND r.stat_date <= ? THEN r.total_pay ELSE 0 END),0) AS previous"
+			args = append(args, previousBegin, previousEnd)
+		}
+		if err := summaryQuery("tab_sw_report_day", 0).Select(selectSQL, args...).Scan(&amounts).Error; err != nil {
+			return business.ErrStorage
+		}
+		result.CurrentPay = amounts.Current
+		if !scope.NoCompare {
+			result.PreviousPay = amounts.Previous
+		}
+		return nil
+	})
+	group.Go(func() error {
+		if scope.Range.Hourly {
+			return portraitHourly(queryDB, scope, result)
+		}
+		return portraitDaily(queryDB, scope, mode, begin, end, result)
+	})
 	if scope.RankLimit > 0 {
-		register, err := portraitRankQuery(db, scope, mode, begin, end, false)
-		if err != nil {
-			return nil, err
-		}
-		result.RegisterRanks = register
-		pay, err := portraitRankQuery(db, scope, 0, begin, end, true)
-		if err != nil {
-			return nil, err
-		}
-		result.PayRanks = pay
+		group.Go(func() error {
+			register, err := portraitRankQuery(queryDB, scope, mode, begin, end, false)
+			result.RegisterRanks = register
+			return err
+		})
+		group.Go(func() error {
+			pay, err := portraitRankQuery(queryDB, scope, 0, begin, end, true)
+			result.PayRanks = pay
+			return err
+		})
+	}
+	if err := group.Wait(); err != nil {
+		return nil, err
 	}
 	return result, nil
 }
@@ -247,11 +274,17 @@ func portraitDaily(db *gorm.DB, scope business.PortraitReportScope, mode int, be
 }
 func portraitHourly(db *gorm.DB, scope business.PortraitReportScope, result *business.PortraitStatistics) error {
 	start, end := scope.Range.Begin.Unix(), scope.Range.End.AddDate(0, 0, 1).Unix()
+	// register_time 是历史字符串列；数值比较不能利用其字符串索引。
+	// 复用已有数值生成列和时间范围索引，未迁移的库保留 CAST 回退。
+	registerTime, err := registrationTimeExpression(db)
+	if err != nil {
+		return err
+	}
 	for _, metric := range []struct {
 		table, column, aggregate, condition string
 		series                              map[string]string
-	}{{"tab_user", "register_time", "COUNT(*)", "(r.puid = 0 OR r.puid IS NULL)", result.RegisterSeries}, {"tab_spend", "pay_time", "COALESCE(SUM(r.pay_amount),0)", "r.pay_status = 1", result.PaySeries}} {
-		q := db.Table(metric.table+" r").Where("r."+metric.column+" >= ? AND r."+metric.column+" < ?", start, end).Where(metric.condition).Where("(r.source_db,r.promote_id) IN (?)", portraitPromoteScope(db, scope))
+	}{{"tab_user", registerTime, "COUNT(*)", "(r.puid = 0 OR r.puid IS NULL)", result.RegisterSeries}, {"tab_spend", "r.pay_time", "COALESCE(SUM(r.pay_amount),0)", "r.pay_status = 1", result.PaySeries}} {
+		q := db.Table(metric.table+" r").Where(metric.column+" >= ? AND "+metric.column+" < ?", start, end).Where(metric.condition).Where("(r.source_db,r.promote_id) IN (?)", portraitPromoteScope(db, scope))
 		if len(scope.Sources) > 0 {
 			q = q.Where("r.source_db IN ?", scope.Sources)
 		}
@@ -261,12 +294,12 @@ func portraitHourly(db *gorm.DB, scope business.PortraitReportScope, result *bus
 		}
 		if db.Dialector.Name() == "mysql" {
 			// 保留原 Mapper 的数据库会话时区口径，最终实库验收同时核对时区配置。
-			hour := "HOUR(FROM_UNIXTIME(r." + metric.column + "))"
+			hour := "HOUR(FROM_UNIXTIME(" + metric.column + "))"
 			if err := q.Select(hour + " AS hour," + metric.aggregate + " AS value").Group(hour).Scan(&rows).Error; err != nil {
 				return business.ErrStorage
 			}
 		} else {
-			if err := q.Select("CAST((r."+metric.column+" - ?) / 3600 AS INTEGER) AS hour,"+metric.aggregate+" AS value", start).Group("hour").Scan(&rows).Error; err != nil {
+			if err := q.Select("CAST(("+metric.column+" - ?) / 3600 AS INTEGER) AS hour,"+metric.aggregate+" AS value", start).Group("hour").Scan(&rows).Error; err != nil {
 				return business.ErrStorage
 			}
 		}
@@ -3,10 +3,13 @@ package business
 import (
 	"context"
 	"math/big"
+	"reflect"
+	"strings"
 	"testing"
 	"time"

 	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
 	"kra/internal/biz/business"
 )

@@ -151,3 +154,117 @@ func TestPortraitCountLimitAppliesBeforeTopN(t *testing.T) {
 		t.Fatalf("人数归一上限必须先于排行截取: %+v %v", stats, err)
 	}
 }
+
+func TestPortraitComparisonKeepsPeriodDedupAndExcludesGaps(t *testing.T) {
+	db, repo, scope := preparePortraitData(t)
+	for _, sql := range []string{
+		"INSERT INTO tab_sw_report_day_key VALUES('lx',100,1,1,'2026-10-01',1,'A'),('lx',101,1,1,'2026-10-02',1,'A'),('hw',100,1,1,'2026-10-03',1,'P'),('lx',100,1,1,'2026-10-15',1,'gap'),('lx',100,1,1,'2026-10-03',2,'active'),('lx',100,1,1,'2026-10-03',3,'payer'),('lx',100,1,1,'2026-10-03',4,'new-payer')",
+		"INSERT INTO tab_sw_report_day VALUES('lx',100,1,0,'2026-10-03',2.20),('lx',100,1,0,'2026-10-15',99999.99)",
+	} {
+		if err := db.Exec(sql).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	stats, err := repo.Statistics(context.Background(), scope)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wantPrevious := business.PortraitCounts{Register: 2, Active: 1, Pay: 1, NewPay: 1}
+	if stats.Current.Register != 3 || stats.Previous != wantPrevious || portraitTestMoney(stats.PreviousPay) != "3.30" || portraitTestMoney(stats.CurrentPay) != "2058.60" {
+		t.Fatalf("合并扫描改变了周期内去重或混入同期之间的日期: %+v", stats)
+	}
+	scope.NoCompare = true
+	withoutCompare, err := repo.Statistics(context.Background(), scope)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if withoutCompare.Current != stats.Current || withoutCompare.CurrentPay != stats.CurrentPay || withoutCompare.Previous != (business.PortraitCounts{}) || portraitTestMoney(withoutCompare.PreviousPay) != "0.00" {
+		t.Fatalf("关闭对比仍然返回了上期统计: %+v", withoutCompare)
+	}
+	scope.NoCompare = false
+	scope.Range.PreviousBegin, scope.Range.PreviousEnd = scope.Range.Begin, scope.Range.End
+	overlap, err := repo.Statistics(context.Background(), scope)
+	if err != nil || overlap.Current != overlap.Previous || overlap.CurrentPay != overlap.PreviousPay {
+		t.Fatalf("重叠周期必须独立统计: %+v %v", overlap, err)
+	}
+}
+
+type portraitSQLCapture struct {
+	logger.Interface
+	queries []string
+}
+
+func (l *portraitSQLCapture) Trace(_ context.Context, _ time.Time, fc func() (string, int64), _ error) {
+	sql, _ := fc()
+	if strings.Contains(sql, "HOUR(FROM_UNIXTIME(") {
+		l.queries = append(l.queries, sql)
+	}
+}
+
+func TestMySQLPortraitHourlyUsesRegistrationRangeAndPreservesLegacyValues(t *testing.T) {
+	db := openRehearsalMySQL(t)
+	pool, _ := db.DB()
+	pool.SetMaxOpenConns(1)
+	for _, sql := range []string{
+		"SET SESSION time_zone='+08:00'",
+		"CREATE TABLE tab_promote(id INT,source_db VARCHAR(10),busier_id INT,promote_level INT,PRIMARY KEY(id,source_db),KEY(source_db,busier_id))",
+		"INSERT INTO tab_promote VALUES(100,'lx',14,1),(100,'hw',15,1)",
+		"CREATE TABLE tab_user(id INT,source_db VARCHAR(10),promote_id INT,puid INT,register_time VARCHAR(11),PRIMARY KEY(id,source_db),KEY(promote_id,puid),KEY(register_time))",
+		"INSERT INTO tab_user VALUES(1,'lx',100,0,'1793638800'),(2,'lx',100,NULL,'01793638800'),(3,'lx',100,0,'1793638800 '),(4,'lx',100,1,'1793638800'),(5,'hw',100,0,'1793638800'),(6,'lx',100,0,'1793721600'),(7,'lx',100,0,'1793635199'),(8,'lx',100,0,'1793635200')",
+		"CREATE TABLE tab_spend(source_db VARCHAR(10),promote_id INT,pay_status INT,pay_time INT,pay_amount DECIMAL(20,2))",
+		"INSERT INTO tab_spend VALUES('lx',100,1,1793638800,1.10),('lx',100,0,1793638800,999.00),('hw',100,1,1793638800,999.00),('lx',100,1,1793721600,999.00)",
+	} {
+		if err := db.Exec(sql).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	oldUsers := make([]map[string]any, 2000)
+	for i := range oldUsers {
+		oldUsers[i] = map[string]any{"id": i + 100, "source_db": "lx", "promote_id": 100 + i%200, "puid": 0, "register_time": "1793000000"}
+	}
+	if err := db.Table("tab_user").CreateInBatches(oldUsers, 200).Error; err != nil {
+		t.Fatal(err)
+	}
+	channels := make([]map[string]any, 199)
+	for i := range channels {
+		channels[i] = map[string]any{"id": i + 101, "source_db": "lx", "busier_id": 14, "promote_level": 1}
+	}
+	if err := db.Table("tab_promote").CreateInBatches(channels, 200).Error; err != nil {
+		t.Fatal(err)
+	}
+	day := time.Date(2026, 11, 3, 0, 0, 0, 0, time.FixedZone("CST", 8*3600))
+	scope := business.PortraitReportScope{Pairs: []business.PlatformKey{{SourceDB: "lx", BusierID: 14}}, Sources: []string{"lx", "hw"}, IncludeSub: true, Range: business.PortraitRange{Begin: day, End: day, Hourly: true}}
+	read := func(queryDB *gorm.DB) *business.PortraitStatistics {
+		t.Helper()
+		out := &business.PortraitStatistics{RegisterSeries: map[string]string{}, PaySeries: map[string]string{}}
+		if err := portraitHourly(queryDB, scope, out); err != nil {
+			t.Fatal(err)
+		}
+		return out
+	}
+	before := read(db)
+	if !reflect.DeepEqual(before.RegisterSeries, map[string]string{"0": "1", "1": "3"}) || portraitTestMoney(before.PaySeries["1"]) != "1.10" || len(before.PaySeries) != 1 {
+		t.Fatalf("旧时间字符串、主账号、跨平台或零点边界口径错误: %+v", before)
+	}
+	if err := migrateRegistrationIndex(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Exec("ANALYZE TABLE tab_user,tab_promote").Error; err != nil {
+		t.Fatal(err)
+	}
+	capture := &portraitSQLCapture{Interface: logger.Default}
+	after := read(db.Session(&gorm.Session{Logger: capture}))
+	if !reflect.DeepEqual(before, after) || len(capture.queries) != 2 {
+		t.Fatalf("已有索引和未迁移时的结果不同: before=%+v after=%+v", before, after)
+	}
+	var plan []struct{ Table, Key, Type string }
+	if err := db.Raw("EXPLAIN " + capture.queries[0]).Scan(&plan).Error; err != nil {
+		t.Fatal(err)
+	}
+	for _, step := range plan {
+		if step.Table == "r" && step.Key == "idx_sw_user_registration" && step.Type == "range" {
+			return
+		}
+	}
+	t.Fatalf("小时注册趋势没有按数值时间索引做范围查找: %+v", plan)
+}