Apache Hudi 学习笔记 05：COW 与 MOR 表类型，以及三种查询方式

前面几篇已经把 Hudi 的定位、测试环境、Timeline、FileGroup、FileSlice 和索引机制串起来了。到这里再看 COW 和 MOR，就不是背两个名词，而是在回答一个很实际的问题：更新数据写进数据湖以后，到底是在写入时就整理好，还是先快速写进去，等读取或后台服务再合并？

Hudi 的两种表类型 Copy On Write 和 Merge On Read，本质上就是围绕这个问题做的取舍。COW 更像“写时整理”，MOR 更像“读时合并”。它们都会通过 Timeline 保证事务提交，但文件形态、写入延迟、查询成本、compaction 依赖完全不同。

1. 表类型与查询语义

1.1 背景问题

还是用订单表举例。

第一次写入三条订单：

order_id=2001, status=CREATED
order_id=2002, status=CREATED
order_id=2003, status=CREATED

过几分钟，订单 2001 支付成功，同时新增订单 2004：

order_id=2001, status=PAID
order_id=2004, status=CREATED

如果这张表是普通 Parquet 表，通常有两种处理方式：

1. 重写整个分区，把 2001 改成 PAID，再加上 2004。
2. 追加一批增量文件，查询时自己合并最新状态。

第一种方式查询简单，但写入成本高；第二种方式写入快，但查询复杂。Hudi 的 COW 和 MOR 也是围绕这两类思路设计的，只是它把文件版本、提交状态、索引定位、增量读取和表服务都收敛到了表自身。

1.2 表类型和查询类型先分清

Hudi 里有两组概念容易混在一起：

类型	关注点	常见取值
表类型	数据如何写入和组织文件	Copy On Write、Merge On Read
查询类型	读取时看什么文件视图	Snapshot Query、Incremental Query、Read Optimized Query

表类型决定的是写入和存储形态。查询类型决定的是读取视角。

一张 COW 表可以做 Snapshot Query，也可以做 Incremental Query。一张 MOR 表可以做 Snapshot Query、Incremental Query，也可以做 Read Optimized Query。不要把 COW 等同于 Snapshot，也不要把 MOR 等同于 Read Optimized。

1.3 Copy On Write

Copy On Write 通常简称 COW。

COW 表中，数据主要存放在 Base File 里，通常是 Parquet 文件。发生更新时，Hudi 不会原地修改旧 Parquet，而是读取旧文件、合并新记录、写出新的 Base File，再通过 Timeline 提交新的 FileSlice。

图示如下：

这张图主要说明 COW 表在更新时会读取旧 Base File、合并新记录并写出新的 Base File，成本主要发生在写入阶段。

可以把 COW 理解成：

更新到来
  -> 找到旧 Base File
  -> 读取旧记录
  -> 合并新记录
  -> 写出新 Base File
  -> 提交 commit

1.3.1 COW 的优点

COW 最大的优点是查询简单。

读取端通常只需要读当前有效的 Base File，不需要在查询时合并大量增量日志。Hive、Spark SQL、Trino 这类查询引擎读取 COW 表时，整体模型更接近普通 Parquet 表。

这也意味着 COW 更适合：

1. 读多写少的表。
2. 更新比例不高的事实表。
3. 查询延迟要求稳定的明细表。
4. 需要让多种查询引擎低成本读取的表。

1.3.2 COW 的缺点

COW 的代价在写入侧。

如果一个 Base File 里只有一条记录发生变化，COW 仍然可能要重写整个 Base File。更新比例越高、文件越大、写入越频繁，写放大越明显。

所以 COW 不适合所有场景。如果表是高频 CDC 更新，写入端会承受更大的 merge 和 rewrite 成本。

1.4 Merge On Read

Merge On Read 通常简称 MOR。

MOR 表中，一个 FileSlice 可以由 Base File 和 Log File 组成。更新到来时，Hudi 可以先把变更追加到 Log File，降低写入延迟；读取最新状态时，再把 Base File 和 Log File 合并起来。

图示如下：

这张图主要说明 MOR 表可以先把更新写入 Log File，再通过 Snapshot Query 或后台 Compaction 完成合并。

可以把 MOR 理解成：

