问题与控制单位：并行度究竟修改什么

引言：为什么“多开几个 task”不是一个简单的问题

并行度（Degree of Parallelism，DOP）必须连同控制单位讨论，task 数、pipeline driver 数和正在使用的线程数并不等价。在 ClickHouse、StarRocks、Doris 等大规模并行处理（Massively Parallel Processing，MPP）引擎中，pipeline 并发同时影响小查询延迟、大查询资源占用和集群并发；在 Spark 等批处理引擎中，Shuffle 后分区数是重要决策，但文件扫描分区、executor 数和同时运行 task 数仍是不同层次。

这个区别决定了调整成本：增加线程可能只涉及调度，修改 reader mapping 需要重组读取范围，改变已有 Hash 分区则可能重新搬运数据。因此，本文按信息可获得的时点，将机制分成三类：

• 运行前：执行前根据 estimate 给出初值；
• 运行中：本次执行用 actual 修正尚未完成的工作；
• 运行后：下次执行复用经过验证的事实。重点不是设计一个统一的 DOP advisor，而是看清不同系统为什么采用不同机制，以及这些机制能做什么、不能做什么。

明确概念：partition、task、driver 与 slot

工作分区数量决定可组织的任务，资源配额决定可同时执行的任务，两者之间还存在映射与排队。以下先统一术语，再讨论何时能修改它们。

四层并行度，以及独立的重试维度

不同系统说的“DOP”可能指向完全不同的对象。把文件切分、task 调度和线程池放在同一张表里比较，会产生伪等价。

例如，一个 Shuffle 有 1,000 个 bucket，执行器可以把它们合并成 100 个 consumer task；这 100 个 task 又可能只得到 20 个同时运行的槽位。另一方面，一个 MPP task 内可以实例化多个 pipeline driver，但 driver 通常由共享线程池调度，并不是一个 driver 永久占有一条线程。

分层关系可以概括为：

1逻辑工作单元 P
2  -> 按语义与输入映射形成 N 个 task
3  -> 调度器让其中 Q(t) 个 task 在时刻 t 获得执行机会
4  -> 每个 task 内又包含一个或多个 pipeline 的 driver 集合
5
6重试/投机增加 attempts；扩容改变资源容量；两者都不必修改 P。

• 对于“一个 partition 一个逻辑 task”的 Spark 常见路径，资源不足主要意味着多轮执行；
• 对于 Trino FTE，bucket 到 task 本来就可以是多对一；
• 对于 StarRocks，更多时候需要同时看 fragment placement 与 pipeline DOP。不存在跨系统通用的 DOP = worker 数 × 线程数。

bytes/task 实用，但单独用它会漏掉关键信息

按目标工作量切分能给出初值，但不能反映每份工作的计算成本和状态大小。部分初始策略可以抽象为：

这里的 work 可以是 bytes、rows 或估计成本。随后还要应用显式设置、Singleton、bucket mapping、物理拓扑与上下界。这是理解算法的结构，不是所有产品的实际公式。

bytes/task 容易获得、容易汇总，与 I/O 和内存相关。但相同 bytes 不代表相同工作：

• 窄行的高基数聚合可能 CPU 密集；
• 少量宽行可能内存密集；
• UDF 的单行计算成本可能差几个数量级；
• Join 输出还会放大输入。

仅凭 bytes 判定的 DOP，对 CPU-bound 聚合和宽表 Join 可能同时给出错误答案。

讨论一个 Stage 的时间，也不能只看总量除以 task 数。一个概念性分解是：

其中工作时间取决于可用资源与负载分布；倾斜、spill 和同步又会改变尾部。该式是分析框架，不是某个系统的 cost model。

• 并行度过低，可能造成利用率不足和单 task 状态过大；
• 并行度过高，则可能增加调度、网络请求、buffer、局部状态和输出小文件。更关键的是：单查询延迟最低的 DOP，不一定是共享集群吞吐最高的 DOP。

第一次机会：执行前给出怎样的基础值

初值必须在有限统计下满足语义和拓扑约束，因此不同系统会采用装箱、节点封顶或规模估计。源码结论以下文固定提交为准，在线产品文档的默认值还需对照实际部署版本。

Spark：扫描装箱与 Shuffle 分区是两条路径

扫描分区组织文件读取，Shuffle 分区组织重分布后的消费；二者输入信息不同，不能共用一个分区数解释。

Scan 并发度

Spark 的内置文件扫描不是简单用文件总大小除以 128 MiB。FilePartition.maxSplitBytes 同时考虑文件打开成本和期望并行度：

1totalBytes = sum(fileLength + openCostInBytes)
2
3minPartitionNum = files.minPartitionNum
4               or leafNodeDefaultParallelism
5               or SparkContext.defaultParallelism
6
7maxSplitBytes = min(
8    files.maxPartitionBytes,
9    max(openCostInBytes, totalBytes / minPartitionNum))

随后，可切分文件先形成 PartitionedFile，扫描路径按大小降序交给装箱逻辑。下面是 getFilePartitions 的关键节选：

 1// 节选；closePartition() 将当前文件集合封装为一个 FilePartition。
 2partitionedFiles.foreach { file =>
 3  if (currentSize + file.length > maxSplitBytes) {
 4    closePartition()
 5  }
 6
 7  // 文件放入后还累加固定打开成本，使大量小文件不至于无限装入一个 task。
 8  currentSize += file.length + openCostInBytes
 9  currentFiles += file
10}
11closePartition()

两处细节影响实际行为。

• 首先，检查当前文件是否放得下时使用 file.length，放入后再累加 open cost，因此 target 不是严格的物理字节上限。
• 其次，不可切分的大文件仍可能超过 target；配置 files.maxPartitionNum 后触发的是重新估算 split size 并装箱，不是把任何输入强制切成精确数量。

扫描 task 数因此由可切分性、文件布局与装箱目标共同决定，不能直接由 Shuffle 分区配置推导。

Shuffle 并发度

Shuffle 的 200 则主要是配置初值。它不是 Shuffle 写端 task 数，也不是同时运行的 task 数。显式 repartition/coalesce、SQL hints、已有 partitioning、bucket 与自定义 DataSource 都可能走不同路径。

源码与配置：FilePartition.scala、Spark SQL Performance Tuning。

PrestoDB 与 Trino：拓扑封顶与统计缩小

