From 475e3d191d7d77765760f8875e98a3287b157568 Mon Sep 17 00:00:00 2001 From: zhx Date: Sat, 11 Jul 2026 02:32:29 +0800 Subject: [PATCH] fix: optimize app tracking funnel aggregation --- .../mysql/initdb/001_statistics_service.sql | 1 + .../internal/storage/mysql/query.go | 193 ++++++++++-------- .../internal/storage/mysql/query_test.go | 53 ++++- .../internal/storage/mysql/repository.go | 3 + 4 files changed, 164 insertions(+), 86 deletions(-) diff --git a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql index 6a3749ca..23a95e0a 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -48,6 +48,7 @@ CREATE TABLE IF NOT EXISTS app_tracking_events ( KEY idx_app_tracking_user (app_code, user_id, occurred_at_ms), KEY idx_app_tracking_device (app_code, device_id, occurred_at_ms), KEY idx_app_tracking_event_window (app_code, event_name, occurred_at_ms, identity_key, device_id), + KEY idx_app_tracking_time_window (app_code, occurred_at_ms, event_name, identity_key, device_id), KEY idx_app_tracking_retention_identity (app_code, stat_day, identity_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='App 埋点原始明细表'; diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index f91432ee..2cf79deb 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -752,17 +752,13 @@ func (r *Repository) QueryAppTrackingFunnel(ctx context.Context, query AppTracki } out.Steps = buildAppTrackingFunnelSteps(steps) out.Totals = appTrackingTotals(out.Steps) - d1Base, d1Users, err := r.queryAppTrackingD1Total(ctx, app, startMS, endMS, query.CountryID, query.RegionID, out.RegionIDs) + d1Base, d1Users, cohorts, err := r.queryAppTrackingD1Cohorts(ctx, app, startMS, endMS, query.CountryID, query.RegionID, out.RegionIDs) if err != nil { return AppTrackingFunnel{}, err } out.Totals.D1RetentionBaseUsers = d1Base out.Totals.D1RetentionUsers = d1Users out.Totals.D1RetentionRate = ratio(d1Users, d1Base) - cohorts, err := r.queryAppTrackingD1Cohorts(ctx, app, startMS, endMS, query.CountryID, query.RegionID, out.RegionIDs) - if err != nil { - return AppTrackingFunnel{}, err - } out.Cohorts = cohorts return out, nil } @@ -2097,7 +2093,7 @@ func (r *Repository) queryAppTrackingFunnelSteps(ctx context.Context, app string } rows, err := r.db.QueryContext(ctx, ` SELECT event_name, COUNT(*), COUNT(DISTINCT `+appTrackingIdentitySQL("")+`), COUNT(DISTINCT NULLIF(device_id, '')) - FROM app_tracking_events + FROM app_tracking_events FORCE INDEX (idx_app_tracking_time_window) WHERE `+filter+` AND event_name IN (`+placeholders+`) GROUP BY event_name`, args...) if err != nil { @@ -2172,97 +2168,119 @@ func appTrackingTotals(steps []AppTrackingFunnelStep) AppTrackingFunnelTotals { } } -func (r *Repository) queryAppTrackingD1Total(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (int64, int64, error) { - filter, args := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) +func (r *Repository) queryAppTrackingD1Cohorts(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (int64, int64, []AppTrackingFunnelCohort, error) { + filter, filterArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) + totalArgs := append([]any{}, filterArgs...) + dimensionSQL, dimensionArgs := appTrackingCohortUnionSQL(appTrackingD1CohortSourceSQL, filter, filterArgs, "source.stat_day, source.identity, source.value") + args := append(totalArgs, dimensionArgs...) args = append(args, app) - row := r.db.QueryRowContext(ctx, ` - WITH base AS ( + // retention_base 对每个 stat_day+identity 只判断一次次日活跃，再与五个维度关联；旧实现会为每个维度重复执行同一 EXISTS。 + rows, err := r.db.QueryContext(ctx, ` + WITH total_base AS ( SELECT DISTINCT stat_day, `+appTrackingIdentitySQL("")+` AS identity FROM app_tracking_events WHERE `+filter+` AND event_name = 'login_success' + ), + dimension_base AS (`+dimensionSQL+`), + retention_base AS ( + SELECT total_base.stat_day, total_base.identity, + CASE WHEN EXISTS ( + SELECT 1 FROM app_tracking_events n + WHERE n.app_code = ? AND n.stat_day = DATE_ADD(total_base.stat_day, INTERVAL 1 DAY) + AND `+appTrackingIdentitySQL("n")+` = total_base.identity + LIMIT 1 + ) THEN 1 ELSE 0 END AS retained + FROM total_base + ), + dimension_aggregated AS ( + SELECT dimension_base.dimension, dimension_base.value, COUNT(*) AS base_users, + COALESCE(SUM(retention_base.retained), 0) AS retained_users + FROM dimension_base + JOIN retention_base ON retention_base.stat_day = dimension_base.stat_day + AND retention_base.identity = dimension_base.identity + GROUP BY dimension_base.dimension, dimension_base.value + ), + dimension_ranked AS ( + SELECT dimension, value, base_users, retained_users, + ROW_NUMBER() OVER (PARTITION BY dimension ORDER BY base_users DESC, value ASC) AS dimension_rank + FROM dimension_aggregated ) - SELECT COUNT(*), - COALESCE(SUM(CASE WHEN EXISTS ( - SELECT 1 FROM app_tracking_events n - WHERE n.app_code = ? AND n.stat_day = DATE_ADD(base.stat_day, INTERVAL 1 DAY) - AND `+appTrackingIdentitySQL("n")+` = base.identity - LIMIT 1 - ) THEN 1 ELSE 0 END),0) - FROM base`, args...) - var baseUsers, retainedUsers int64 - if err := row.Scan(&baseUsers, &retainedUsers); err != nil { - return 0, 0, err - } - return baseUsers, retainedUsers, nil -} - -func (r *Repository) queryAppTrackingD1Cohorts(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) ([]AppTrackingFunnelCohort, error) { - out := []AppTrackingFunnelCohort{} - for _, def := range appTrackingCohortDefs { - rows, err := r.queryAppTrackingD1CohortDimension(ctx, app, startMS, endMS, countryID, regionID, regionIDs, def) - if err != nil { - return nil, err - } - stepCounts, err := r.queryAppTrackingCohortSteps(ctx, app, startMS, endMS, countryID, regionID, regionIDs, def) - if err != nil { - return nil, err - } - for index := range rows { - rows[index].Steps = buildAppTrackingCohortSteps(stepCounts[rows[index].Value]) - } - out = append(out, rows...) - } - return out, nil -} - -func (r *Repository) queryAppTrackingD1CohortDimension(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64, def appTrackingCohortDef) ([]AppTrackingFunnelCohort, error) { - filter, filterArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) - sourceSQL := appTrackingD1CohortSourceSQL(def.Key, filter) - args := appTrackingCohortSourceArgs(def.Key, filterArgs) - args = append(args, app) - rows, err := r.db.QueryContext(ctx, ` - WITH base AS (`+sourceSQL+`) - SELECT value, COUNT(*), - COALESCE(SUM(CASE WHEN EXISTS ( - SELECT 1 FROM app_tracking_events n - WHERE n.app_code = ? AND n.stat_day = DATE_ADD(base.stat_day, INTERVAL 1 DAY) - AND `+appTrackingIdentitySQL("n")+` = base.identity - LIMIT 1 - ) THEN 1 ELSE 0 END),0) - FROM base - GROUP BY value - ORDER BY COUNT(*) DESC, value ASC - LIMIT 50`, args...) + SELECT '' AS dimension, '' AS value, COUNT(*) AS base_users, + COALESCE(SUM(retained), 0) AS retained_users, 0 AS dimension_rank + FROM retention_base + UNION ALL + SELECT dimension, value, base_users, retained_users, dimension_rank + FROM dimension_ranked + WHERE dimension_rank <= 50 + ORDER BY dimension ASC, dimension_rank ASC`, args...) if err != nil { - return nil, err + return 0, 0, nil, err } defer rows.Close() - out := []AppTrackingFunnelCohort{} + var baseUsers int64 + var retainedUsers int64 + byDimension := map[string][]AppTrackingFunnelCohort{} + labels := map[string]string{} + for _, def := range appTrackingCohortDefs { + labels[def.Key] = def.Label + } for rows.Next() { + var dimension string var value sql.NullString - var baseUsers, retainedUsers int64 - if err := rows.Scan(&value, &baseUsers, &retainedUsers); err != nil { - return nil, err + var rowBaseUsers, rowRetainedUsers, dimensionRank int64 + if err := rows.Scan(&dimension, &value, &rowBaseUsers, &rowRetainedUsers, &dimensionRank); err != nil { + return 0, 0, nil, err + } + if dimension == "" { + baseUsers = rowBaseUsers + retainedUsers = rowRetainedUsers + continue } cohortValue := normalizeCohortValue(value.String) - out = append(out, AppTrackingFunnelCohort{ - Dimension: def.Key, - DimensionLabel: def.Label, + byDimension[dimension] = append(byDimension[dimension], AppTrackingFunnelCohort{ + Dimension: dimension, + DimensionLabel: labels[dimension], Value: cohortValue, Label: cohortValue, - BaseUsers: baseUsers, - D1RetentionUsers: retainedUsers, - D1RetentionRate: ratio(retainedUsers, baseUsers), + BaseUsers: rowBaseUsers, + D1RetentionUsers: rowRetainedUsers, + D1RetentionRate: ratio(rowRetainedUsers, rowBaseUsers), Steps: []AppTrackingCohortStepCount{}, }) } - return out, rows.Err() + if err := rows.Err(); err != nil { + return 0, 0, nil, err + } + stepCounts, err := r.queryAppTrackingCohortSteps(ctx, app, startMS, endMS, countryID, regionID, regionIDs) + if err != nil { + return 0, 0, nil, err + } + out := []AppTrackingFunnelCohort{} + for _, def := range appTrackingCohortDefs { + for _, cohort := range byDimension[def.Key] { + cohort.Steps = buildAppTrackingCohortSteps(stepCounts[def.Key][cohort.Value]) + out = append(out, cohort) + } + } + return baseUsers, retainedUsers, out, nil } -func (r *Repository) queryAppTrackingCohortSteps(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64, def appTrackingCohortDef) (map[string]map[string]int64, error) { +type appTrackingCohortSourceBuilder func(dimension string, filter string) string + +func appTrackingCohortUnionSQL(builder appTrackingCohortSourceBuilder, filter string, filterArgs []any, columns string) (string, []any) { + parts := make([]string, 0, len(appTrackingCohortDefs)) + args := []any{} + for _, def := range appTrackingCohortDefs { + parts = append(parts, "SELECT ? AS dimension, "+columns+" FROM ("+builder(def.Key, filter)+") source") + args = append(args, def.Key) + args = append(args, appTrackingCohortSourceArgs(def.Key, filterArgs)...) + } + return strings.Join(parts, "\nUNION ALL\n"), args +} + +func (r *Repository) queryAppTrackingCohortSteps(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (map[string]map[string]map[string]int64, error) { filter, filterArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) - cohortSQL := appTrackingStepCohortSourceSQL(def.Key, filter) - args := appTrackingCohortSourceArgs(def.Key, filterArgs) + cohortSQL, args := appTrackingCohortUnionSQL(appTrackingStepCohortSourceSQL, filter, filterArgs, "source.identity, source.value") eventFilter, eventArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) args = append(args, eventArgs...) names := appTrackingEventNames() @@ -2274,30 +2292,35 @@ func (r *Repository) queryAppTrackingCohortSteps(ctx context.Context, app string WITH cohort AS (`+cohortSQL+`), events AS ( SELECT DISTINCT `+appTrackingIdentitySQL("")+` AS identity, event_name - FROM app_tracking_events + -- 同一窗口会覆盖几乎全部漏斗事件；强制时间优先覆盖索引，避免优化器只按 app_code 扫描主键后再回表去重。 + FROM app_tracking_events FORCE INDEX (idx_app_tracking_time_window) WHERE `+eventFilter+` AND event_name IN (`+placeholders+`) ) - SELECT cohort.value, events.event_name, COUNT(DISTINCT cohort.identity) + SELECT cohort.dimension, cohort.value, events.event_name, COUNT(DISTINCT cohort.identity) FROM cohort JOIN events ON events.identity = cohort.identity - GROUP BY cohort.value, events.event_name`, args...) + GROUP BY cohort.dimension, cohort.value, events.event_name`, args...) if err != nil { return nil, err } defer rows.Close() - out := map[string]map[string]int64{} + out := map[string]map[string]map[string]int64{} for rows.Next() { + var dimension string var value sql.NullString var eventName string var userCount int64 - if err := rows.Scan(&value, &eventName, &userCount); err != nil { + if err := rows.Scan(&dimension, &value, &eventName, &userCount); err != nil { return nil, err } cohortValue := normalizeCohortValue(value.String) - if out[cohortValue] == nil { - out[cohortValue] = map[string]int64{} + if out[dimension] == nil { + out[dimension] = map[string]map[string]int64{} } - out[cohortValue][eventName] = userCount + if out[dimension][cohortValue] == nil { + out[dimension][cohortValue] = map[string]int64{} + } + out[dimension][cohortValue][eventName] = userCount } return out, rows.Err() } diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index d7f08c33..b320c072 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -84,6 +84,31 @@ func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(t *testing.T) { } t.Cleanup(func() { _ = repository.Close() }) + // 漏斗读取按 App+毫秒窗口裁剪，索引顺序属于性能契约；event_name 放在时间前会让全事件漏斗退化为按 App 扫描。 + indexRows, err := repository.db.QueryContext(ctx, ` + SELECT COLUMN_NAME + FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'app_tracking_events' AND INDEX_NAME = 'idx_app_tracking_time_window' + ORDER BY SEQ_IN_INDEX`) + if err != nil { + t.Fatalf("query app tracking time window index: %v", err) + } + var indexColumns []string + for indexRows.Next() { + var column string + if err := indexRows.Scan(&column); err != nil { + _ = indexRows.Close() + t.Fatalf("scan app tracking time window index: %v", err) + } + indexColumns = append(indexColumns, column) + } + if err := indexRows.Close(); err != nil { + t.Fatalf("close app tracking time window index rows: %v", err) + } + if got, want := strings.Join(indexColumns, ","), "app_code,occurred_at_ms,event_name,identity_key,device_id"; got != want { + t.Fatalf("app tracking time window index mismatch: got=%s want=%s", got, want) + } + occurredAt := time.Date(2026, 7, 2, 23, 59, 59, 0, time.UTC).UnixMilli() result, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{ { @@ -183,10 +208,17 @@ func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) { } } user1Props := `{"country":"US","channel":"ads","login_method":"google"}` + user1AffiliateProps := `{"country":"US","channel":"affiliate","login_method":"google"}` user2Props := `{"country":"BR","channel":"organic","login_method":"phone"}` + affiliateLogin := event(1, "dev-1", "login_success", 1500*time.Millisecond, user1AffiliateProps) + affiliateLogin.EventID += "-affiliate" + retainedOutsideFilter := event(1, "dev-1", "send_message", 24*time.Hour+time.Second, user1Props) + // D1 表示次日任意活跃，不能继承首日查询的区域过滤，否则用户次日跨区会被错误判定流失。 + retainedOutsideFilter.RegionID = 999 events := []AppTrackingEvent{ event(1, "dev-1", "login_start", 0, user1Props), event(1, "dev-1", "login_success", time.Second, user1Props), + affiliateLogin, event(1, "dev-1", "profile_start", 2*time.Second, user1Props), event(1, "dev-1", "profile_complete", 3*time.Second, user1Props), event(1, "dev-1", "home_room_impression", 4*time.Second, user1Props), @@ -197,7 +229,7 @@ func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) { event(1, "dev-1", "send_message", 4*time.Minute, user1Props), event(1, "dev-1", "follow_host", 5*time.Minute, user1Props), event(1, "dev-1", "push_permission", 6*time.Minute, user1Props), - event(1, "dev-1", "send_message", 24*time.Hour+time.Second, user1Props), + retainedOutsideFilter, event(2, "dev-2", "login_start", 0, user2Props), event(2, "dev-2", "login_success", time.Second, user2Props), event(2, "dev-2", "home_room_impression", 2*time.Second, user2Props), @@ -252,6 +284,25 @@ func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) { if us.BaseUsers != 1 || us.D1RetentionUsers != 1 || math.Abs(us.D1RetentionRate-1) > 0.0001 { t.Fatalf("unexpected US cohort: %+v", us) } + cohortSteps := func(item AppTrackingFunnelCohort) map[string]int64 { + steps := map[string]int64{} + for _, step := range item.Steps { + steps[step.EventName] = step.UserCount + } + return steps + } + usSteps := cohortSteps(us) + if usSteps["login_success"] != 1 || usSteps["room_join_success"] != 1 || usSteps["send_message"] != 1 { + t.Fatalf("unexpected US cohort steps: %+v", usSteps) + } + brSteps := cohortSteps(cohort("country", "BR")) + if brSteps["login_success"] != 1 || brSteps["room_join_success"] != 0 || brSteps["room_join_fail"] != 1 { + t.Fatalf("unexpected BR cohort steps: %+v", brSteps) + } + // 同一 identity+day 的相同国家只计一次；同日出现两个渠道值时，各渠道仍按既有口径分别计入。 + if cohort("country", "US").BaseUsers != 1 || cohort("channel", "ads").BaseUsers != 1 || cohort("channel", "affiliate").BaseUsers != 1 { + t.Fatalf("multi-value cohort deduplication mismatch: %+v", funnel.Cohorts) + } stay := cohort("first_room_stay", "3m-10m") if stay.BaseUsers != 1 || stay.D1RetentionUsers != 1 { t.Fatalf("unexpected stay cohort: %+v", stay) diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index 83c8669c..fefbc6be 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -139,6 +139,7 @@ func (r *Repository) Migrate(ctx context.Context) error { KEY idx_app_tracking_user (app_code, user_id, occurred_at_ms), KEY idx_app_tracking_device (app_code, device_id, occurred_at_ms), KEY idx_app_tracking_event_window (app_code, event_name, occurred_at_ms, identity_key, device_id), + KEY idx_app_tracking_time_window (app_code, occurred_at_ms, event_name, identity_key, device_id), KEY idx_app_tracking_retention_identity (app_code, stat_day, identity_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_app_day_country ( @@ -457,6 +458,8 @@ func (r *Repository) Migrate(ctx context.Context) error { {"app_tracking_events", "idx_app_tracking_user", []string{"app_code", "user_id", "occurred_at_ms"}}, {"app_tracking_events", "idx_app_tracking_device", []string{"app_code", "device_id", "occurred_at_ms"}}, {"app_tracking_events", "idx_app_tracking_event_window", []string{"app_code", "event_name", "occurred_at_ms", "identity_key", "device_id"}}, + // 漏斗步骤会一次读取全部事件名；时间列必须紧跟 app_code，避免 event_name IN 覆盖大多数事件时退化为按 App 扫描主键。 + {"app_tracking_events", "idx_app_tracking_time_window", []string{"app_code", "occurred_at_ms", "event_name", "identity_key", "device_id"}}, {"app_tracking_events", "idx_app_tracking_retention_identity", []string{"app_code", "stat_day", "identity_key"}}, {"stat_app_day_country", "idx_stat_app_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}}, {"stat_user_day_activity", "idx_stat_user_day_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},