更新到来
  -> 找到 FileGroup
  -> 典型情况下把增量追加到 Log File
  -> 提交 deltacommit
  -> Snapshot Query 读取时按需要合并 Base File + Log File
  -> 后台 Compaction 合并成新的 Base File

1.4.1 MOR 的优点

MOR 的优势在写入侧。

更新不一定马上重写 Base File，可以先追加 Log File。这对实时入湖、CDC 写入、低延迟写入很有吸引力。

MOR 更适合：

1. 写入频繁、更新比例高的表。
2. 希望降低写入延迟的 CDC 入湖。
3. 可以接受后台 compaction 的场景。
4. 查询侧愿意在新鲜度和性能之间做权衡的场景。

1.4.2 MOR 的缺点

MOR 的复杂度在读取和表服务。

如果 Log File 积累太多，Snapshot Query 每次都要合并 Base File 和 Log File，查询会变慢。为了控制长期查询性能，需要配置 compaction，把 Log File 合并进新的 Base File。

所以 MOR 的核心不是“写入变快就完了”，而是要持续管理：

1. compaction 触发频率。
2. Log File 大小。
3. Snapshot Query 的合并成本。
4. Read Optimized Query 能接受的数据延迟。
5. Flink / Spark 写入任务和 compaction 任务的资源竞争。

1.4.3 MOR 写入不等于所有数据都写 Log File

刚开始学 MOR 时，很容易把它理解成“所有写入都先进 Log File”。这个理解不够准确。

MOR 的文件形态会受到索引能力和写入引擎影响。以 Spark 常见的 Bloom Index 为例，Bloom Index 的定位能力更多依赖 Base File 侧的文件级信息，因此在 Spark 小规模 DataSource 示例中，MOR 表也可能先写出 Base File，未必立刻形成明显的业务 Log File。UPDATE 是否追加到 Log File，还会受到 small file 策略、索引、写入引擎和 compaction 状态影响。

在 Flink 长流写入中，如果使用 State Index，任务可以在 state 中维护 key 到 FileGroup 的映射，写入行为又会不一样。它更容易持续追加 Log File，并在达到大小阈值后 roll over。

所以判断 MOR 表不要只看“有没有 .log 文件”这一件事。更可靠的判断方式是：

1. 看 hoodie.table.type 是否为 MERGE_ON_READ。
2. 看 Timeline 中是否出现 deltacommit。
3. 看当前写入引擎和索引是否支持对 Log File 做有效定位。
4. 看 compaction 是否已经把 Log File 合并成新的 Base File。

1.4.4 Compaction 有内联和异步两种思路

MOR 表迟早要面对 compaction。Log File 越积越多，Snapshot Query 的合并成本就越高；compaction 太频繁，又会挤占写入资源。

常见思路有两种：

模式	含义	适合场景
Inline Compaction	写入任务内顺带执行 compaction	单任务、低复杂度、数据量不大的测试或简单链路
Async Compaction	写入和 compaction 分开调度	持续流式写入、希望隔离写入延迟和整理资源

触发策略也有多种，学习阶段最容易理解的是按 delta commits 数量触发。比如按 delta commits 数量触发：每产生若干次 deltacommit，就安排一次 compaction。具体参数名和默认值需要按 Spark / Flink 写入引擎与 Hudi 1.0.2 文档确认，后续实测文章再展开。

后面讲 Flink 参数和底层读写原理时，还会看到更具体的配置，例如 compaction 是否异步、多少个 delta commits 触发、compaction task 并行度、每次 compaction 能处理多少 IO。本文先记住一件事：MOR 的性能不是只由写入决定，compaction 才是长期稳定运行的关键。

1.5 COW 和 MOR 对比

先用表格放在一起看。

维度	Copy On Write	Merge On Read
文件形态	主要是 Base File	Base File + Log File
写入方式	更新时合并并重写 Base File	更新可先追加 Log File
写入延迟	较高	较低
Snapshot Query	读当前 Base File	合并 Base File + Log File
Read Optimized Query	通常不单独强调，Snapshot 读取最新 Base File 即可	只读已压缩 / 列式 Base File，可能不包含未合并 Log File 中的最新变更
Compaction 依赖	不强	强
适合场景	读多写少、查询稳定	写入频繁、近实时入湖、可以通过 compaction 管理查询成本的场景
主要风险	写放大	Log File 膨胀、compaction 滞后

