5.5 配套代码：数据集生成与分布验证

// src/main/kotlin/dataset/ZipfGenerator.kt
package dataset

import kotlin.math.pow
import kotlin.random.Random

/**
 * 近似齐夫分布采样器。
 *
 * alpha 的含义：
 *   alpha = 0    → 退化为均匀分布
 *   alpha ≈ 1    → 接近真实齐夫
 *   alpha = 3    → 更集中（前 1% 承担绝大多数访问）
 *
 * 真实业务的 alpha 通常通过生产访问日志的 log-log 图斜率估算。
 */
class ZipfSampler(
    private val n: Int,
    private val alpha: Double = 1.2,
    seed: Int = 42,
) {
    private val rnd = Random(seed)

    fun next(): Int {
        val u = rnd.nextDouble()
        return (1 + (n * u.pow(alpha)).toInt()).coerceIn(1, n)
    }
}

/**
 * 让数据分布可控的生成器：可以指定「前 x% 的 key 承担 y% 的访问」。
 * 比直接用 alpha 更直观——因为业务方通常用这种方式描述热点。
 */
class HotspotSampler(
    private val totalKeys: Int,
    private val hotKeyRatio: Double = 0.01,      // 前 1% 的 key
    private val hotTrafficRatio: Double = 0.40,  // 承担 40% 的访问
    seed: Int = 42,
) {
    private val rnd = Random(seed)
    private val hotKeyCount = (totalKeys * hotKeyRatio).toInt().coerceAtLeast(1)

    fun next(): Int = if (rnd.nextDouble() < hotTrafficRatio) {
        // 热点区（前 hotKeyRatio 的 key）
        rnd.nextInt(1, hotKeyCount + 1)
    } else {
        // 长尾区（其余 key）
        rnd.nextInt(hotKeyCount + 1, totalKeys + 1)
    }
}

fun main() {
    val N = 1_000_000
    val SAMPLES = 200_000

    println("═══ 分布对比（%d 个 key，%d 次采样）═══".format(N, SAMPLES))
    println()

    for ((label, sampler) in listOf(
        "均匀分布（alpha=0）" to ZipfSampler(N, alpha = 0.0),
        "轻微集中（alpha=1.0）" to ZipfSampler(N, alpha = 1.0),
        "强热点（alpha=3.0）" to ZipfSampler(N, alpha = 3.0),
    )) {
        val counts = IntArray(10)      // 把 key 空间分成 10 段，统计各段访问占比
        repeat(SAMPLES) {
            val key = sampler.next()
            val segment = ((key - 1).toLong() * 10 / N).toInt().coerceIn(0, 9)
            counts[segment]++
        }
        println("%-22s 前10%%的key承担 %5.1f%% 的访问".format(
            label, counts[0] * 100.0 / SAMPLES))
        println("  各段分布: " + counts.joinToString(" ") { "%5.1f%%".format(it * 100.0 / SAMPLES) })
        println()
    }

    // 用 HotspotSampler 精确控制热点
    println("═══ 精确控制热点 ═══")
    val hs = HotspotSampler(totalKeys = N, hotKeyRatio = 0.01, hotTrafficRatio = 0.40)
    val hot = IntArray(1)
    val cold = IntArray(1)
    repeat(SAMPLES) {
        if (hs.next() <= N / 100) hot[0]++ else cold[0]++
    }
    println("HotspotSampler(前1%%的key, 40%%流量):")
    println("  实测：前 1%% 的 key 承担 %.1f%% 的访问（目标 40%%）".format(hot[0] * 100.0 / SAMPLES))
}

# tools/verify_distribution.py <access_log.csv> [--top-pct 1]
"""
验证数据集的分布是否符合预期。

输入格式：CSV，每行一个被访问的 key（或 key 的哈希）
输出：热点集中度报告 + log-log 斜率估计（alpha）
"""
import csv
import math
import sys
from collections import Counter


def load_keys(path):
    keys = []
    with open(path) as f:
        reader = csv.reader(f)
        for row in reader:
            if row:
                keys.append(row[0].strip())
    return keys


def concentration_report(counter, top_pcts=(0.001, 0.01, 0.1)):
    total = sum(counter.values())
    distinct = len(counter)
    print(f"总访问次数 : {total:,}")
    print(f"不同 key 数 : {distinct:,}")
    print()

    print("热点集中度：")
    sorted_counts = sorted(counter.values(), reverse=True)
    for pct in top_pcts:
        k = max(1, int(distinct * pct))
        top_sum = sum(sorted_counts[:k])
        print(f"  前 {pct*100:5.1f}% 的 key 承担 {top_sum/total*100:5.1f}% 的访问"
              f"  （{k:,} 个 key）")
    print()