普通固定分布 fragment 的 task 数受执行节点映射限制，估计值不直接等于可独立排队的 task 数。Trino 的统计缩小还需放在这一执行模式下理解。

PrestoDB 普通 system FIXED fragment 的逻辑，从 SystemPartitioningHandle.getNodePartitionMap 可以概括为：

1P_cap = min(
2    fragment.partitionCount（如果显式存在）,
3    query.hash-partition-count,
4    stage.max-tasks-per-stage)
5
6selectedNodes = selectRandomNodes(P_cap)
7N_fixed_tasks <= min(P_cap, eligible execution nodes)

query.hash-partition-count 典型默认值为 100。因此，对这个特定路径，集群较小时常常近似为每个 eligible node 一个固定 task，而不是把 100 个 bucket 都作为独立可排队 task。SINGLE、SOURCE、connector bucket mapping 和 grouped execution 是另外的语义。task.concurrency 控制 task 内 driver 并发，不能拿来乘出所有 Stage 的 task 数。

源码与配置：Presto SystemPartitioningHandle、Presto properties。

Trino 在类似拓扑约束之外，还提供 DeterminePartitionCount：使用当前 query-wide 统计缩小 eligible remote repartition exchanges 的 partition count。

 1B = max(
 2    所有 TableScan / Values 的 estimated output bytes 之和,
 3    expanding Join / Union / 多输入 Exchange 的最大 estimated output bytes)
 4
 5R = 对 estimated rows 做同类汇总
 6
 7P_candidate = max(
 8    minPartitionCount,
 9    max(1, floor(B / minInputSizePerTask)),
10    max(1, floor(R / minInputRowsPerTask)),
11    floor(2 * B / queryMaxMemoryPerNode))

典型参数是 5 GiB/task 与 10,000,000 rows/task。普通执行的 hash partition 默认范围为 4–100；写查询的 minimum 为 50，而且默认不启用该自动缩小。TASK-retry 模式使用独立的 FTE partition 配置，不沿用普通执行上下界。

下面是方法的决策骨架，属于保留控制语义的 Java 伪代码：

 1// 没有 eligible Exchange、写路径未开启、统计缺失或存在危险放大节点：保留原计划。
 2if (!eligibleExchange || unsupportedWrite || unsafeExpansion || missingStats) {
 3    return originalPlan;
 4}
 5
 6int candidate = max(countByBytes, countByRows, countByMemory, minCount);
 7
 8// 这是缩小规则；不是把任意估计 clamp 到 maxCount 后强制写回。
 9if (candidate >= maxCount) {
10    return originalPlan;
11}
12
13// 普通执行的可行 task 数还受节点数约束，收益不足时不值得重新设置。
14if (!taskRetry && candidate * 2 >= maxPossibleTaskCount) {
15    return originalPlan;
16}
17
18return rewriteEligibleRemoteExchangesWithSameCount(candidate);

这段代码体现三个设计取舍：

• 同时看 rows 与 bytes：小行也可能有很高 CPU 成本，大行也可能有很高内存成本。
• 采用 query-wide 的保守规模：避免只按 leaf 扫描量低估中间 Join 放大，但不是逐 Stage 求最优 DOP。
• 把“不调整”作为正常结果：未知统计、UNNEST、部分不安全 Join 或收益不足时返回原计划，不用不可靠的数字覆盖已有策略。

其中 2 * B / memory 只是规则的内存近似，不是 Hash Table 峰值的严格证明，也不保证每个 task 不 spill。普通执行最终仍受 execution nodes 约束；FTE 的 task formation 则是下一节的另一套机制。

源码与配置：DeterminePartitionCount.java、Trino optimizer properties、Trino query management properties。

自动初值的公开边界：Databricks、BigQuery 与 Snowflake

产品公开“自动并行度”不等于公开内部 task 形成公式，必须区分可观察行为和不可见实现。以下仅陈述文档说明的作用域。

Databricks serverless notebooks/jobs 当前公开的 spark.sql.shuffle.partitions 默认是 auto。官方 AQE 文档将 Auto-Optimized Shuffle 描述为依据 query plan 与 input size 选择初始 partition 数，但没有公开完整成本公式。这里要区分两个时点：auto 在 Shuffle 前选初值，AQE 在本次执行中用真实结果修正。 不能把某个运行形态的默认值推广到所有 Runtime、SQL warehouse 与 serverless 产品。Databricks Spark configuration、Databricks AQE。

BigQuery 的计划可见 parallelInputs：扫描时可以代表 columnar segments，下游可以代表 Shuffle partitions。slots 则是抽象的计算资源单位。work units、requested slots 与实际分配的 slots 没有公开的一一公式；排队与公平调度会继续改变实际资源。应当确认“会选择 initial parallelism”，但不能通过 slot 图反推出内部 task 数算法。BigQuery query plan explanation、Understand slots。

Snowflake 的 micro-partitions 是约 50–500 MB 未压缩数据的存储与 pruning 单元，不等于查询 task。warehouse size、multi-cluster、Query Acceleration Service 与 MAX_CONCURRENCY_LEVEL 分别涉及资源容量、查询并发和加速服务，也不是公开的 Stage partition-count 公式。扫描掉的 micro-partition 变少，只能说明工作减少，不能自动推出 DOP 已重算。Snowflake micro-partitions、Snowflake EXPLAIN、Warehouse overview、Query Acceleration Service。

这一层的核心差异不是“谁有自动化”，而是我们能否沿证据回答：输入是什么、修改哪个字段、哪些 guard 阻止应用。

本次执行中的修正：物化、延迟绑定与局部派工

第二次机会：Spark AQE 如何改变 Shuffle 读法

自适应查询执行（Adaptive Query Execution，AQE）利用已完成 Shuffle 的统计修改后续消费方式，因此可以复用已有数据，而不必每次重新 Hash。

自适应的窗口来自执行边界

运行时 actual 只有在仍存在可修改工作时才有价值。对 Spark 的常见路径，Shuffle 物化提供了一个自然窗口：上游结果已经存在，下游 consumer 尚未启动。

1Shuffle 写端完成
2  -> MapOutputStatistics：每个 reducer bucket 的真实 bytes
3  -> AQE 改写剩余计划及 Shuffle reader specs
4  -> 根据新 specs 创建下游逻辑 tasks
5
6已经完成的 map 输出继续复用，不因 coalesce 重新写一遍。

