DataFusion внедряет оптимизацию Projection Pushdown для минимизации чтения ненужных столбцов

上一篇介紹了 Query Optimizer 的目的，今天深入其中一項常見最佳化： Projection Pushdown 它的核心概念是： 查詢只需要哪些欄位，就盡量只讀取哪些欄位。 這能減少不必要的 I/O、資料解碼、記憶體使用與後續運算量。 有讀者提問： Filter Pushdown 是否能真正發生，取決於資料來源是否支援這項能力。這句話是什麼意思？ 答案是：不同資料來源確實會影響 Filter Pushdown 能否在「資料讀取階段」發生。 假設查詢是： SELECT city, amount FROM orders WHERE is_member = true; Optimizer 可以知道這段查詢有一個篩選條件： is_member = true 但接下來有兩種可能： 資料來源本身可以理解或處理這個條件： 資料來源只讀取符合條件的資料 例如 Parquet 可以利用 Row Group Statistics，跳過確定不符合條件的 Row Group。 DataFusion 仍然可以建立： FilterExec 但流程會變成： 先讀取資料 ↓ 建立 RecordBatch ↓ 由 FilterExec 篩選 也就是查詢結果仍然正確，只是無法在最底層減少資料讀取量。 因此要區分： Optimizer 知道有 Filter 與： 資料來源真的能在讀取時套用 Filter 這兩件事不一定相同。 Parquet：欄位裁剪 + Row Group Pruning CSV：讀取並解析整列後再 Filter MemoryTable：MemoryExec → FilterExec 外部資料庫：DataFusion Filter → 資料來源條件 不同資料來源對 Projection Pushdown 與 Filter Pushdown 的支援方式不同；Parquet 可利用欄式儲存與統計資訊減少資料讀取，CSV 則通常需要先解析資料列。 Parquet 是欄式檔案格式，資料會以欄位與 Row Group 組織，並儲存一些統計資訊。 例如查詢： SELECT city, amount FROM orders WHERE amount > 10000; 如果某個 Row Group 的統計資訊顯示： amount 最小值：100 amount 最大值：5000 DataFusion 就能知道這個 Row Group 不可能有 amount > 10000 的資料，因此直接跳過。 Row Group 1：100 ~ 5,000 → 跳過 Row Group 2：8,000 ~ 20,000 → 讀取 此時 Filter Pushdown 不只是把 Filter 放到計畫較前面，而是可能進一步減少實際的檔案掃描量。 CSV 是文字格式，欄位位於同一列中，通常沒有 Parquet 那種 Row Group Statistics。 例如： city,amount,is_member Taipei,1200.0,true Taichung,850.0,false 執行： CSV 讀取器通常需要先讀取並解析每一列，DataFusion 才能判斷： is_member 是否為 true 因此流程比較接近： 讀取 CSV ↓ 解析成 RecordBatch ↓ FilterExec 篩選 is_member = true 查詢結果仍然正確，但不一定能像 Parquet 一樣在磁碟層級跳過大量資料區塊。 如果資料已經在記憶體中，例如 MemTable，資料讀取成本與檔案格式不同。 資料可能已經存在： Arrow RecordBatch 此時 DataFusion 通常直接建立： 在記憶體中的 RecordBatch 上進行篩選。 MemoryExec ↓ FilterExec ↓ 符合條件的資料 因為資料已經在記憶體中，Pushdown 的重點不一定是減少磁碟 I/O，而是讓較少資料流入後續算子。 如果 DataFusion 使用自訂 TableProvider 連接外部資料庫，資料來源可能將條件轉換成資料庫自己的查詢： 這樣外部資料庫可以先執行篩選，再將結果傳回 DataFusion。 DataFusion Filter ↓ 轉換成資料來源條件 ↓ 外部資料庫先篩選 ↓ 只回傳符合條件的資料 是否能做到這一步，取決於自訂 TableProvider 與資料來源的實作。 在 SQL 中，Projection 通常指的是 SELECT 指定的輸出欄位。 SELECT city, amount FROM orders; 這段查詢只需要： city amount 假設 orders 的完整 Schema 是： order_id: Int64 city: Utf8 amount: Float64 is_member: Boolean created_at: Date32 這次查詢不需要： order_id is_member created_at 如果資料來源仍然讀取所有欄位，就會產生額外成本。 未進行欄位裁剪時，流程可能是： 讀取所有欄位 ↓ 建立完整 RecordBatch ↓ 只保留 city、amount ↓ 輸出結果 也就是： order_id ─┐ city ├─ 讀取 → 建立 RecordBatch → 丟棄不需要欄位 amount │ is_member ┤ created_at ┘ 即使最後只使用兩個欄位，其他欄位可能已經被讀取與解碼。 啟用欄位裁剪後，DataFusion 會將欄位需求往資料來源方向傳遞： SELECT city, amount ↓ 需要 city、amount ↓ 資料來源只讀取 city、amount 流程變成： 只讀取 city、amount ↓ 建立較小的 RecordBatch ↓ 交給後續算子處理 不需要的欄位不會進入資料流： order_id ✗ city ✓ amount ✓ is_member ✗ created_at ✗ 原始 Logical Plan 可能是： Projection: city, amount TableScan: orders Optimizer 會分析整棵計畫，確認 TableScan 真正需要的欄位： Projection: city, amount TableScan: orders projection=[city, amount] 這裡的： projection=[city, amount] 表示資料來源只需要提供這兩個欄位。 Projection Operator 與 Projection Pushdown 不是同一件事： Projection → 產生查詢要求的輸出欄位 Projection Pushdown → 讓資料來源只讀取必要欄位 Projection Pushdown 對 Parquet 特別有效。 假設 Parquet 檔案包含： order_id city amount is_member created_at 資料來源有機會只讀取： 而不必載入其他欄位。 這可以減少： 欄位越多、查詢只使用少數欄位時，效果通常越明顯。 考慮以下查詢： 這段查詢需要的欄位是： city amount is_member 雖然 is_member 不會出現在輸出結果中，但它是 WHERE 條件需要的欄位，因此不能被裁掉。 最佳化後可能接近： Projection: city, amount Filter: is_member = true TableScan: orders projection=[city, amount, is_member] 這裡同時使用了兩種最佳化： Projection Pushdown → 只讀取 city、amount、is_member Filter Pushdown → 嘗試盡早套用 is_member = true order_id 與 created_at 仍然不需要讀取。 可以在 DataFusion CLI 執行： EXPLAIN FORMAT INDENT SELECT city, amount FROM orders WHERE is_member = true; 可能會看到類似： logical_plan Projection: orders.city, orders.amount Filter: orders.is_member = Boolean(true) TableScan: orders projection=[city, amount, is_member] 重點是觀察： projection=[city, amount, is_member] 這表示計畫層知道查詢只需要這三個欄位。 但要注意，這不一定代表每一種資料來源都能在最底層真正只讀取這三欄。實際效果仍取決於資料來源是否支援欄位裁剪。 可以使用： EXPLAIN VERBOSE SELECT city, amount FROM orders WHERE is_member = true; 觀察最佳化前後的計畫是否發生變化。 也可以比較不同資料來源： Parquet → 可能出現欄位裁剪與 Row Group Pruning CSV → 可能仍需先解析整列資料，再由 FilterExec 篩選 MemoryTable → 通常由 MemoryExec 提供資料，再由 FilterExec 處理 即使 FilterExec 出現在 Physical Plan 中，也不代表資料來源沒有支援 Pushdown；有些情況會同時存在資料來源層級篩選與執行算子篩選。 假設資料表有 100 個欄位，但查詢只使用 3 個欄位： 完整讀取 → 100 個欄位 Projection Pushdown → 只讀取 3 個欄位 可能降低的成本包括： I/O 成本 讀取較少的檔案資料 解碼成本 解析較少的欄位 記憶體成本 建立較少的 Arrow Array 傳遞成本 RecordBatch 包含較少欄位 運算成本 下游算子處理較少資料 實際改善程度仍取決於： Projection Pushdown 並不是任何情況都能完整套用。 SELECT *SELECT * FROM orders; 查詢要求所有欄位，因此沒有欄位可以裁剪。 雖然輸出只需要 city 與 amount，但 is_member 是篩選條件需要的欄位，因此仍然必須讀取。 即使 Logical Plan 已經知道需要哪些欄位，資料來源仍可能必須讀取完整資料，最後再由執行算子保留必要欄位。 計畫層知道需要哪些欄位 資料來源真的只讀取這些欄位 兩者不一定完全相同。 今天介紹了 Projection Pushdown，並回應了不同資料來源對 Pushdown 的影響。 我們學到： TableProvider 可能自行處理下推條件EXPLAIN 可以觀察計畫層的欄位需求與執行算子 可以用一句話總結： Projection Pushdown 讓查詢少讀不需要的欄位；Filter Pushdown 讓查詢盡早排除不符合條件的資料，而兩者能否真正降低 I/O，取決於資料來源是否支援相應能力。 下一篇將介紹 Filter Pushdown，進一步觀察 WHERE 條件如何從查詢計畫一路傳遞到資料來源，以及 Parquet 如何利用統計資訊跳過不必要的 Row Group。