Apache Iceberg 中引入索引提升查詢性能
Apache Iceberg 是一種開源數據 Lakehouse 表格式，提供強大的功能和開放的生態系統，如：Time travel，ACID 事務，partition evolution，schema evolution 等功能。
本文將討論火山引擎EMR團隊針對 Iceberg 組件的優化思路，通過引入索引來提高查詢性能。
采用 Iceberg 構建數據湖倉
火山引擎 E-MapReduce（簡稱 EMR）是火山引擎數智平臺（VeDI）旗下的云原生開源大數據平臺產品， 提供了企業級的 Hadoop、Spark、Flink、Hive、Presto、Kafka、StarRocks、Doris、Hudi、Iceberg 等大數據生態組件，100% 開源兼容，可以幫助企業快速構建企業級大數據平臺，降低運維門檻。秉承業界領先的 EMR Stateless 理念，火山引擎 EMR 可以實現集群級別的彈性伸縮，即無業務需求時釋放集群，有業務需求時再拉起集群，配合智能化的冷熱數據分層存儲能力，助力企業在大數據基建領域進一步降本提效。
基于火山引擎 EMR 產品，可以構建數據湖倉、近實時數倉、實時數倉等場景。例如，使用 Iceberg 構建數據湖倉，從 ODS 到 DWD 等不同的分層進行建模，將數據 HFDS 或 TOS（火山引擎對象存儲產品）上，然后采用 Trino 或者 Spark 去做分析。
如何加速查詢性能，使其盡可能接近專門的分布式數倉（如 ClickHouse 等），是需要思考和探究的問題。
索引是業界常用的提高查詢性能的手段之一，針對 Iceberg 我們也采用了增加索引的方式。對常用的列字段構建 Index，在進行 table scan 時利用 Index 只返回匹配的數據，降低匹配數據量，從而大大提高查詢性能。
Iceberg 介紹
介紹 Iceberg Index 功能之前，我們先簡單介紹下 Iceberg 的架構。Iceberg 具有分層的元數據架構，如下如所示。
Spark、Presto、Flink 等多種引擎讀取 Iceberg 的數據，就是利用分層的元數據找到 data file 列表。例如，Spark 引擎解析 SQL 語句，然后調用 Iceberg 的接口，獲取 data file 并進行 task 切分。
在 Manifest file 中記錄了 data file 中字段的最大值和最小值。
"data_file": {
"content": 0,
"file_path": "hdfs://emr-cluster/warehouse/hive/db.db/sample/data/ts_day=2020-12-31/category=diamond/00000-0-220aa9a6-4530-499f-9450-da946d667624-00001.parquet",
"file_format": "PARQUET",
......
"lower_bounds": {
"array": [{
"key": 1,
"value": "\u0006\u0000\u0000\u0000"
}, {
"key": 2,
"value": "diamond"
}, {
"key": 3,
"value": "\u0000\u0004ü??·\u0005\u0000"
}]
},
"upper_bounds": {
"array": [{
"key": 1,
"value": "\u0007\u0000\u0000\u0000"
}, {
"key": 2,
"value": "diamond"
}, {
"key": 3,
"value": "\u0000¨od?·\u0005\u0000"
}]
},
......
}利用這些信息，可以進行 data file 級別的初步過濾，把不符合條件的 data file 過濾掉，進而減少一部分數據的讀取。
實現索引的必要性
既然 Iceberg 已經提供 data file 級別的過濾。為什么我們還需要引入索引呢？以下面例子進行介紹，左邊兩個表格分別是 data file 文件里面的內容，右邊表格是 data file 對應的 manifest file。
針對SELECT * FROM table WHERE age > 50，利用 min-max 統計信息，很容易發現 data file 1 中沒有滿足條件的數據，因此 data file 1 就不會參與計算。
但是針對多維分析，如name = 'LiLy' AND age > 30，利用name和age的min-max的統計信息分別對條件name = 'LiLy'和age > 30進行判斷，得到
data file 1 和 data file 2 都滿足條件。然而，仔細分析 data file 1 和 data file 2
的數據，并不存在符合條件的數據，因此 min-max 過濾效果不太理想。所以通過引入合適的索引功能，可以提高 data skipping
的概率，提高查詢性能。
1. 首先探究索引類型
索引類型有多種，如 BloomFilter、Ribbon Filter、Dictionary Index、BitMap 等。為了滿足多維分析場景，我們選擇了[Range-Encoded BitMap]https://www.featurebase.com/blog/range-encoded-bitmaps （ Base-2, Bit-sliced Index），可適用于高基數場景，滿足=、<、>、IN、BETWEEN 等操作的多維分析。
例如，對上面的
name 和 age 兩列分別計算索引信息。由于 name 屬于字符串類型，需要先進行字典編碼再進行計算索引信息。采用
Range-Encoded 技術，根據數據的二進制相關信息以及對應的 pos 信息生成索引數據。利用索引數據分析得到，同時滿足name = 'LiLy' 和age > 30的數據不在同一行，恰好可利用 Range-Encoded 的交并運算將數據進行過濾掉，因此 data file 1 不用參與計算。
也就是說，BitMap 的交并運算可以更好地在復雜過濾條件的情況下過濾掉更多的數據文件。
2. 接下來探究索引的粒度。
Iceberg 提供的 min-max，也是一種文件級別的索引。文件級別的索引就是根據 filter 條件過濾掉不符合條件的 data file。文件級別的索引可適用于多種文件類型，但這種粒度比較粗，只要 data file 中有一條數據符合條件，該 data file 中的數據就會全部讀取出來參與計算，從而影響 SQL 的查詢性能。
對于 Parquet、ORC 的文件格式，提供有 file chunk 的概念(row group or stripe)，我們完全可以按照 row group / stripe 粒度，對數據進行過濾。（為了方便描述，我們將 row group 和 stripe 統稱 split。）
如：SQL語句：SELECT * FROM table WHERE col_1> v1 AND col_2 = v2，其中對 col_1 字段和 col_2 字段已構建 Index 信息。現在利用索引對 SQL 語句作用。
SQL
語句解析后，將符合條件的 data file 列表進行切分后，得到很多 split 的列表。利用索引，分析 split
中數據是否滿足條件，如果不滿足則跳過。如上圖 data file 列表切分后，得到數萬級別數量的 split 列表。將索引數據作用在
split1，發現 split1 中沒有同時col_1> v1 AND col_2 = v2滿足條件的數據，該 split1 中的數據就不會參與計算。最后處理后，只得到了少量的 split 列表，數據過濾度達到 10% 以上，查詢性能有明顯提升。
因此，采用 row group / stripe 級別的細粒度索引，可以過濾大部分數據。
細粒度索引實現邏輯
Iceberg 元數據中 manifest file 中除了提供 min-max 等統計信息，還提供有 split 相關信息："split_offsets":{"array":[4,...]}，極大方便我們實現 row group / stripe 級別的細粒度索引。
- 提供索引的構建 API
Iceberg 中提供構建索引的 API，引擎端調用該 API 即可實現索引構建功能。對于 Spark 3.3 及以上版本，已經提供有索引的 SQL 語句，在 Iceberg 的 Spark 模塊實現 Spark 提供的索引接口即可。
- 構建索引
我們采用異步構建索引，不影響主線任務。也提供了增量構建索引功能，只對 append 數據進行構建索引。調用 TableScan 讀取數據，按照 data file 的 split offset 切分數據，進行構建索引，并保存索引數據和對應的元數據信息。為了避免出現小文件存在，我們會進行索引數據合并。
- 索引文件存儲
索引文件格式采用[puffin]https://iceberg.apache.org/puffin-spec/格式，這是一種二進制格式。 Magic Blob? Blob? ... Blob? Footer
在 Footer 中保存每個 blob 的元數據信息。索引構建成功后，會生成類似于下面內容的文件。
索引帶來的收益
Range-Encoded BitMap 適用于多維分析場景，且 Ranger 范圍較小時，效果非常明顯。下面我們基于 Spark 引擎性能測試。
- 構造 1TB 的 SSB 測試數據，分別在構建 Index 前后，對以下用例進行測試。
Q1: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice = 19665277
Q2: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice = 19665277 AND lo_revenue = 2141624
Q3: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice = 19665277 AND lo_revenue >=10304000
Q4: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice = 21877827 AND lo_revenue >= 83800 AND lo_revenue <= 103800
Q5: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice > 21877827 AND lo_revenue >= 83800 AND lo_revenue <= 93800
Q6: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice >= 93565 AND lo_ordtotalprice < 93909
Q7: SELECT count(*) FROM lineorder WHERE lo_ordtotalprice >= 93565 AND lo_ordtotalprice < 91003562 AND lo_revenue >=904300 AND lo_revenue <= 9904300
左圖展示了 7 條 SQL 語句分別在沒有 Index 和采用 Index 情況下的執行時間。右圖展示采用 Index 后，7 條 SQL 語句讀數據的 split 數量。很明顯讀數據的 split 數量越少，Index 效果越好。最糟糕的情況，所有的 split 都參數計算，這時和沒有構建索引的效果類似。
- 采用 SSB 基準測試
由于 SSB 提供的測試場景，和 Range-Encoded 有利的場景，不太匹配，所以 Index 的效果并沒有明顯的效果。但也不會比不采用 Index 的效果差。如下面左圖，分別是構建索引前后，SQL 語句的執行時間，構建索引的優勢并沒有體現出來。右圖中，可以看到所有的 split 都參與了計算。
總結
根據上面的介紹，這里總結下 Iceberg 中索引實現的一些特征：
- 細粒度索引級別：提供 RowGroup/Stripe 級別的索引，可以更加精確的定位數據的查詢范圍，減少不必要數據輸入，從而提高查詢性能；
- 索引作用于執行端：查詢任務被分配多個執行端，每個執行端只判斷該節點上的 RowGroup/Stripe 數據是否符合即可；
- 適配多種引擎：索引構建后，可用于多種引擎；
- 提供異步構建 Index，從而不影響主業務的進行；
- 適用于高基數 & 低基數場景，且占有存儲空間小。滿足范圍查詢、等值查詢等場景。且范圍越小，收益效果越明顯。