更广义的 AQE 可以观察运行中或失败的 Stage，取消并重新规划受影响工作，因此“绝不触碰运行中的 Stage”不是普遍定律。准确边界是：不回写已经完成的计算事实；若修改正在执行的工作，必须承担取消、重试、幂等（Idempotency）与状态迁移成本。本文后续要区分延迟绑定与 mid-flight recovery。

Coalesce：合并的是 reader mapping，不是重新 Hash

CoalesceShufflePartitions 收集可共同合并的 Shuffle groups，读取 bytes statistics，并调用 ShufflePartitionsUtil.coalescePartitions 生成新 specs。

1// 节选；每组有自己的 minNumPartitions、advisoryTargetSize 与 minPartitionSize。
2val newPartitionSpecs = ShufflePartitionsUtil.coalescePartitions(
3  coalesceGroup.shuffleStages.map(_.shuffleStage.mapStats),
4  coalesceGroup.shuffleStages.map(_.partitionSpecs),
5  advisoryTargetSize = advisoryTargetSize,
6  minNumPartitions = minNumPartitions,
7  minPartitionSize = minPartitionSize)

例如，原本 6 个 reducer bucket 可以映射成 3 个 reader task：

1原始 reducer id    0     1     2     3     4     5
2真实 bytes       12M   10M   35M    8M   40M    5M
3
4新 reader 0     [0, 1, 2]   -> 同一 task 读取这些 bucket
5新 reader 1     [3, 4]
6新 reader 2     [5]
7
8示意图只说明 mapping；具体分组由 target、minimum 与兼容性共同决定。

为什么 Join 的两侧不能分别随意合并？因为相同 key 的数据必须被对应的 reader 处理。源码将需要对齐的 Shuffle 放入同一 coalesce group，统一计算边界；Union 等可独立执行的分支则允许分别合并。输入对齐是语义约束，不是性能选项。

典型默认是 advisory 64 MiB、minimum 1 MiB、parallelismFirst=true。最后一个开关意味着：没有显式 minimum partition count 时，规则参考 SparkContext.defaultParallelism 尽量保留并行度，因此最终 task 大小可能低于 64 MiB。target 是 advisory，不是硬上限或精确的最终分区数。

源码：CoalesceShufflePartitions.scala、ShufflePartitionsUtil.scala。

Skew split：一个 reducer partition 能拆，但拆分粒度有限

热点 Join partition 的默认识别条件近似是同时超过 median 的 5 倍与 256 MiB。识别之后，createSkewPartitionSpecs 读取同一个 reducer id 来自各 map task 的 block 大小，按目标大小生成 map-index ranges。

1// 关键节选；startMapIndex/endMapIndex 是 map block 范围，不是 row offset。
2val mapPartitionSizes = getMapSizesForReduceId(shuffleId, reducerId)
3if (mapPartitionSizes.exists(_ < 0)) return None
4
5val mapStartIndices = splitSizeListByTargetSize(
6  mapPartitionSizes, targetSize, smallPartitionFactor)
7
8// 只有形成多个范围时才生成多个 PartialReducerPartitionSpec。
9// 每个 spec 包含同一 reducerId，以及一个 [startMapIndex, endMapIndex)。

因此，原稿中“一个完整的大 Shuffle partition 不能再拆，必须由一个 task 完整处理”需要订正：

Spark 能拆分一个 reducer partition 中来自多个 map task 的 block 集合；这条路径不能任意把单个 map→reduce block 再按行或字节切开。

 1同一个 reducer r
 2  map 0 -> r : block A
 3  map 1 -> r : block B
 4  map 2 -> r : block C
 5  map 3 -> r : block D
 6
 7拆分后
 8  consumer 0 -> r 的 [map 0, map 2) : A + B
 9  consumer 1 -> r 的 [map 2, map 4) : C + D
10
11若热点主要集中在不可再切的单个 block A，增加 ranges 不能消除 A 本身的长尾。

Join 还需要处理对应关系：只拆左侧时，右侧相应 partition 要重复读取；两侧都拆时，需要配对（Paired）组合这些范围。哪些侧可以拆受 Join 类型约束，否则可能破坏 outer/semi/anti join 的结果语义。收益也应扣除复制读取、更多 task 与可能的额外 Exchange 成本。

普通 coalesce 和 skew split 主要修改读取 specs，不等同于重新执行 Hash Partition。REBALANCE 的 skew 优化又是独立规则，不能把所有场景都套进 Join skew 的阈值。

源码：OptimizeSkewedJoin.scala、ShufflePartitionsUtil.scala。

AQE 不只是在调一个数字

真实 relation size 还可以触发 Join 策略切换；local Shuffle reader 改变 locality 与 mapping；动态 filter 可以减少尚未完成的扫描。它们可能间接改变后续 task 数，但控制对象分别是算法、物理属性（Physical Property）和工作规模。

2024 年 Adaptive and Robust Query Execution for Lakehouses at Scale 的重点也不仅是 coalesce：QueryStage 包装已提交的 fragment，LogicalLink 关联逻辑/物理计划；完成、失败或运行指标触发事件循环；重优化剩余逻辑计划后再次形成可调度 Stage。论文进一步讨论 broadcast fallback、Shuffle elimination fallback 与 skew remediation。

它说明 AQE 的价值既是更快，也是判断错误时仍能完成查询。但 Photon 的 mid-flight recovery 不是所有 OSS Spark 版本都具备的相同实现；论文整体 benchmark 收益也不能归因于 DOP 单项。coalesce、Join 改写、filter 与恢复机制必须分别阅读、分别评价。

Trino FTE：bucket 数与 task 数为什么可以解耦

容错执行（Fault-Tolerant Execution，FTE）通过持久化中间数据分离 bucket 与执行 task 的生命周期，任务形成不再等同于普通执行的节点分配。

FTE 是执行模式，不是普通调度器上的一个补丁

Trino 默认 retry-policy=NONE。TASK-retry Fault-Tolerant Execution 需要相应 Exchange/spooling 支持，让中间结果与单个执行 task 的生命周期解耦。task 可以多于 execution nodes，排队分批运行；下游如何把 splits 与 Exchange 数据组织成 task 也有专门路径。

这与普通 FIXED HASH 的 node mapping 不同。不能先拿 Standard execution 的“节点封顶 task 数”，再声称 FTE 只是在其上做一次 AQE。

这些增长参数属于 arbitrary task sizing，不能套到 hash compute 的全部路径。当前 hash compute target 与 hash writer target 有各自的配置，partition 分布与语义限制也仍然存在。