def estimate_alpha(counter):
    """
    用 log-log 回归估计齐夫指数 alpha。
    齐夫分布：freq(rank) ∝ rank^(-alpha)
    取对数后是直线，斜率 = -alpha
    """
    sorted_counts = sorted(counter.values(), reverse=True)
    # 取前 1000 个（尾部噪声大）
    points = [(math.log(i + 1), math.log(c)) for i, c in enumerate(sorted_counts[:1000]) if c > 0]
    if len(points) < 10:
        return None

    n = len(points)
    mx = sum(p[0] for p in points) / n
    my = sum(p[1] for p in points) / n
    num = sum((p[0] - mx) * (p[1] - my) for p in points)
    den = sum((p[0] - mx) ** 2 for p in points)
    slope = num / den if den else 0
    return -slope


def main():
    if len(sys.argv) < 2:
        print("usage: verify_distribution.py <access_log.csv> [--top-pct N]")
        sys.exit(1)

    path = sys.argv[1]
    keys = load_keys(path)
    if not keys:
        print("❌ 没有读到数据")
        return

    counter = Counter(keys)

    print("═" * 70)
    print(f"分布验证：{path}")
    print("═" * 70)
    print()

    concentration_report(counter)

    alpha = estimate_alpha(counter)
    print("齐夫指数估计：")
    if alpha:
        print(f"  alpha ≈ {alpha:.2f}")
        if alpha < 0.3:
            print("  → 接近均匀分布（不像真实业务）")
        elif alpha < 0.8:
            print("  → 轻微集中")
        elif alpha < 1.5:
            print("  → 接近真实齐夫分布 ✅")
        else:
            print("  → 高度集中（热点极强）")
    else:
        print("  样本不足，无法估计")
    print()

    print("判读要点：")
    print("  ① 前 1% 的 key 承担 30%~50% 的访问 → 接近真实业务")
    print("  ② 如果前 1% 只承担 1%~2% → 这是均匀分布，不真实")
    print("  ③ 用这个 alpha 配置你的压测脚本（不要用均匀随机）")
    print("═" * 70)


if __name__ == "__main__":
    main()

# ① 从 PostgreSQL 统计真实的 ID 访问分布（如果有访问日志表）
psql -c "
SELECT user_id, count(*) AS freq
FROM request_log
WHERE created_at > now() - interval '1 day'
GROUP BY user_id
ORDER BY freq DESC
LIMIT 1000;" > docs/experiments/E0x/results/top-keys.csv

# ② 直接导出为压测脚本用的 ID 列表（含重复，保持真实频率）
psql -tAc "
SELECT user_id FROM request_log
WHERE created_at > now() - interval '1 day'
ORDER BY random()
LIMIT 100000;" > docs/experiments/E0x/results/sampled-ids.txt

# ③ 在 k6 里从这个列表里随机取（保持真实分布）
# export default function () {
#   const ids = open('./sampled-ids.txt').split('\n');
#   const id = ids[Math.floor(Math.random() * ids.length)];
#   http.get(`${__BASE_URL}/users/${id}`);
# }

#!/usr/bin/env bash
# tools/verify-execution-plan.sh <QUERY> <SMALL_TABLE> <LARGE_TABLE>
#
# 对比同一查询在不同数据量下的执行计划，确认规模足够大。
set -uo pipefail

QUERY="${1:?usage: verify-execution-plan.sh <query> <table>}"
PG_CONN="${PG_CONN:-}"

echo "═══ 执行计划对比 ═══"
echo "查询: $QUERY"
echo

for SCALE in "10%" "50%" "100%"; do
  echo "── 数据量 $SCALE ──"

  # 用一个临时表模拟不同规模（或者用 LIMIT + 子查询，但注意这会改变计划）
  # 更可靠的做法是针对不同规模的测试库分别测
  psql "$PG_CONN" -c "EXPLAIN (ANALYZE, BUFFERS) $QUERY" 2>&1 | \
    grep -E "Seq Scan|Index Scan|Bitmap|actual time|rows=|Execution Time" | head -8

  echo
done

echo "判读："
echo "  如果三个规模的计划【相同】→ 规模足够，结论可迁移"
echo "  如果计划发生变化（如 10% 走索引、100% 走全表扫描）→ 规模不够，需要加大"
echo
echo "⚠️  严格做法：准备 10% / 50% / 100% 三份数据分别测试"
echo "    （本脚本的简化版只演示流程，实际需要多份数据）"