D14 讀完 morsel-driven 論文，知道一句 SQL 丟進多核機器時，並行執行的骨架該長什麼樣：morsel 派工、work-stealing、pipeline breaker。D15 把 Comet 六個接點提出來，Spark 跟 DataFusion 之間如何計畫、資料、記憶體與怎麼交接，而 Rust 這邊是 async、JVM 那邊是 blocking，兩邊怎麼接起來是第一個工程難題。
這兩天介紹的是外面怎麼接與多核怎麼分，但 Comet 底下的引擎 DataFusion 自己內部是怎麼讓一棵 physical plan 動起來的？EXPLAIN 畫出來的樹不會自己跑呢？
今天來講講這層：資料在運算之間是被推(push)還是被拉(pull)？
Pull之後，又為什麼交給 Tokio 這個本來給 I/O 用的 async runtime 排班？
假如現在有一條 SQL 要怎麼被解析成 LogicalPlan、再被 optimizer 改寫成 PhysicalPlan
跑一次 EXPLAIN，就會看到一棵漂亮的樹：
AggregateExec
└── FilterExec
└── DataSourceExec
問題來了：這棵樹畫在那裡，但它並不會自己動
真正執行的瞬間，一定有某個節點需要先動做，究竟是最底下的 DataSourceExec 先開始讀檔案、讀到資料就往上送？還是最上面的 AggregateExec 先開口要資料、還是這句話一路往下傳到 leaf？
聽起來像在鑽牛角尖，但這個誰先開口決定了三件很實際的事：
所以今天第一題是：
資料在運算子之間到底是被「推」還是被「拉」？
CPU 密集的查詢工作，為什麼會跑在一個本來給 I/O 用的 async runtime 上？
在丟術語之前，先借一個生活場景
拉（pull）像板前壽司
客人開口說再來一貫，師傅才動手，師傅缺料才回頭跟後場要，如果沒人點整條線就自動停住
推（push）像工廠自動化流水線
開關一按，原料自己一路衝到包裝，中間不問任何人
這兩種店的差別，就是接下來要講的兩派執行模型，而 DataFusion 是板前，但客人一次點一整盤 8192 貫。
Pull-based（Volcano，Graefe 1994）：資料從下游往上拉
Root 節點呼叫子節點的 next()，子節點再呼叫它自己子節點的 next()，一路傳到 leaf（scan）。leaf 讀一列出來、往上回傳一列、每一層都可以做點事（filter、project）、root 拿到最後結果
傳統 DBMS 幾乎都是這一派，一次一行（row-at-a-time）
Push-based（compilation model，HyPer 2011 之後）：資料從上游往下推
整個 plan tree 被 JIT 編譯成一個大迴圈，leaf 讀進一列就一路呼叫下去，每一層 inline 到同一個 function 裡，資料留在暫存器不落記憶體
這是把整棵 plan 編譯成 machine code 而 Photon、Umbra、HyPer 是代表
兩種其實都各有好壞
pull 好在動態組合
運算之間介面單純，加新運算子不用重新編譯整棵樹，而且 backpressure 是天然的：下游沒拉，上游就沒事發生，也就是板前那句「沒人點就停手」
push 好在性能極致
資料不落記憶體、branch 少、CPU 榨得乾，代價是編譯成本高，而且加一個新運算要重寫 codegen，等於停線改機台。
DataFusion 選了 pull-based，但不是傳統的一次一行，而是一次一個 RecordBatch（欄式的、8192 行一批）
這個組合 pull-based + columnar batch，在 2020 年之前很少見，DataFusion 跟 DuckDB 這一輩才變主流。
一次一 batch 這件事有兩個好處：
攤銷 interpret 成本：傳統 Volcano 每 row 都要走一遍 virtual call chain（next() 是虛函式），interpret 成本高到爆。用板前的話講，就是每一貫都要重新對話一次「還要嗎」「要」「好」，改成一次一 batch，這個對話成本一次攤到 8192 行上，等於沒有
向量化天然發生：一個 filter 節點拿到一整個 RecordBatch，內部用 Arrow compute kernel 處理，SIMD 天然發生，不需要 codegen。師傅面對一整盤同樣的料，可以一刀切八片
也就是 DataFusion 拿到了 pull 的組合彈性跟 batch 的向量化紅利，代價是沒有 push 那種資料留在暫存器的極致
那 DataFusion 具體怎麼實作這個 pull the batch ？
答案是把 pull-based 跟 Rust 的 async 湊在一起：
ExecutionPlan::execute(partition) 回傳一個 SendableRecordBatchStream
Stream 是 Rust futures crate 的 trait，poll_next 拿 Poll<Option<Result<RecordBatch>>>
await 上游的 poll_next：拿到 Poll::Ready(Some(batch)) 就處理、拿到 Poll::Pending 就交回 runtime、拿到 Poll::Ready(None) 就是 EOF用一個很小的心智模型講：
每個 ExecutionPlan 節點都是一個 async iterator.next().await，它給下一個 batch，沒 ready 就把 CPU 讓給別人
這個設計的關鍵差別是：傳統 Volcano 的 next() 是 blocking，DataFusion 的 poll_next 是 async。
回到壽司店的比喻
blocking 版的服務生跟廚房點單後就站在出餐口等，菜沒好就一直站著，這時候他不能去幫別桌倒水點餐，一個人被一張單綁死。
async 版的服務生問「好了嗎」，廚房說「還沒」（這就是 Poll::Pending），他馬上轉頭去服務別桌，廚房好了會叫他，所以說同一個服務生可以同時掛著十張單
這裡有個常見反直覺點，教科書講 async runtime 是給 I/O 密集用的（socket、disk、timer）：你 await 一個 I/O 完成、runtime 去做別的事、I/O 完成再回來。
CPU 密集不是 async 的主場，因為 CPU 密集沒得 await，工作在跑就是在跑、跑完才 return，中間沒有等外部完成的 yield 點
做壽司是純體力活，哪來的等？
Andrew Lamb 有一篇 blog（"Using Rustlang's Async Tokio Runtime for CPU-Bound Tasks"）專門講這件事，可以整理成三個論點：
論點一：Tokio 自帶排班表 一家店要有人分配哪個師傅接哪一單，還要避免有人閒著有人忙死。自己寫這套很麻煩，要處理 queue、要處理 core affinity、要處理 work-stealing。Tokio scheduler 這些本來就有，spawn 一堆 task 進去，scheduler 自己 work-steal，多核心的利用免費拿到。
論點二：有些菜就是要等 sort、hash aggregation、hash join 的 build side 這些是 pipeline breaker，下游要等上游把資料物化完才能開始拉。用 async 表達這件事很自然，await 那個 stream，runtime 幫你排。相對之下，如果用純 CPU 平行框架（像 Rayon），pipeline breaker 之間要自己寫同步原語（channel、barrier），寫起來就沒 async 直接。
論點三：真的會 block 的操作，要用 API 標出來 有些工作是真的會佔死執行緒很久，這種要 spawn_blocking 丟到別的 pool，避免佔死 tokio worker。tokio worker 一被佔死，掛在上面的所有 async task 都停擺。async 世界的界線在這裡很清楚：這件事會 block 是一個要顯式宣告的性質。
DataFusion 選擇讓 CPU 工作跑在 tokio 上，換到的是排班和 async 語意的方便，付出的是 runtime 隔離要靠使用者自己處理
把上面拼起來，一條 SQL 在 DataFusion 跑起來的路徑大概是：
SessionContext::sql(query)
│
▼
LogicalPlan (Analyzer + Optimizer)
│
▼
PhysicalPlan (ExecutionPlan tree)
│
│ execute(partition) → SendableRecordBatchStream
▼
spawn 到 tokio runtime
│
┌─────┴──────────────────────────────────┐
│ Root (CollectExec) │
│ poll_next.await │ ← pull 從這裡往下傳
│ │ │
│ AggregateExec (Final) │ ← pipeline breaker
│ poll_next.await │ 吃完輸入才吐第一筆
│ │ │
│ RepartitionExec (Exchange) │ ← 並行在這裡發生
│ 每個 partition 一條 stream │ 每條 spawn 成 tokio task
│ │ │ 串流，不是 pipeline breaker
│ FilterExec │
│ poll_next.await │
│ │ │
│ DataSourceExec / Parquet │ ← decode 也在 tokio 上跑
│ │ │
└───────┴────────────────────────────────┘
│
▼
Poll::Ready(None) = EOF
幾個關鍵動作：
poll_next 一路往下呼叫，傳到 leafSortExec、AggregateExec（Final）、HashJoinExec 的 build side，下游 await 上游把資料物化完RepartitionExec 把資料 split 成 target_partitions 條 stream，每條 spawn 成 tokio task，scheduler 自己 work-stealSpark 的 task 是 blocking 的，JVM 有一個 thread pool，一個 task 佔一條 thread，跑完 return，這是 Spark 從 2010 年沿用至今的模型。DataFusion 的 Stream 是 async 的，poll_next 需要一個 runtime 才能推進。
用壽司店講：Spark 那邊是老派食堂，服務生接一張單就站在窗口等到菜出來為止，不會走開。DataFusion 這邊的服務生本來會跑來跑去。
兩者要對接，只有兩條路：
Comet 走第二條，在 native 側持有一個 tokio runtime，用 block_on 把 stream 推進：JVM 側每要一個 batch，就透過 JNI 呼叫 Rust，Rust 這邊 block_on(stream.next()) 拿到下一個 batch，然後把 Arrow C Data Interface 的指標交回 JVM。
等於是讓新派服務生配合老派站著等，這可以工作但有兩個結構性成本：
block_on 包起來就退化成「等就是等」，Rust 這邊的 async 排程只在同一次 poll_next 內起作用這是 pull + async 硬要跟 blocking runtime 對接的結構性代價，不是實作 bug。D26 深挖 JNI 邊界時會再回來，那時候會看到 Comet 為了把這趟路走短做了哪些細部優化（0.17 快取 offsetBufferAddress 就是其中一個）。
DataFusion 的執行模型講完了：pull-based + columnar batch + tokio，加上 Exchange operator 提供並行。這幾件事湊在一起讓 pipeline breaker 的同步、多核心的排班、跨 partition 的資料重分配全部有一個統一的語意骨架。Comet 在 JNI boundary 上把它同步等出來給 JVM 用，這是結構性代價。
明天 D17 接執行的第二個關鍵題：多個運算搶記憶體怎麼辦？
MemoryPool 有兩種公平（greedy / fair），Comet 接點：JVM 那邊有自己的 Spark memory manager、DataFusion 這邊有 MemoryPool，兩個池怎麼協調是 Comet 的真實痛點。
RepartitionExec API 文件（Exchange operator 的設計說明）: https://docs.rs/datafusion/latest/datafusion/physical_plan/repartition/struct.RepartitionExec.html
tokio::task::spawn_blocking 官方文件: https://docs.rs/tokio/latest/tokio/task/fn.spawn_blocking.html
iThome鐵人賽
看影片追技術