文档与默认值：Trino Fault-tolerant execution、Trino QueryManagerConfig.java。

HashDistributionSplitAssigner：把小 bucket 装进较少 task

createSourcePartitionToTaskPartition 使用每个输入源的 partition bytes estimate，形成 bucket 到 task partition 的映射。其算法骨架是：

 1// 根据当前源码整理的伪代码。
 2if (explicitPartitionToNodeMapping || missingRequiredSourceEstimates) {
 3    return oneTaskPartitionPerBucket();
 4}
 5
 6target = adjustTargetUsingTotalBytesAndTaskCountPreferences();
 7
 8for (int bucket = 0; bucket < partitionCount; bucket++) {
 9    long bytes = sumSourceEstimatesForBucket(bucket);
10    Assignment lightest = assignments.peek();
11
12    if (lightest == null || lightest.bytes + bytes > target || !canMerge) {
13        createTaskPartitionFor(bucket);
14    }
15    else {
16        mergeBucketInto(lightest.taskPartition, bucket);
17    }
18}

优先队列使新 bucket 尽量放入当前较轻的 task partition，但它不是不受约束的最优 bin packing。明确 connector mapping 或缺少统计时，直接退回 one bucket → one task partition；write fragment 的 merge 权限也与普通 compute 不同。

源码注释还指出：target max task count 用来调整 target bytes，实际 task 数仍取决于分布，可能超过这个偏好值。因此它不是严格 admission cap。

源码：HashDistributionSplitAssigner.java。

大 bucket 的拆分为什么只适用于特定路径

同一个 assigner 中，canSplit 由 fragment 的 scale-writers 属性决定。超大 bucket 只有在允许拆分，并且除最大输入源之外的其他输入之和不超过 target size 的四分之一时，才形成多个 subpartitions。

分发时，最大源的 splits 分散到 subpartitions，其他相关输入复制到每个 subpartition。因此，它不是把普通 Join/Aggregate 的所有 Hash bucket 任意切开。普通 hash compute 在这条机制中不能仅因 bytes 过大，就把一个 bucket 拆成多个 task。

这里与 Spark skew join 的共通问题是状态所有权：相同 key 的 build/probe 或 aggregation state 如何完整可见？复制、二次聚合、重新分区可以改变答案，但都有额外语义与资源成本。task formation 不能跳过这些约束。

AdaptivePartitioning：必要时改变未来 Exchange，而不是“开 1,000 个 task”

TASK-retry 下还有独立、默认未开启的 runtime adaptive partitioning。AdaptivePartitioning 检查尚未执行的 hash-input fragment，结合 runtime/estimated output sizes 估计内存压力；压力超过“max task size × 当前 partition count”时，触发计划改写。

1上游输出统计可用
2  -> 估计未来 hash-input fragment 的内存需求
3  -> 若当前 partition 数不足且 guard 通过
4       未执行 remote fixed-hash Exchange：提高 partition count
5       已物化 RemoteSource：可能在其后插入新的 repartition Exchange
6  -> 重新 fragment
7  -> FTE task assigner 仍可把多个小 bucket 合入一个 task

当前默认目标 1,000 指的是 Exchange partition count，不是 guaranteed task count，也不是同时运行槽位。源码的内存近似是：单个 partitioned input 取其总 bytes；多个输入取总和减去最小输入，假设最小输入可以流式经过。broadcast inputs 则按规则的假设排除在这项估计之外。这些近似不是严格的峰值内存上界。

遇到 TableExecute、不合适的 fragment 状态，或整个计划已经包含达到目标的 fragment 时，会保留原计划。它不是逐 Stage、连续可微地寻找一个全局最优整数。

与 Spark reader-spec merge/split 相比，这里可能插入真正的新 repartition，意味着额外数据移动。“改 mapping”与“重写数据分布”必须分别计算成本。

源码：AdaptivePartitioning.java。

Writer scaling：在当前 Stage 增加专项执行能力

Writer scaling 是运行中可以增加 task 的另一种情形。普通 ScaledWriterScheduler 先启动一个 writer task，随后检查上游 output buffers 是否拥塞、已处理数据是否足够、现有 writer 是否初始化以及节点上界；满足条件才增加 writer node/task。

task-local writer scaling 则增加单个 task 内的 writer operators，使用自己的内存与吞吐 guard。不能把 task-local 的 memory ratio 条件写成所有分布式 writer scheduler 的同一套算法。

这是一种受限、通常只增不减的控制器：它解决写入瓶颈，不证明整个查询的所有 Stage DOP 都被重新计算。

源码与文档：ScaledWriterScheduler.java、Writer scaling properties。

延迟绑定：Flink、Tez 与 SCOPE 的不同选择

延迟决定下游规模可以使用更可靠的统计，但等待成本与可修改范围取决于已有物化边界。三个系统分别控制 vertex、原始分区合并和剩余计划。

Flink Adaptive Batch：先知道消费量，再实例化下游

Flink Adaptive Batch Scheduler 的关键不是让一个已运行 vertex 随时扩缩，而是延迟初始化未决定 parallelism 的 JobVertex。当输入结果满足相应 ready 条件并且统计可用，scheduler 调用 decider，同时决定 parallelism 与每个 subtask 的 input ranges，然后初始化执行图节点。

1AdaptiveBatchScheduler
2  -> 检查输入结果与 source inference 是否 ready
3  -> tryDecideParallelismAndInputInfos
4  -> DefaultVertexParallelismAndInputInfosDecider
5  -> ParallelismAndInputInfos
6  -> initializeJobVertex
7  -> 更新执行拓扑、调度新 subtasks

对没有显式初值的非 source vertex，基础估计可近似表示为：

真实实现还要区别 pointwise/all-to-all、兼容多个输入、调整 input ranges，并遵守 vertex max parallelism。不能只算一个整数，再随意把所有上游 subpartition 平均分给它。broadcast input 不按总复制量进入这条简单估计；source 可以通过 DynamicParallelismInference 提供输入相关初值。

在 release-2.3 文档及所核验配置中，auto-parallelism 的典型默认 target 是 16 MiB，min 1、max 128；具体上界还受显式配置、环境默认并行度与 vertex 限制影响。用户显式 parallelism、scheduler 选择、batch execution 与 shuffle 模式决定是否使用该机制；blocking/hybrid readiness 也不是所有算子相同的“一律等待全部任务结束”。当前实现不应继续套用旧版“所有结果必须为 2 的幂”之类概括。

