huyuanfeng2018 opened a new issue, #10708: URL: https://github.com/apache/paimon/issues/10708
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. ### Motivation <html><head></head><body><h2><span>Title</span></h2><p><span>[Feature] Support cumulative hourly deltas over daily snapshot baselines in Chain Tables</span></p><h2><span>Motivation</span></h2><p><span>In our batch data warehouse pipelines, we maintain daily full-state tables (</span><code><span>_d</span></code><span>) and hourly full-state tables (</span><code><span>_h</span></code><span>).</span></p><p><span>Each hourly job currently:</span></p><ol start="1"><li><p><span>Reads the previous day's full dataset.</span></p></li><li><p><span>Reads the cumulative changes from the start of the current day through the end of the target hour.</span></p></li><li><p><span>Performs a full outer join and resolves updates by business key.</span></p></li><li><p><span>Writes the resulting full dataset into the target hourly partition.</span></p></li></ol><p><span>For example, a job running after </span><code><span>2026-10-20 02:00</span></code><span> processes changes in </span><code><span>[2026 -10-20 00:00, 2026-10-20 02:00)</span></code><span> and produces hourly partition </span><code><span>2026102001</span></code><span>.</span></p><p><span>Most rows remain unchanged between hourly results, so this pattern repeatedly reads, merges, and stores a large amount of unchanged data.</span></p><p><span>Chain Tables already support hourly partitioning and reconstruction from a snapshot and subsequent delta partitions. We would like to extend this model to support </span><strong><span>daily snapshot baselines and independently reconstructable hourly cumulative deltas</span></strong><span>, matching the output of our existing batch pipelines.</span></p><p><span>The desired reconstruction is:</span></p><pre><code><span>FullState(target_hour) = Merge(previous_day_snapshot, cumulative_delta(target_hour))</span></code></pre><p><span>Earlier hourly delta partitions would not be inputs to this reconstruction.</span></p><h2><span>Solution</span></h2><h3><span>1. Daily snapshots and hourly cumulative deltas</span></h3><p><span>Use the snapshot branch for daily full-state partitions and the delta branch for hourly cumulative change sets.</span></p><p><span>The partition naming conventions would be:</span></p><ul><li><p><strong><span>Daily snapshot:</span></strong><span> </span><code><span>yyyyMMdd</span></code><span> identifies the day whose end-of-day full state is stored.</span></p></li><li><p><strong><span>Hourly delta:</span></strong><span> </span><code><span>yyyyMMddHH</span></code><span> identifies the hour by its start time. Its contents include the cumulative changes from the start of that day through the end of that hour.</span></p></li></ul><p><span>For example:</span></p> Branch | Partition | Contents -- | -- | -- snapshot | 20261019 | Full state at the end of October 19, serving as the baseline at October 20, 00:00 delta | 2026102000 | Cumulative changes during October 20, [00:00, 01:00) delta | 2026102001 | Cumulative changes during October 20, [00:00, 02:00) delta | 2026102006 | Cumulative changes during October 20, [00:00, 07:00) delta | 2026102023 | Cumulative changes during October 20, [00:00, October 21 00:00) snapshot | 20261020 | Full state at the end of October 20, serving as the next day's baseline <p><span>All boundaries would use the same configured business time zone.</span></p><p><span>Querying logical hourly partition </span><code><span>2026102006</span></code><span> should therefore reconstruct:</span></p><pre><code><span>FullState(2026102006) = Merge( Snapshot(20261019), CumulativeDelta(2026102006) )</span></code></pre><p><span>The query should read the daily baseline and the target delta partition, without reading hourly delta partitions </span><code><span>2026102000</span></code><span> through </span><code><span>2026102005</span></code><span>.</span></p><h3><span>2. Cumulative delta contract</span></h3><p><span>Each hourly cumulative delta contains the final complete row image for every business key affected since the start of the day.</span></p><p><span>For example, if a key changes during </span><code><span>00:00–01:00</span></code><span> and does not change again, its updated row must still appear in </span><code><span>CumulativeDelta(2026102006)</span></code><span>.</span></p><p><span>Rows unchanged since the daily baseline do not need to be repeated.</span></p><p><span>This gives the following semantics:</span></p><ul><li><p><span>A key present in the cumulative delta contributes its final state according to the configured key and sequence semantics.</span></p ></li><li><p><span>A key absent from the cumulative delta retains its baseline >state.</span></p></li><li><p><span>If deletion support is included, it >requires explicit deletion semantics; absence from the cumulative delta must >not implicitly mean deletion.</span></p></li></ul><p><span>An initial scope >could focus on </span><strong><span>batch reads with deduplicate semantics >and complete row images</span></strong><span>.</span></p><h3><span>3. >Explicit mapping between hourly targets and daily >baselines</span></h3><p><span>For the example above, the mapping >is:</span></p><pre><code><span>target delta partition = 2026102006 baseline partition = 20261019</span></code></pre><p><span>This mapping should follow the declared baseline relationship, rather than selecting a snapshot through raw partition-value comparison.</span></p><p><span>In particular, </span><code><span>snapshot/20261020</span></code><span> represents the state at October 21, 00:00. Even after it becomes available, it must not be used to reconstruct an earlier hourly result such as </span><code><span>2026102006</span></code><span>.</span></p><p><span>A possible implementation could retain the same partition schema on both branches, such as a single </span><code><span>pt STRING</span></code><span> partition column, while interpreting snapshot values as daily partitions and delta values as hourly partitions. The initial implementation would not need different partition-column layouts between branches.</span></p><h3><span>4. Proposed integration with Chain Tables</span></h3><p><span>This could be introduced as an opt-in cumulative-delta mode, preserving the existing behavior by default.</span></p><p><span>The read planner would:</span></p><ol start="1"><li><p><span>Resolve the requested logical hourly partition.</span></p></li><li><p><span>Locate its declared daily baseline.</span></p></li><li><p><span>Select only the corresponding cumulative delta partition.</span></p></li><li><p><span>Reuse the existing Chain Table merge reader to produce the full result under the requested logical hourly partition.</span></p></li></ol><p><span>The current </span><code><span>ChainGroupReadTable.ChainTableBatchScan</span></code><span> seems a relevant place to investigate the planning changes. Baseline resolution, partition-time interpretation, compaction, and expiration would also need to follow the new semantics.</span></p><h3><span>5. Correctness and lifecycle considerations</span></h3><p><span>The following behaviors would need to be defined:</span></p><ul><li><p><strong><span>Empty results:</span></strong><span> distinguish a successfully published empty cumulative delta from a partition that has not been produced successfully.</span></p></li><li><p><strong><span>Baseline availability:</span></strong><span> a missing required baseline must not silently fall back to an unrelated daily snapshot.</span></p></li><li><p><strong><span>Reruns:</span></strong><span> define how an hourly overwrite or a daily baseline correction affects published results and any materialized full-state partitions.</span></p></li><li><p><strong><span>Retention:</span></strong><span> retain a daily baseline while any retained hourly version depends on it. Earlier hourly deltas should be independently removable without changing later hourly results.</span></p></li><li><p><strong><span>Compaction:</span></strong><span> preserve the distinction between daily baselines and any optional materialized hourly results.</span></p></li><li><p><strong><span>Read-path compatibility:</span></strong><span> define the behavior of streaming and lo okup reads before enabling them for cumulative deltas.</span></p></li></ul><h2><span>Anything else?</span></h2><p><span>The expected benefits are to reduce repeated full-data writes and avoid scanning redundant earlier cumulative delta partitions when reconstructing a target hour.</span></p><p><span>There are also trade-offs: cumulative deltas repeat some changed rows across hours, and reconstructing a full hourly result still requires reading the relevant baseline data. The storage and read-cost benefits should be evaluated against both hourly full copies and ordinary interval-delta Chain Tables.</span></p><p><span>I would appreciate community feedback on:</span></p><ol start="1"><li><p><span>Whether this use case fits an extension of the existing Chain Table model.</span></p></li><li><p><span>Whether baseline selection should use an explicit baseline reference, a configurable period-based rule, or another existing mechanism.</span></p></li><li><p><span>What would be an appropriate initial scope for supporting daily snapshots with hourly cumulative deltas.</span></p></li></ol><p><span>Related references:</span></p><ul><li><p><a href="https://cwiki.apache.org/confluence/display/PAIMON/PIP-37%3A+Introduce+Chain+Table";><span>PIP-37: Introduce Chain Table</span></a></p></li><li><p><a href="https://paimon.apache.org/docs/master/primary-key-table/chain-table/";><span>Chain Table documentation</span></a></p></li><li><p><a href="https://github.com/apache/paimon/blob/8c09824c060520f3f1f4a03ce2369de371e5c7c6/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java#L363-L499";><span>ChainGroupReadTable planning implementation</span></a></p></li></ul></body></html>## Title [Feature] Support cumulative hourly deltas over daily snapshot baselines in Chain Tables ## Motivation In our batch data warehouse pipelines, we maintain daily full-state tables (`_d`) and hourly full-state tables (`_h`). Each hourly job currently: 1. Reads the previous day's full dataset. 2. Reads the cumulative changes from the start of the current day through the end of the target hour. 3. Performs a full outer join and resolves updates by business key. 4. Writes the resulting full dataset into the target hourly partition. For example, a job running after `2026-10-20 02:00` processes changes in `[2026-10-20 00:00, 2026-10-20 02:00)` and produces hourly partition `2026102001`. Most rows remain unchanged between hourly results, so this pattern repeatedly reads, merges, and stores a large amount of unchanged data. Chain Tables already support hourly partitioning and reconstruction from a snapshot and subsequent delta partitions. We would like to extend this model to support **daily snapshot baselines and independently reconstructable hourly cumulative deltas**, matching the output of our existing batch pipelines. The desired reconstruction is: ```text FullState(target_hour) = Merge(previous_day_snapshot, cumulative_delta(target_hour)) ``` Earlier hourly delta partitions would not be inputs to this reconstruction. ## Solution ### 1. Daily snapshots and hourly cumulative deltas Use the snapshot branch for daily full-state partitions and the delta branch for hourly cumulative change sets. The partition naming conventions would be: - **Daily snapshot:** `yyyyMMdd` identifies the day whose end-of-day full state is stored. - **Hourly delta:** `yyyyMMddHH` identifies the hour by its start time. Its contents include the cumulative changes from the start of that day through the end of that hour. For example: | Branch | Partition | Contents | | --- | --- | --- | | `snapshot` | `20261019` | Full state at the end of October 19, serving as the baseline at October 20, 00:00 | | `delta` | `2026102000` | Cumulative changes during October 20, `[00:00, 01:00)` | | `delta` | `2026102001` | Cumulative changes during October 20, `[00:00, 02:00)` | | `delta` | `2026102006` | Cumulative changes during October 20, `[00:00, 07:00)` | | `delta` | `2026102023` | Cumulative changes during October 20, `[00:00, October 21 00:00)` | | `snapshot` | `20261020` | Full state at the end of October 20, serving as the next day's baseline | All boundaries would use the same configured business time zone. Querying logical hourly partition `2026102006` should therefore reconstruct: ```text FullState(2026102006) = Merge( Snapshot(20261019), CumulativeDelta(2026102006) ) ``` The query should read the daily baseline and the target delta partition, without reading hourly delta partitions `2026102000` through `2026102005`. ### 2. Cumulative delta contract Each hourly cumulative delta contains the final complete row image for every business key affected since the start of the day. For example, if a key changes during `00:00–01:00` and does not change again, its updated row must still appear in `CumulativeDelta(2026102006)`. Rows unchanged since the daily baseline do not need to be repeated. This gives the following semantics: - A key present in the cumulative delta contributes its final state according to the configured key and sequence semantics. - A key absent from the cumulative delta retains its baseline state. - If deletion support is included, it requires explicit deletion semantics; absence from the cumulative delta must not implicitly mean deletion. An initial scope could focus on **batch reads with deduplicate semantics and complete row images**. ### 3. Explicit mapping between hourly targets and daily baselines For the example above, the mapping is: ```text target delta partition = 2026102006 baseline partition = 20261019 ``` This mapping should follow the declared baseline relationship, rather than selecting a snapshot through raw partition-value comparison. In particular, `snapshot/20261020` represents the state at October 21, 00:00. Even after it becomes available, it must not be used to reconstruct an earlier hourly result such as `2026102006`. A possible implementation could retain the same partition schema on both branches, such as a single `pt STRING` partition column, while interpreting snapshot values as daily partitions and delta values as hourly partitions. The initial implementation would not need different partition-column layouts between branches. ### 4. Proposed integration with Chain Tables This could be introduced as an opt-in cumulative-delta mode, preserving the existing behavior by default. The read planner would: 1. Resolve the requested logical hourly partition. 2. Locate its declared daily baseline. 3. Select only the corresponding cumulative delta partition. 4. Reuse the existing Chain Table merge reader to produce the full result under the requested logical hourly partition. The current `ChainGroupReadTable.ChainTableBatchScan` seems a relevant place to investigate the planning changes. Baseline resolution, partition-time interpretation, compaction, and expiration would also need to follow the new semantics. ### 5. Correctness and lifecycle considerations The following behaviors would need to be defined: - **Empty results:** distinguish a successfully published empty cumulative delta from a partition that has not been produced successfully. - **Baseline availability:** a missing required baseline must not silently fall back to an unrelated daily snapshot. - **Reruns:** define how an hourly overwrite or a daily baseline correction affects published results and any materialized full-state partitions. - **Retention:** retain a daily baseline while any retained hourly version depends on it. Earlier hourly deltas should be independently removable without changing later hourly results. - **Compaction:** preserve the distinction between daily baselines and any optional materialized hourly results. - **Read-path compatibility:** define the behavior of streaming and lookup reads before enabling them for cumulative deltas. ## Anything else? The expected benefits are to reduce repeated full-data writes and avoid scanning redundant earlier cumulative delta partitions when reconstructing a target hour. There are also trade-offs: cumulative deltas repeat some changed rows across hours, and reconstructing a full hourly result still requires reading the relevant baseline data. The storage and read-cost benefits should be evaluated against both hourly full copies and ordinary interval-delta Chain Tables. I would appreciate community feedback on: 1. Whether this use case fits an extension of the existing Chain Table model. 2. Whether baseline selection should use an explicit baseline reference, a configurable period-based rule, or another existing mechanism. 3. What would be an appropriate initial scope for supporting daily snapshots with hourly cumulative deltas. Related references: - [[PIP-37: Introduce Chain Table](https://cwiki.apache.org/confluence/display/PAIMON/PIP-37%3A+Introduce+Chain+Table)](https://cwiki.apache.org/confluence/display/PAIMON/PIP-37%3A+Introduce+Chain+Table) - [[Chain Table documentation](https://paimon.apache.org/docs/master/primary-key-table/chain-table/)](https://paimon.apache.org/docs/master/primary-key-table/chain-table/) - [[ChainGroupReadTable planning implementation](https://github.com/apache/paimon/blob/8c09824c060520f3f1f4a03ce2369de371e5c7c6/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java#L363-L499)](https://github.com/apache/paimon/blob/8c09824c060520f3f1f4a03ce2369de371e5c7c6/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java#L363-L499) ### Solution _No response_ ### Anything else? _No response_ ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