我的经验判断是：

COW = 写入时付成本，查询时少付成本
MOR = 写入时少付成本，查询或 compaction 时再付成本

如果一张表是 BI 查询、离线分析、更新不频繁，COW 通常更省心。如果一张表持续接收 CDC，业务希望几分钟内可见，MOR 更值得考虑。

1.6 三种查询方式

在本文这一阶段，先重点理解三类最常用的查询视角：Snapshot Query、Incremental Query、Read Optimized Query。CDC Query、Time Travel Query 等能力后续再单独展开。

图示如下：

这张图主要说明表类型和查询类型是两组概念：COW / MOR 决定文件组织方式，Snapshot / Incremental / Read Optimized 决定读取视角。

不同表类型对查询方式的支持可以先这样理解：

表类型	Snapshot Query	Incremental Query	Read Optimized Query
Copy On Write	支持，读取最新 Base File	支持，按 commit 范围读取	通常不单独强调，直接读 Base File 已经是优化后的列式文件
Merge On Read	支持，需要合并 Base File 和 Log File	支持，按 commit / deltacommit 范围读取	支持，只读已压缩 / 列式 Base File，可能牺牲新鲜度

也就是说，Read Optimized Query 主要是为了 MOR 表存在的。COW 表本身没有未合并的 Log File，读取最新 Base File 已经足够直接。

MOR 表的写入类 Instant 常见为 deltacommit，COW 表常见为 commit。

1.6.1 Snapshot Query

Snapshot Query 查询最新表状态。

对 COW 表来说，Snapshot Query 通常读取最新有效的 Base File。

对 MOR 表来说，Snapshot Query 需要把 Base File 和 Log File 合并，返回最新状态。也就是说，Snapshot Query 更关注数据新鲜度。

1.6.2 Incremental Query

Incremental Query 按 commit 范围读取增量。

它适合下游增量 Pipeline。比如 DWD 层任务记录自己上一次消费到的 commit，下次只读取这个 commit 之后的数据变化。

但要注意：Incremental Query 不等同于数据库 binlog 那种逐行 before / after 明细。在某些 COW 写入和读取配置下，更新重写 Base File 后，增量读结果可能包含同一 FileGroup / FileSlice 中被带出的记录。因此下游不能只依赖“这一定是业务变化行”的假设，而要结合 record key、commit time、precombine 字段做幂等处理。

如果后续需要精确变更明细，应单独研究 Hudi CDC Query / changelog 相关能力，而不是把普通 Incremental Query 当成数据库 CDC 明细。

1.6.3 Read Optimized Query

Read Optimized Query 只暴露当前可用于读优化视图的 Base / Columnar 文件，不合并尚未压缩的 Log File。

对 MOR 表来说，它可以绕开 Log File 合并，查询成本更低，但可能看不到最新 Log File 中尚未 compaction 的变更。

这是一种典型取舍：

Snapshot Query: 更新鲜，但可能要合并 Log File
Read Optimized Query: 更快，但可能不是最新

2. 实测验证

2.1 Spark 示例：创建 COW 和 MOR 表

下面用当前环境真实跑一遍。脚本会创建两张表：

hdfs://localhost:9000/hudi/demo_cow_query_orders
hdfs://localhost:9000/hudi/demo_mor_query_orders

核心脚本如下：

cat > /tmp/hudi_table_query_demo.py <<'PY'
from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("hudi-table-query-demo")
    .master("local[2]")
    .config("spark.sql.catalogImplementation", "in-memory")
    .getOrCreate())

spark.sparkContext.setLogLevel("WARN")

cow_path = "hdfs://localhost:9000/hudi/demo_cow_query_orders"
mor_path = "hdfs://localhost:9000/hudi/demo_mor_query_orders"

rows1 = [
    (2001, "2025-07-17", "CREATED", 120.5, 20250717195000),
    (2002, "2025-07-17", "CREATED", 35.0, 20250717195010),
    (2003, "2025-07-17", "CREATED", 78.8, 20250717195020),
]