Flink 当前的 adaptive batch 也已超出均匀装箱：AdaptiveSkewedJoinOptimizationStrategy 收集两侧 aggregated subpartition bytes，在相应上游完成后按 median、factor 与绝对 threshold 判断热点，再结合 Join 类型决定可拆的一侧并尝试修改输入/输出 edges。AUTO 与 FORCED 对额外 Shuffle 的态度不同；广播 Join、MultiInput 与恢复功能也存在兼容性限制。因此，“Flink 只是把 consumed bytes 除以 target”同样不完整。

源码与文档：AdaptiveBatchScheduler.java、DefaultVertexParallelismAndInputInfosDecider.java、AdaptiveSkewedJoinOptimizationStrategy.java、Flink Adaptive Batch release-2.3。

Hive/Tez：先预留 partition 上界，采样后向下合并

Hive 的初始 reducer 估计通常以 input bytes 除以 bytes/reducer，再受 max reducers 约束；Tez auto reducer 模式还会用 partition factors 预留范围。公开配置的典型值包括 hive.exec.reducers.bytes.per.reducer=256000000、hive.exec.reducers.max=1009，hive.tez.auto.reducer.parallelism 默认关闭。

ShuffleVertexManager 根据已完成 source task 的事件估计总输出，再决定是否降低 destination parallelism。关键边界是：它通过合并原始 partition 的连续范围减少 consumer tasks，不能在这条 auto-reduce 路径中突破最初预留的 partition 数向上扩张。分组粒度、minimum、采样量与调度时点也使最终值不一定等于简单的 desired reducers。

1编译：estimate -> 预留较细的原始 reducer partitions
2运行：source events -> 估计总输出 -> 合并原始 partition ranges
3启动：按新范围调度 destination tasks
4
5初值给得过小，单靠这条“只减”的机制无法补救。

这与 Spark coalesce 有相似之处，但并不意味着 Tez 的任意 DAG 都不能增 vertex parallelism；这里讨论的是 ShuffleVertexManager 的特定 auto-reduce 算法。

资料：Hive configuration properties、How ShuffleVertexManager Auto Reduce Parallelism works。

SCOPE：将 actual 与可复用结果一起交回优化器

2013 年 Continuous Cloud-Scale Query Optimization and Processing 给出了更完整的 remaining-DAG reoptimization：vertex 完成后，Job Manager 汇总 statistics package 与 materialization package，异步调用 optimizer，并在收益与可行性满足时替换剩余计划。

statistics 不只有 cardinality 与平均行宽，还包括 UDF inclusive cost、partition histogram；materialization 则说明哪些中间结果真实存在，以及它们的 partitioning/ordering。逻辑 signature 用于关联等价表达式与统计；物理属性决定已完成结果可以怎样复用。

因此调整对象可以超出 task count：partition 数、partition key、Join variant，以及小规模剩余工作是否串行化。这里是论文公开架构，不代表 2026 年某个现行服务的全部能力。它的重要启示是：只交回 bytes 太少，只交回 estimate 又太抽象；优化器既需要知道真实工作规模，也需要知道不可免费撤销的执行事实。

Task 内的另一层 DOP：StarRocks 与 morsel-driven 执行

本地 driver 数与分布式 task 数改变的是不同状态，不能直接比较参数大小。StarRocks 在 pipeline 边界调整实例化规模，morsel 调度则改变细粒度工作的派发。

StarRocks 初值：pipeline DOP 与 fragment placement 分开决定

StarRocks 的 SessionVariable.getDegreeOfParallelism 在 pipeline engine 开启时，优先返回显式 pipeline_dop；否则从 BackendResourceStat 获取 default DOP，并受 max_pipeline_dop 约束。当前配置初值是 pipeline_dop=0、max_pipeline_dop=64。

 1// 核心分支节选；warehouseId 对应资源统计的查询范围。
 2if (enablePipelineEngine) {
 3    if (pipelineDop > 0) {
 4        return pipelineDop;
 5    }
 6    if (maxPipelineDop <= 0) {
 7        return BackendResourceStat.getInstance().getDefaultDOP(warehouseId);
 8    }
 9    return Math.min(maxPipelineDop,
10            BackendResourceStat.getInstance().getDefaultDOP(warehouseId));
11}
12return parallelExecInstanceNum;

该方法前的旧注释提到乘以 instance 数，但当前方法体并没有这样做——阅读时必须以实际调用链为准。

PlanFragment 与 assignment strategy 还会处理 pipeline 开关、scan ranges、bucket/local-shuffle 和 instance placement。一个 fragment 在哪些 BE 上运行，与每个 instance 创建多少 driver，是两个问题；少量 scan ranges 也可能限制可用 pipeline DOP。不能把 session 初值直接当成每个物理 pipeline 的最终值。

源码：SessionVariable.java、PlanFragment.java。

自适应状态机：小结果缩并发，大结果保流水

BE 的 CollectStatsContext 在上、下游 pipeline 之间收集 chunks，用有限缓冲换取小结果规模信息，达到阈值则恢复流水执行。其核心状态机是：

1BLOCK：暂存上游 chunks，累计 rows；下游尚未输出
2  ├─ 累计 rows 达到 buffer threshold
3  │    -> PASSTHROUGH：保留 upstream DOP，释放下游继续流水执行
4  └─ 所有 upstream drivers 已完成且结果较小
5       -> ROUND_ROBIN：计算较低 DOP，将旧 driver buffers 合给新 drivers

这里不是为了统计而把任何大结果完整物化。达到 threshold 就转 passthrough，避免为了准确知道总量而无限延迟下游。若上游全部结束，BlockState.set_finishing 才走收缩路径：

 1// 核心节选；只有全部 upstream driver sequences 完成才执行。
 2size_t num_partial_rows = _ctx->_max_block_rows_per_driver_seq;
 3size_t adjusted_dop = _num_rows / num_partial_rows;
 4
 5// 注意是整数除法并向下取 2 的幂，不是 ceil(rows / target)。
 6adjusted_dop = compute_max_le_power2(adjusted_dop);
 7adjusted_dop = std::max<size_t>(1, adjusted_dop);
 8adjusted_dop = std::min<size_t>(adjusted_dop, _ctx->_upstream_dop);
 9
10_ctx->_transform_state(CollectStatsStateEnum::ROUND_ROBIN, adjusted_dop);

