kryonix 提交的DuckDB并行化有序缓冲 CTE 扫描 - #24361PR

并行化有序缓冲 CTE 扫描 - #24361

管道交换（pipeline exchange）可以通过批次索引在并行生产者之间保持顺序，但一个有序缓冲 CTE 消费者（consumer）之前被迫只能使用单个源任务。缓冲扫描暴露了一个平坦的数据块游标（flat chunk cursor），因此让多个任务从中拉取数据会在它们到达下游接收器之前重新排序这些数据块。当一个有序 CTE 馈送给一个批次感知的接收器（如CREATE TABLE AS、COPY或缓冲结果收集器）时，这个问题尤为明显：生产者可以并行运行，但交换到接收器的管道部分变成了串行执行。

我修改了缓冲交换，使其可以将完整的生产者批次分配给独立的扫描任务。一个任务仍然会按顺序读取其批次内的数据块，而不同的任务可以并发地处理不同的批次。下游接收器接收原始的批次索引，并使用其现有的批次排序机制来按顺序发布结果。

这种选择是基于物理管道需求（而非在 CTE 被绑定或优化时附加到它）进行的。这很重要，因为 CTE 可能在优化过程中被引入，并且有用的执行策略取决于最终的消费者管道。在PhysicalCTE::BuildPipelines期间，一个缓冲消费者使用批次扫描，仅当满足以下所有条件时：

1. 交换拥有批次索引；
2. 下游接收器需要批次索引；并且
3. 接收器不需要分区列。

所有其他缓冲消费者都保持现有的数据块扫描。这包括无序交换、没有批次索引的顺序交换、不需要批次索引的结果路径，以及分区列接收器。直接和完全物化的 CTE 消费者保持不变。

这种区别在EXPLAIN ANALYZE中对缓冲 CTE 扫描是可见的：

• CTE Mode: BUFFERED, Scan Mode: BATCH意味着源任务声明完整的交换批次范围。扫描可以使用交换的常规并行性，因为接收器可以从批次索引重建所需的顺序。
• CTE Mode: BUFFERED, Scan Mode: CHUNK意味着消费者读取现有的共享数据块流。如果交换保持顺序，此路径仍然是单线程的。

我有意保留了现有的内存中/假脱机（spooled）数据块流作为权威存储。批次扫描在该流之上添加了一个轻量级目录；它不会将数据块复制到第二个集合中。每个目录条目记录生产者和交换批次索引，以及在平坦数据块流中的起始/结束位置。当一个批次到达、生产者推进其最小批次索引，或生产完成时，一个范围（range）被关闭。

每个本地扫描一次可以拥有一个范围。消费者全局状态记录下一个未声明的范围、每个已声明范围的下一个未读位置，以及当前在交换锁之外的任何假脱机读取。范围的退役受这三个条件限制，因此只要任务还能读取，无论是内存中的数据块还是假脱机数据块都不会被退役。空闲任务还会在所有当前可见范围被声明后发布一个批次下限（batch floor）；否则，一个空闲任务可能会无限期地让一个批次感知接收器的全局最小值固定在一个旧的批次上。

这里有一个不太明显的执行过渡。动态源可能在一个批次的数据可用之前就完成了该批次。返回一个空数据块是不够的，因为接收器仍然需要关闭旧的本地批次，然后获知新的全局最小值。我为此添加了只用于控制流的源结果类型SourceResultType::BATCH_ADVANCED。

管道执行器分两个有序步骤处理该结果。它首先运行现有的NextBatch路径，此时旧的最小值仍然可见。然后它更新本地最小值并在接收器上调用UpdateMinBatchIndex。这个顺序对于批次插入和批次复制很重要：旧批次必须在其最后一个拥有者从最小值中消失之前被最终化。这两个操作都可以独立地阻塞和恢复。最小值的更新也会通过直接和嵌套的 CTE 消费者传播，因此当交换被链接时，批次边界不会丢失。

PhysicalOperator::UpdateMinBatchIndex默认是一个空操作。缓冲批次收集器、批次插入、批次复制和PhysicalCTE会覆盖它，因为它们维护依赖于全局最小值的状态。这使得通用的执行器转换变得明确，而无需不相关的接收器去实现它。

此变更最广泛的开销是管道执行器中新的源控制结果和接收器回调。在交换内部，批次消费者添加了范围和声明元数据以及更多的退役簿记（bookkeeping）。该元数据仅在一个批次扫描消费者被注册后才维护，因此现有的CHUNK路径不会构建一个并行批次目录。假脱机 I/O 和回调保持在交换互斥锁之外，并且取消、早期源终止、重置和消费者注销会在唤醒被阻塞的读写者之前清除未完成的声明。

我为最容易弄错此协议的情况添加了测试覆盖：重复的并行有序扫描、产生的批次少于扫描任务、早期LIMIT、混合的直接/缓冲消费者共享一个假脱机、稀疏和空批次、嵌套的缓冲交换、单个执行器、挂起结果执行和放弃，以及通过直接嵌套 CTE 传播到有序 Parquet 输出。这些测试验证了查询结果和所选的扫描模式。

为了进行本地性能检查，我在 Apple Silicon/macOS 上将确切的主分支提交bf753f906d与此提交9b22d512c9进行了比较。两者都是 Release 构建，具有相同的链接扩展，两个源码树都没有跟踪修改，查询使用了八个线程。输入由两个包含 1000 万行的表组成：

EXPLAINANALYZECREATEORREPLACETABLEoutASWITHinputASMATERIALIZED(SELECT*FROMlUNIONALLSELECT*FROMr)SELECTi,hash(i)AShFROMinput;

经过两次预热和九次交替采样，中位数总时间从主分支的 0.717 秒变为批次扫描的 0.279 秒，加速比为 2.57 倍。主分支的样本范围从 0.651 秒到 0.766 秒；此分支的样本范围从 0.265 秒到 0.361 秒。两个输出都恰好包含 2000 万行，其中min(i) = 0，max(i) = 19999999，以及 2000 万个不同的哈希值。

作为保护性测量，我还将一个 1 亿行的物化 CTE 运行到一个非批次感知的结果路径中。其计划保持为Scan Mode: CHUNK。经过三次预热和 31 次交替采样，主分支和此分支的中位数时间均为 0.360 秒。它们的范围重叠（0.241 秒-0.455 秒 和 0.241 秒-0.460 秒），修整均值（trimmed mean）也重叠（0.363 秒 和 0.345 秒）。我不会从这些数字中解读出性能变化，但它们表明，在此工作负载中，后备路径（CHUNK 模式）没有继承批次目录的元数据或调度开销。