rows2 = [
    (2001, "2025-07-17", "PAID", 120.5, 20250717195300),
    (2004, "2025-07-17", "CREATED", 19.9, 20250717195310),
]

schema = ["order_id", "dt", "status", "amount", "ts"]

def write_table(path, table_name, table_type, rows, mode):
    df = spark.createDataFrame(rows, schema)
    writer = (df.write.format("hudi")
        .option("hoodie.table.name", table_name)
        .option("hoodie.datasource.write.table.type", table_type)
        .option("hoodie.datasource.write.recordkey.field", "order_id")
        .option("hoodie.datasource.write.partitionpath.field", "dt")
        .option("hoodie.datasource.write.precombine.field", "ts")
        .option("hoodie.datasource.write.operation", "upsert")
        .option("hoodie.metadata.enable", "true"))

    if table_type == "MERGE_ON_READ":
        writer = (writer
            .option("hoodie.compact.inline", "false")
            .option("hoodie.compact.schedule.inline", "false"))

    writer.mode(mode).save(path)

for path, table, table_type in [
    (cow_path, "demo_cow_query_orders", "COPY_ON_WRITE"),
    (mor_path, "demo_mor_query_orders", "MERGE_ON_READ"),
]:
    write_table(path, table, table_type, rows1, "overwrite")
    write_table(path, table, table_type, rows2, "append")