RoundRobinState 让新 driver 消费 driver_sequence + k * downstream_dop 对应的原 buffers，结合 ChunkAccumulator 输出。缩 DOP 必须同时改变输入所有权，否则只少创建几个 driver 就会丢数据。

之后 CollectStatsSourceOperatorFactory.adjust_dop 还考虑 dependent pipelines 的 DOP 与输出放大系数，可能把收缩后的值抬高，但最终不超过 upstream DOP。因此不能将整个实现简化为“根据总 rows 任意扩缩”。它更接近一个有预设上界、考虑依赖关系的小结果收缩机制。

源码：collect_stats_context.cpp、collect_stats_source_operator.cpp。

延迟的是 driver 实例化，不是任意缩减运行线程

该机制改变尚未创建的 driver 数，而不是随时销毁正在运行的线程。CollectStatsSourceInitializeEvent.process 先调整 DOP，再实例化 driver：

1// 节选；pipelines 按拓扑序排列。
2for (auto* pipeline : _pipelines) {
3    pipeline->source_operator_factory()->adjust_dop();
4    pipeline->instantiate_drivers(state);
5}
6// 后续 prepare drivers，再交给执行器调度。

先 adjust、再 instantiate，说明这条路径是在下游 drivers 尚未创建时决定其规模。它与 Flink delayed vertex initialization 有架构上的相似性，但修改单位不同：一个是本地 pipeline drivers，一个是 JobVertex subtasks。

此外，pipeline DOP 不等于线程池大小。driver 的 ready/blocked 状态与共享执行器共同决定谁正在消费 CPU。观察 driver 数、线程利用率与 I/O 阻塞，需要分别解释。

源码：adaptive/event.cpp。

Morsel-driven：把工作单元做细，让资源分配保持弹性

2014 年 Morsel-Driven Parallelism: A NUMA-Aware Query Evaluation Framework for the Many-Core Age 讨论了另一种解耦：输入划成小 morsels，由 dispatcher 在运行时分配给固定、机器相关的 worker pool；worker 执行一个 pipeline 到下一个 breaker，调度优先考虑 NUMA locality，并在 morsel 边界调节不同查询获得的执行机会。论文全文。

这意味着输入切分数量可以远大于正在运行的 worker 数。弹性来自动态派工，而不是每次资源变化都重写 Hash partition scheme。算子需要理解共享状态和同步，也不能据此断言分布式 Shuffle bucket 可以无成本切开。

StarRocks 的 morsel queue、pipeline scheduling 与自适应 DOP 可以放在这个坐标下理解，但不是对 HyPer 论文实现逐行复刻的声明。共同问题是：如何在负载均衡、局部性、共享状态与控制开销之间保留选择空间。

第三次机会：历史究竟教给优化器什么

历史反馈（History-Based Feedback）减少的是重复执行中的估计与纠偏成本，前提是计划身份、数据与资源仍具有可比性。记录一次成功值并不能证明它是最优 DOP。

历史反馈不等于保存上次的整数

判断一次历史执行能否用于当前决策，需要核对五项条件：

上次最终 task 数可能受配额、失败、connector bucket 或用户设置限制；上次运行更快，也可能只是缓存更热、竞争更少。历史结果不能自动成为最优标签。

BigQuery：公开的 history-to-initial-parallelism

BigQuery HBO 的 parallelism_adjustment 明确将历史 workload distribution 用于改善后续 Stage initial parallelism，公开说明的历史优化作用域是对应 Google Cloud project。它只在有较高置信度时使用反馈，并继续验证收益；效果不足或回退时可以撤销优化。

1本次执行的 runtime distribution
2  -> 相同或相似查询的可用历史
3  -> 下次执行选择更好的 initial parallelism
4  -> 本次运行仍保留动态调整
5  -> 验证收益，必要时撤销

这是公开产品中直接关联历史与 Stage 初始并行度的清晰例子。但内部 key、TTL、distribution 表达、drift threshold，以及 actuator 到 parallelInputs/bucket/requested slots 的具体映射没有公开。外部 normalized_literals 等观测字段不能被直接当成内部 history key。

资料：BigQuery history-based optimizations、New BigQuery history-based optimizations speed query performance。

Presto HBO：统计是通用的，直接 task-count 应用是 Writer

Presto HBO 对 canonicalized Stats-Equivalent subplan 计算 SHA-256，以输入 rows/bytes pattern 选择相近历史。普通应用主要改善 cardinality、Join order/distribution、aggregation 与 null-skew，不是通用 Stage DOP advisor。

直接 task-count 应用是可选的 history-based Scaled Writer：历史存在有效 final writer task count，且开关开启后，ScaledWriterRule 使用下面的种子值。

1initialWriterTasks = ceil(historyFinalWriterTasks / 2)

为什么只用一半？writer controller 在当前执行中仍可增加 task，而不是把历史值固定为最终值。该策略减少从 1 开始逐步爬升的成本，也保留对输入变化的在线适配空间；它不是证明上次的一半必然最优。

历史 tracking/use 与 history-based writer 功能都有对应开关，不应把默认关闭的行为写成所有查询天然生效。有限 history patterns、输入距离和存储过期共同限制适用性。当前文档也允许收集失败查询中已完成 fragment 的统计，因此“Presto 只学习成功查询”并不准确；更一般的风险是，未完成和未探索的方案仍然缺少可比较样本。

2024 年 Presto’s History-based Query Optimizer 解释了统计与应用的组织方式。论文 Scaled Writer 部分报告的是该专项的延迟收益，不能拿整个 HBO 的 CPU/latency 收益来证明通用 DOP 已被优化。

源码与配置：ScaledWriterRule.java、Presto HBO documentation。

Ultron 2026：将 runtime partition scheme 变成下一次初值

2026 年 Ultron: History-Based Query Optimization at Databricks（PVLDB 19(12): 4358–4371，DOI 10.14778/3827998.3828038）提供了比产品概述更具体的 partition-count application。

Softstore 保存跨集群历史，QuickPredict 提供本地检索，applications 消费 subplan properties。partition application 的核心是：若上次因 partition 过大执行了 adaptive repartition，记录最终 scheme；重复执行时直接使用该 scheme，减少再组织数据的成本；输入规模变化则由 size factors 使自适应重新介入。

1第一次：initial scheme -> runtime 发现过大 -> adaptive repartition -> final scheme
2第二次：命中适用历史 -> 从上次 final scheme 开始 -> 保留 runtime fallback

