Hudi COW vs MOR: Incremental Queries Need Compaction, Not Another Format

The incremental job that usually read 40 GB started scanning 2.1 TB. The Hudi table was merge-on-read. Compaction had been “temporarily” disabled in March. By August the timeline was a pile of log files on top of base Parquet, and hoodie.datasource.query.type=incremental still had to merge them. We thought the source had exploded. The source was fine. We had stopped operating the table.

Hudi is the format I reach for when Spark is the writer and the job is CDC upserts into S3 with a consumer that wants “everything since commit X.” It is not a third lake “just in case.” Iceberg and Delta already cover most lakes — I compared those two in Iceberg vs Delta. This post is the Hudi-shaped choice: COW, MOR, incremental, compaction.

COW versus MOR

Copy-on-write updates a file by writing a new Parquet file and pointing the timeline at it. Readers see a consistent snapshot of base files. Upsert cost tracks how many files the batch touches. Good for read-heavy tables and modest change rates.

Merge-on-read writes row updates into log files next to base files. Ingest stays cheap. Every query merges logs unless you have compacted. Good for high-churn CDC. Bad if nobody runs compaction.

flowchart TD
  cdc["CDC batch"] --> cow["COW: rewrite Parquet"]
  cdc --> mor["MOR: append log files"]
  cow --> snap["Snapshot readers"]
  mor --> merge["Readers merge base + logs"]
  mor --> compact["Compaction job"]
  compact --> snap2["New base files"]
  snap2 --> snap

COW pays at write. MOR pays at read until compaction moves the debt.

# Spark: MOR table, incremental consume after last checkpoint
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
last = spark.read.json("s3://ops/hudi/orders_ckpt.json").first().commit_time

inc = (
    spark.read.format("hudi")
    .option("hoodie.datasource.query.type", "incremental")
    .option("hoodie.datasource.read.begin.instanttime", last)
    .load("s3://lake/hudi/orders")
)
# Snapshot of the same table is a different query:
#   .option("hoodie.datasource.query.type", "snapshot")
inc.createOrReplaceTempView("orders_inc")

If you omit the begin instant, you are not incremental. You are a full snapshot with extra ceremony. I store the last successful commit in the same place I store a dbt watermark. Losing it means a full read, same class of bug as a missing incremental filter.

Compaction is a pipeline

MOR without compaction is a time bomb with a long fuse. I schedule compaction on the write cluster, not “when queries feel slow.” Cleaning old files is a second job. Clustering is a third if the file layout skews. Skipping any one looks like “S3 got slow.”

-- Spark SQL against a Hudi table registered in the catalog
-- Compaction is usually a Spark job, not a casual SELECT.
-- Runbook check I want in Grafana:
--   file count, log-to-base ratio, oldest uncompacted instant

SELECT
    count(*) AS files,
    sum(CASE WHEN path LIKE '%.log%' THEN 1 ELSE 0 END) AS logish
FROM hive.lake.orders_file_listing;

That listing query is a stand-in. In practice I watch Hudi metrics: commit duration, records upserted, and pending compactions. When pending compactions grow for three days, I page before the 2 TB scan comes back.

One writer, one format

I have seen a team write Hudi from Flink and Iceberg from Spark “so we can migrate later.” Two timelines, two compactors, one bucket of Parquet-looking files, and a week of arguing which snapshot was truth. Pick one format per table. If you need Iceberg readers, use Iceberg, or a documented read bridge — not two writers.

# Writer config I pin in the ingest job. One table type per path.
hudi_options = {
    "hoodie.table.name": "orders",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.precombine.field": "updated_at",
    "hoodie.datasource.write.table.type": "MERGE_ON_READ",
    "hoodie.compact.inline": "false",  # compaction is a separate DAG
    "hoodie.cleaner.policy": "KEEP_LATEST_COMMITS",
    "hoodie.cleaner.commits.retained": "20",
}

(
    batch.write.format("hudi")
    .options(**hudi_options)
    .mode("append")
    .save("s3://lake/hudi/orders")
)

Inline compaction looks convenient and then collides with the ingest SLA. I run compaction on a schedule after the CDC window, with a lock so two clusters cannot compact the same table. Cleaning is not optional: 20 commits retained is a starting point, not a forever archive. Time travel needs a number you chose.

Pitfalls

MOR plus no compaction for a quarter. Query time grows like interest. The 2.1 TB scan was five months of logs. Turn compaction back on before you rewrite the consumer.

Incremental without a stored commit. A “start from now” after a deploy drops a day. Persist end_instant next to the job state the way you persist a dbt watermark.

COW on a 40 million row CDC hour. You will rewrite the same hot files all morning. Switch to MOR or buffer and compact in batch. COW is not wrong; it is wrong at that change rate.

Collecting Hudi, Iceberg, and Delta on one domain. Three runbooks, three “why is this table slow” paths. One writer, one format. Migrate with a planned cut, not a dual-write weekend.

Treating Hudi as Kafka. It is a table format with an incremental pull. You still need a stream or a Spark job to land events. Hudi does not replace the log.

When do I still choose Hudi in 2026? Spark-first CDC, incremental downstream jobs that are already Hudi-native, and an ops team that will run compaction. When do I choose Iceberg? Trino, Snowflake, and Flink all need to read, and maybe write, the same lake. Delta when Databricks is the write engine. The feature lists overlap. The maintenance clock does not. We did not need a new format. We needed the compaction DAG to stay enabled.