print("=== COW snapshot ===")
(spark.read.format("hudi")
    .load(cow_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

print("=== MOR snapshot ===")
(spark.read.format("hudi")
    .load(mor_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

print("=== MOR read optimized ===")
(spark.read.format("hudi")
    .option("hoodie.datasource.query.type", "read_optimized")
    .load(mor_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

print("=== COW commits ===")
commits = ["20250717195456598", "20250717195510091"]
for commit in commits:
    print(commit)

begin_instant = "20250717195456599"
print(f"=== incremental begin {begin_instant} ===")
(spark.read.format("hudi")
    .option("hoodie.datasource.query.type", "incremental")
    .option("hoodie.datasource.read.begin.instanttime", begin_instant)
    .load(cow_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

spark.stop()
PY

执行命令：

source /etc/profile.d/bigdata_env.sh

hdfs dfs -rm -r -f \
  /hudi/demo_cow_query_orders \
  /hudi/demo_mor_query_orders

spark-submit \
  --master local[2] \
  --conf spark.sql.catalogImplementation=in-memory \
  --jars /opt/module/hudi/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  /tmp/hudi_table_query_demo.py

后续 hdfs dfs 命令中的 /hudi/demo_cow_query_orders 与 Spark 代码中的 hdfs://localhost:9000/hudi/demo_cow_query_orders 指向同一张 Hudi 表，只是省略了默认文件系统前缀；MOR 表路径同理。本文示例使用 --jars 显式指定 Hudi Spark bundle；如果已经在 Spark jars 目录中放入同版本 bundle，可以省略 --jars，但不要同时混入多个 Hudi 版本的 bundle。

这里关闭 inline compaction 的两个参数，是本文 Hudi 1.0.2 Spark DataSource 环境中的实测配置；不同版本参数名可能变化，落地时要回到当前版本配置文档核对。

2.2 Snapshot Query 验证

脚本中对应的读取代码是：

(spark.read.format("hudi")
    .load(cow_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

(spark.read.format("hudi")
    .load(mor_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

COW Snapshot Query 输出如下：

=== COW snapshot ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250717195510091  |2001    |2025-07-17|PAID   |120.5 |20250717195300|
|20250717195456598  |2002    |2025-07-17|CREATED|35.0  |20250717195010|
|20250717195456598  |2003    |2025-07-17|CREATED|78.8  |20250717195020|
|20250717195510091  |2004    |2025-07-17|CREATED|19.9  |20250717195310|
+-------------------+--------+----------+-------+------+--------------+

MOR Snapshot Query 输出如下：

=== MOR snapshot ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250717195519248  |2001    |2025-07-17|PAID   |120.5 |20250717195300|
|20250717195512924  |2002    |2025-07-17|CREATED|35.0  |20250717195010|
|20250717195512924  |2003    |2025-07-17|CREATED|78.8  |20250717195020|
|20250717195519248  |2004    |2025-07-17|CREATED|19.9  |20250717195310|
+-------------------+--------+----------+-------+------+--------------+

两张表都能看到最新状态：order_id=2001 已经从 CREATED 变成 PAID，order_id=2004 已经插入。

2.3 HDFS 文件结构验证

先看 COW 表数据文件：

hdfs dfs -ls /hudi/demo_cow_query_orders/2025-07-17

真实输出：

Found 3 items
-rw-r--r--   1 root supergroup         96 2025-07-17 19:55 /hudi/demo_cow_query_orders/2025-07-17/.hoodie_partition_metadata
-rw-r--r--   1 root supergroup     435642 2025-07-17 19:55 /hudi/demo_cow_query_orders/2025-07-17/4a275f7a-31ff-49f5-9acd-046900a64dc1-0_0-31-32_20250717195456598.parquet
-rw-r--r--   1 root supergroup     435626 2025-07-17 19:55 /hudi/demo_cow_query_orders/2025-07-17/4a275f7a-31ff-49f5-9acd-046900a64dc1-0_0-61-73_20250717195510091.parquet

同一个 fileId 前缀出现了两个 Parquet 文件版本，这就是 COW 更新时重写 Base File 的直观结果。这里判断同一个 FileGroup，主要看文件名前半段的 fileId；后面的 0_0-31-32、0_0-61-73 更接近写入任务相关的 write token，最后的 20250717195456598、20250717195510091 是 commit time，不要把整个文件名都理解成 fileId。

再看 MOR 表：

hdfs dfs -ls /hudi/demo_mor_query_orders/2025-07-17

真实输出：

Found 3 items
-rw-r--r--   1 root supergroup         96 2025-07-17 19:55 /hudi/demo_mor_query_orders/2025-07-17/.hoodie_partition_metadata
-rw-r--r--   1 root supergroup     435650 2025-07-17 19:55 /hudi/demo_mor_query_orders/2025-07-17/ca7c069c-4d62-45d4-900c-daa5d736c815-0_0-108-129_20250717195512924.parquet
-rw-r--r--   1 root supergroup     435629 2025-07-17 19:55 /hudi/demo_mor_query_orders/2025-07-17/ca7c069c-4d62-45d4-900c-daa5d736c815-0_0-138-170_20250717195519248.parquet

这里要如实说明：这个小规模 Spark DataSource 示例里，MOR 表目录没有明显出现业务数据 Log File，而是看到了两个 Parquet 文件版本。但它的表类型确实是 MOR，Timeline 上使用的是 deltacommit。

查看表属性：

hdfs dfs -cat /hudi/demo_mor_query_orders/.hoodie/hoodie.properties \
  | egrep "hoodie.table.name|hoodie.table.type|hoodie.timeline.path"

输出：

hoodie.table.type=MERGE_ON_READ
hoodie.table.name=demo_mor_query_orders
hoodie.timeline.path=timeline

再看 MOR Timeline：

hdfs dfs -ls /hudi/demo_mor_query_orders/.hoodie/timeline

输出：

Found 7 items
-rw-r--r--   1 root supergroup       3430 2025-07-17 19:55 /hudi/demo_mor_query_orders/.hoodie/timeline/20250717195512924.deltacommit.inflight
-rw-r--r--   1 root supergroup          0 2025-07-17 19:55 /hudi/demo_mor_query_orders/.hoodie/timeline/20250717195512924.deltacommit.requested
-rw-r--r--   1 root supergroup       3976 2025-07-17 19:55 /hudi/demo_mor_query_orders/.hoodie/timeline/20250717195512924_20250717195519005.deltacommit
-rw-r--r--   1 root supergroup       3534 2025-07-17 19:55 /hudi/demo_mor_query_orders/.hoodie/timeline/20250717195519248.deltacommit.inflight
-rw-r--r--   1 root supergroup          0 2025-07-17 19:55 /hudi/demo_mor_query_orders/.hoodie/timeline/20250717195519248.deltacommit.requested
-rw-r--r--   1 root supergroup       4065 2025-07-17 19:55 /hudi/demo_mor_query_orders/.hoodie/timeline/20250717195519248_20250717195521809.deltacommit
drwxr-xr-x   - root supergroup          0 2025-07-17 19:55 /hudi/demo_mor_query_orders/.hoodie/timeline/history

这里是 Hudi 1.0.x Timeline 布局下的实测输出，完成态 deltacommit 文件名包含 completion time；requested / inflight 文件用于表达 Instant 状态流转，排查时以完成态文件作为已提交动作的主要依据。

所以判断 COW / MOR 不能只看数据目录里是不是立刻出现 .log 文件，还要看表属性、Timeline action、写入配置和后续 compaction 行为。

2.4 Read Optimized Query 验证

Spark DataSource 中可以这样指定 Read Optimized Query：

(spark.read.format("hudi")
    .option("hoodie.datasource.query.type", "read_optimized")
    .load(mor_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

本次输出如下：

=== MOR read optimized ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250717195519248  |2001    |2025-07-17|PAID   |120.5 |20250717195300|
|20250717195512924  |2002    |2025-07-17|CREATED|35.0  |20250717195010|
|20250717195512924  |2003    |2025-07-17|CREATED|78.8  |20250717195020|
|20250717195519248  |2004    |2025-07-17|CREATED|19.9  |20250717195310|
+-------------------+--------+----------+-------+------+--------------+

这次 Read Optimized 和 Snapshot 输出一致，原因是这个小表的最新数据已经体现在 Base File 里，没有形成明显的“Base File + 未压缩 Log File”差异。

这个结果只说明当前小样本中最新数据已经进入 Base File，不能推出 Read Optimized Query 总是和 Snapshot Query 一样新。

在真实 MOR 流式写入中，如果更新大量追加到 Log File 且尚未 compaction，Snapshot Query 和 Read Optimized Query 就可能看到不同新鲜度的数据。

2.5 Incremental Query 验证

COW 表有两个 commit：

=== COW commits ===
20250717195456598
20250717195510091

使用 Incremental Query：

(spark.read.format("hudi")
    .option("hoodie.datasource.query.type", "incremental")
    .option("hoodie.datasource.read.begin.instanttime", "20250717195456599")
    .load(cow_path)
    .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
    .orderBy("order_id")
    .show(truncate=False))

本文使用的是 Hudi 1.0.2 Spark DataSource 实测参数；Hudi 1.0.x 中 begin instant 与 completion time 语义需要以当前版本文档为准。为了避免包含第一个 commit，这里使用 20250717195456599，而不是直接使用 20250717195456598。

这里把 begin.instanttime 设置为第一个 commit 之后的时间，是为了只读取第二个 commit 之后的变化；如果直接使用第一个 commit time，具体是否包含该 commit 需要以当前版本的增量查询语义为准。

实际输出：

=== incremental begin 20250717195456599 ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250717195510091  |2001    |2025-07-17|PAID   |120.5 |20250717195300|
|20250717195456598  |2002    |2025-07-17|CREATED|35.0  |20250717195010|
|20250717195456598  |2003    |2025-07-17|CREATED|78.8  |20250717195020|
|20250717195510091  |2004    |2025-07-17|CREATED|19.9  |20250717195310|
+-------------------+--------+----------+-------+------+--------------+

这个结果很值得注意：虽然业务上第二次写入只更新了 2001、新增了 2004，但增量查询里也看到了 2002 和 2003。原因是 COW 更新会重写 Base File，增量读看到的是提交影响到的文件视图，不应该简单理解成数据库 binlog 那种精确行级变更。在这个例子里，2002 和 2003 的 _hoodie_commit_time 仍然是第一次提交时间，但它们被包含在第二次重写后的文件视图中，所以出现在增量读取结果里。下游应按 order_id 做 merge / upsert，而不是把每一行都当成新的业务变更事件。

所以增量 Pipeline 里通常要保证下游幂等：

1. 用 record key 做覆盖或 merge。
2. 用 _hoodie_commit_time 判断处理进度。
3. 用业务 precombine.field 解决同 key 多版本顺序。
4. 不要假设 Incremental Query 一定只包含业务变化行。

3. 常见问题与选型

3.1 常见问题

3.1.1 COW 表为什么有多个 Parquet 文件版本

因为 COW 更新时会写出新的 Base File，旧 Base File 不会立刻删除。Hudi 会通过 Timeline 和 FileSlice 判断哪个版本对当前查询可见，再通过 clean 策略清理过旧文件。

3.1.2 MOR 表为什么我没看到 Log File

MOR 的设计支持 Base File + Log File，但小规模 Spark DataSource 示例、写入配置、small file 策略、是否触发 compaction 等因素都会影响实际文件形态。

INSERT、UPSERT、是否命中已有 FileGroup、small file 是否需要继续填充，都会影响最终是写 Base File 还是 Log File。

所以判断 MOR 不要只看数据目录有没有 .log 文件，还要看：

hdfs dfs -cat /path/to/table/.hoodie/hoodie.properties
hdfs dfs -ls /path/to/table/.hoodie/timeline

如果 hoodie.table.type=MERGE_ON_READ，Timeline 中出现 deltacommit，说明表语义已经是 MOR。

3.1.3 Snapshot Query 和 Read Optimized Query 为什么结果一样

如果 MOR 表中的最新数据已经在 Base File 中，或者当前测试没有形成未压缩 Log File，那么二者结果可能一样。

二者真正拉开差异的场景通常是：Log File 中有尚未 compaction 的新变更。Snapshot Query 会合并读取，Read Optimized Query 只看压缩后的 Base File。

3.1.4 Incremental Query 是不是等于 CDC 明细

不是。Incremental Query 是 Hudi 按 Timeline 提供的增量读取能力，不等同于数据库 binlog 的逐行变更明细。

在 COW 表里，更新会导致 Base File 重写，增量结果可能包含同一个文件组里的其他记录。下游应该按 Hudi record key / 业务主键做幂等 merge 或覆盖处理。

3.1.5 COW 和 MOR 能不能随便切换

表类型属于表级设计，不建议在同一张表生命周期中随便切换。尤其是已经被 Hive Sync、Flink 作业、下游查询依赖以后，文件视图和查询语义变化会影响很多链路。

不要简单通过修改 hoodie.properties 中的 hoodie.table.type 来切换表类型。更稳妥的做法是：先根据业务场景选型；如果确实需要切换，应建新表迁移并完整验证读写链路。

3.2 表类型与查询类型选型

COW 和 MOR 的关键，不是“哪个更高级”，而是“成本放在哪里”。

COW 把成本放在写入时。每次更新尽量把文件整理好，查询时就轻松很多。这种表更适合传统离线数仓和 BI 查询，因为下游大多数时候只关心稳定读取。

MOR 把成本拆开。写入时先把变更快速落下来，读取最新状态时再合并，后台通过 compaction 慢慢整理文件。这种表更适合实时入湖，但它要求你认真管理 compaction、Log File、查询新鲜度和资源隔离。

三种查询方式也不要混着用：

1. Snapshot Query 面向“我要看这张表当前最新状态”。
2. Incremental Query 面向“我要接着上次 commit 往后消费”。
3. Read Optimized Query 面向“我愿意牺牲一点新鲜度，换更简单的文件读取”。

如果把这几个概念混在一起，很容易出现错误判断：以为 MOR 一定查询慢、以为 Incremental Query 一定是 CDC 明细、以为 Read Optimized 一定看不到最近写入的数据。实际是否能看到，要看最近写入是否已经进入 Base File 或已经完成 compaction。

4. 总结

这篇主要讲了 Hudi 的两种表类型和三种查询方式。

1. COW 表更新时重写 Base File，查询简单，但写放大明显。
2. MOR 表可以先追加 Log File，写入更轻，但查询和 compaction 更复杂。
3. MOR 是否出现 Log File，受写入引擎、索引能力、small file 策略和 compaction 状态影响。
4. Snapshot Query 查询最新表状态。
5. Incremental Query 按 commit 范围读取，适合增量 Pipeline，但不等同于精确 CDC 明细。
6. Read Optimized Query 主要面向 MOR 表，只读当前可用于读优化视图的 Base / Columnar 文件，适合低成本查询。
7. COW / MOR 是表类型，Snapshot / Incremental / Read Optimized 是查询类型，两组概念不能混淆。
8. 真实环境里要以表属性、Timeline、文件目录和查询输出一起判断，不要只凭某个目录现象下结论。