论文 §5.3 在 30 TB TPC-H、56 个实例、partition-sensitive 子集上评估；归一化基线是整体 workload 最优的固定 scheme，不是逐查询 oracle。§5 明确各 application 的部署、benchmark、测试阶段不同：生产 fleet 的 Join 数据不能证明 partition application 已作为通用产品能力发布。

因此它是已核验的研究与实验依据，不应与 Databricks 产品页面的 HISTORY_BASED_JOIN_STRATEGY 等同。结论是：学习一次纠偏后的 scheme，比学习一个脱离执行语境的“最佳 DOP”标签更稳健。

Snowflake 与 Databricks：有 history，不一定有 Stage DOP actuator

Snowflake Optima Planning 使用执行历史改善后续计划，例如 Join order。2026 年 8 月官方说明其匹配从 exact query text 扩展到 plan shape；这扩大了反馈复用范围，却没有公开 Stage partition count、target bytes 或 task-count 控制器。当前 Planning 页面限定适用于 Gen2 standard warehouses 与 Adaptive Warehouses；Optima 的不同功能不能共享同一组适用性假设。Snowflake Optima、2026 performance improvements。

Databricks Query performance insights 公开的 Beta HISTORY_BASED_JOIN_STRATEGY 依据相似查询历史选择 Join 策略，例子是 broadcast 替代 Shuffle。它可以间接消除 Exchange，但页面没有公开历史 Stage DOP actuator。Ultron 论文能补充研究机制，不代替该产品的可用性说明。Databricks Query performance insights。

“没有公开”不是“内部绝不存在”；它只是这份调研能够负责地作出结论的边界。

SQL Server 与 Oracle：直接调 DOP，但控制单位不同

SQL Server DOP feedback 识别重复查询的过度并行，主要依据 adjusted elapsed time 与 waits 试探更低 DOP，验证后写入 Query Store，回退时恢复已知较好值。其目标也包含整体并发改善，而不仅是某条查询的绝对最短时间。它不把并行计划改成串行计划，也不是分布式 Stage task-count 机制。SQL Server DOP feedback。

还需避免把 SQL Server MAXDOP 当成整个 request 的精确线程总数：其限制与并行执行 tasks 的关系比这个整数更复杂。SQL Server query processing architecture。

Oracle Auto DOP 将 statement 的成本、预期时间与并行执行限制纳入决策，parallel statement queuing 又控制何时获得执行资源。PARALLEL_DEGREE_POLICY=ADAPTIVE 在 Auto DOP 基础上启用 performance feedback；它与已被标记 deprecated 的 PARALLEL_ADAPTIVE_MULTI_USER 不是同一个机制。PX server sets、statement DOP 与资源准入需要区分，不能拿来直接填入 Spark Stage task 矩阵。Oracle Using Parallel Execution。

从论文看演进：估计、纠偏与学习并不是替代关系

编译估计、运行时修正和历史复用分别服务不同信息条件，不能用后一种机制替代首次执行的起点或在线回退。

六篇论文各解决一个缺口，合并起来是完整链条

下表按时间列出六篇论文处理的缺口及其证据边界；细粒度调度与跨执行学习分属不同控制层，不能按发表顺序解释为替代关系。

这些论文不构成一条“最新算法淘汰旧算法”的路线。它们各自填补了 DOP 决策链上的一个缺口，缺任何一环都会把成本转移到另一环——没有编译期估计，第一次执行全盲；没有运行时纠偏，一次错误酿成全程错误；没有历史复用，每次执行重付纠偏成本；没有细粒度调度，资源变化就被迫触发 repartition。

ML DOP：预测曲线比分类一个整数更有启发

2020 年 A Comparative Exploration of ML Techniques for Tuning Query Degree of Parallelism 将 SQL Server 的 plan features 与候选 DOP 输入模型，预测 execution time，再选择 workload-level 或 per-query DOP。特征区分 operator、row/batch mode 与 serial/parallel mode；实验讨论查询模板、数据规模等变化下的迁移问题。论文全文。

该论文的问题表述更有价值：学习 T(plan, DOP)，就能比较候选值的 latency/resource trade-off，而不只是分类一个“最佳整数”。但它仍依赖候选 DOP 的运行样本；共享集群竞争与分布式 state placement 不会因为换成 ML 自动消失。

历史反馈最难的部分，是没有观测到的反事实

同一个查询上次使用 DOP 64，并不意味着 32 或 128 更差。生产运行通常只观察实际选择的路径；被过滤的方案、超时方案、不同资源环境下的结果，构成不完整且带偏的数据集。

因此应当区分三种标签：

• 事实标签：这个 subplan 实际输出了多少 rows/bytes。
• 策略标签：runtime 最后形成了什么 partition scheme 或 writer 数。
• 收益标签：与可比较 baseline 相比，该动作改善了多少 latency、资源消耗或尾延迟（Tail Latency）。

Ultron scheme 复用、Presto writer seed 和 SQL Server verified feedback 的共同逻辑是缩小需要相信的历史结论，而不是要求一次运行证明全局最优。

比较与验证：修改范围比参数大小更重要

横向归纳：每个机制真正修改什么

跨系统比较应以修改对象为单位：reader mapping、Exchange 和资源分配的成本不同，即使最终 task 数相同。

机制矩阵，而不是产品功能榜

相似的自适应标签可能对应不同执行边界，下面将每种机制的信号、控制对象和限制放在同一口径下。

Snowflake Optima 与 Databricks 的公开 Join-history 功能没有在表中强行填一个 task-count actuator：现有证据指向计划选择，Stage DOP 的映射仍未公开。

BigQuery runtime 还需要把三件事分开：repartition/coalesce 改后续工作组织；speculation 增加同一逻辑 work unit 的 attempts；slot reallocation 改执行资源。动态 Stage 图可确认这种执行架构，但内部 DOP 数学公式不可见。BigQuery in-memory query execution、BigQuery query optimization guide。

三种“增加并行度”，成本完全不同

增加可运行 worker 不必重新分区，而拆分已有状态可能需要复制或重做输入。三种路径的增量成本如下。

1已有细粒度工作，增加可运行 worker
2  -> 主要是资源准入与动态派工
3
4已有多个可独立 block/split，增加 consumer mappings
5  -> 更多调度与读取，可能复制另一侧输入
6
7原 partition/state 不可独立分解，需要新 Exchange
8  -> 重新移动数据，甚至取消与重做部分工作

同样叫“从 32 调到 64”，可能是增加资源，也可能是改 reader specs，还可能是新建 Shuffle。单看数字的前后变化，无法评价算法优劣。

我的思考：DOP 的本质是为不确定性保留余地

评价 DOP 初值必须包含选错后的修正成本；同样的估计误差，在可廉价合并与不可拆分的执行架构中会造成不同损失。

初值不是越精确越好，错误之后的代价同样重要

预先得到精确 cardinality、skew 与资源可用性当然理想，但统计与控制本身也有成本。一个允许便宜 coalesce 的细初值，可能比昂贵的准确规划更实用；反过来，如果大 partition 后续很难拆，低估就会产生无法轻易修复的长尾与内存风险。

因此评价初始 DOP 时，应同时看 estimate 的准确度与执行架构的可纠偏性。Tez 的预留上界、Spark 的 reader-spec 调整、Trino 的 FTE spooling 都是在改变“选错以后还能做什么”。

平均大小不足以解释倾斜，状态语义决定拆分能力

总 bytes 相同的两个 Stage，一个负载均匀，一个只有一个 heavy hitter，它们的最优组织可能不同。简单提高 Hash partition 数，也不能把同一 key 自动散到多个独立状态里。

真正需要问的是：能否划分输入工作，能否共享或复制 build state，能否在最后做 second-stage aggregate，能否维持 outer join 与写入语义？源码中的 Join-type guards、scaled-writer split predicate 与输入对齐要求，都是这些问题的具体化。

DOP 调研如果没有讨论 state ownership，就还停留在调度参数层。

资源弹性与逻辑分区应解耦，但不是无限解耦

细工作单元配合动态派工，可以在不重分区的情况下吸收资源变化；过细则会增加 metadata、buffer 与调度成本。更粗的 task formation 可以降低开销，却可能收紧调节空间。

FTE bucket-to-task mapping 与 morsel-to-worker dispatch 分别展示了分布式与本地的解耦方式。它们也提示一个研究问题：怎样选择足够细、又不制造控制面（Control Plane）瓶颈的基础工作粒度？这不是单个 bytes/task 阈值能永久回答的问题。

History 与 runtime 应互补，而不是让历史取消纠偏

历史最有价值的用途，是避免重复犯相同错误与重复支付恢复成本。但数据规模变化只是 drift 的一种形式：分布、行宽、UDF、硬件、并发和实现版本都可能变化。

复用越强，失效判断就越重要。即使 total bytes 没变，key 分布变了也可能使旧 scheme 不再适合；即使 subplan output 相同，不同资源与局部性也可能改变 latency curve。公开资料尚不足以比较所有系统对此的处理，因此这仍应作为问题，而不是替它们补写一套假想实现。

当前工程实践留出的五个缺口

DOP 机制必须同时处理目标、信号、时机、归因和纠偏成本；本文的公开证据不足以断言某个系统已完整解决它们，也不足以断言没有系统做到。可以确定的取舍包括：

• 目标函数：latency、CPU/slot-seconds、内存峰值与共享集群尾延迟之间存在冲突。SQL Server DOP feedback 把并发吞吐纳入目标后，单查询最短延迟便不再是唯一目标——单查询最短时间的 DOP 在共享集群中可能挤占其他查询的槽位。
• 信号质量：bytes、row counts、CPU cost、partition histogram 各有采集成本。StarRocks 的 collect stats 为避免无限等待设了 buffer threshold，超过即转 passthrough——缓冲阈值决定了能获得多少统计，以及下游需要等待多久。
• 时机选择：Flink Adaptive Batch 在输入满足就绪条件、统计可用后决定下游 parallelism，利用的是相应批执行边界。不能把这统一描述为额外插入的 barrier；比较时需确认原计划是否已使用 blocking exchange，以及等待条件带来多少增量延迟。
• 归因与迁移：上次运行更快，可能来自更好的 scheme，也可能来自缓存更热、竞争更少。Ultron 的 size factors 在输入规模变化时让自适应重新介入，但分布、行宽、UDF 版本和硬件变化仍然没有通用检测机制。
• 纠偏成本：mapping 修改（Spark coalesce）、复制读取（skew join 的右侧重复）、repartition（Trino AdaptivePartitioning）和 cancellation 的成本不在同一量级。本文资料没有提供可跨系统套用的统一 cost model。评估仍需分别测量这些操作的字节数、延迟和回收成本，而非只观察 DOP 的前后变化。

结语

调整 DOP 前应先确认控制对象是 reader mapping、Exchange、driver 还是资源配额，并比较调整成本与继续执行的损失。细分区只有能被低成本合并时才提供弹性；大分区只有能保留语义地拆分时才有在线纠偏空间。历史结果也必须重新检查数据、资源和版本条件。

本文的三个时点是信息分类，不是控制器只能调整三次的限制。评估时记录每次调整的信号、状态边界、等待与恢复开销；仅有 task 数下降或单次延迟改善，不足以证明资源效率和共享集群吞吐同时提高。

参考资料与继续阅读

源码机制、论文实验与产品可用性分别提供不同证据，不能互相替代。引用时应保留版本、执行模式和功能作用域。

论文

• Continuous Cloud-Scale Query Optimization and Processing — Bruno, Jain, Zhou，PVLDB 2013
• Morsel-Driven Parallelism — Leis, Boncz, Kemper, Neumann，SIGMOD 2014
• A Comparative Exploration of ML Techniques for Tuning Query Degree of Parallelism — Fan 等，2020
• Adaptive and Robust Query Execution for Lakehouses at Scale — Xue 等，PVLDB 2024
• Presto’s History-based Query Optimizer — Shankhdhar 等，PVLDB 2024
• Ultron: History-Based Query Optimization at Databricks — Nakandala 等，PVLDB 2026，论文全文

官方机制说明

• Spark SQL Performance Tuning
• Trino Fault-tolerant execution
• Flink Adaptive Batch release-2.3
• Presto History-based optimization
• Tez ShuffleVertexManager Auto Reduce Parallelism
• Databricks Adaptive Query Execution
• BigQuery history-based optimizations
• Snowflake Optima
• SQL Server DOP feedback
• Oracle Using Parallel Execution

站内相关调研

• Hash 还是 Sort：从优化器决策到运行时自适应
• 系统、AI Infra 与硬件：2026 年的分布式系统思考