AntiTopQuark opened a new issue, #68225: URL: https://github.com/apache/doris/issues/68225
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/doris/issues?q=is%3Aissue) and found no similar issues. ### Description ### Motivation ROW Binlog records table changes for incremental queries, table streams, and incremental materialized view maintenance. These consumers usually need a configurable window of recent changes rather than an indefinitely retained history. This issue proposes an end-to-end retention policy for ROW Binlog, covering query visibility, background discovery of expired data, compaction, and eventual file reclamation in both shared-nothing and cloud deployments. Retention should continue to advance when a table receives no new writes. It should also work for cloud tablets that are absent from BE metadata caches, and should use a common time source rather than depend on each BE's local clock. This feature applies to ROW Binlog records. Ordinary row TTL, which controls the visibility and reclamation of business rows through a hidden TTL column, is a separate feature. Expiring a Binlog record does not delete its source-table row or generate a business DELETE event. ### Scope The design separates expiration into three stages: 1. New incremental queries exclude records outside the retention window. 2. Background compaction removes fully expired Rowset data from the current version while preserving version continuity. 3. Existing stale-Rowset and recycler mechanisms reclaim the old files asynchronously. Logical expiration therefore does not imply immediate disk or object-storage reclamation. This proposal does not provide a hard storage-size bound or a consumer-acknowledged retention guarantee. ### Configuration Use the existing `binlog.ttl_seconds` property for tables with ROW Binlog enabled: ```sql CREATE TABLE event_log ( event_id BIGINT, event_time DATETIME, payload STRING ) DUPLICATE KEY(event_id) DISTRIBUTED BY HASH(event_id) BUCKETS 8 PROPERTIES ( "binlog.enable" = "true", "binlog.format" = "ROW", "binlog.ttl_seconds" = "86400" ); ALTER TABLE event_log SET ( "binlog.ttl_seconds" = "259200" ); ``` | Configuration | ROW Binlog behavior | |---|---| | Neither database nor table explicitly specifies TTL | Default to 86400 seconds, or one day | | Positive value | Retain changes according to the specified duration | | Zero or negative value | Reject the explicit ROW Binlog TTL configuration | | Database-level configuration | Inherit when creating a table; explicit table properties take precedence | | ALTER DATABASE | Affect future inheritance, without rewriting existing tables' effective TTL | | ALTER TABLE | Update the existing table's retention policy | `binlog.ttl_seconds` is the only retention-duration setting. There is no separate `row_ttl_enabled` flag or persisted effective-TTL field. ROW TTL is derived from `enable && format == ROW && ttl_seconds > 0`. The initial implementation does not support independently disabling ROW Binlog TTL, and does not change the existing restriction on disabling an enabled ROW Binlog through ALTER. Non-ROW CCR behavior remains unchanged. Increasing TTL can make still-retained records visible again, but cannot restore records already removed by compaction. ### Time model and query semantics Use commit TSO as the time basis. LSN remains an ordering identifier and is not interpreted as a timestamp. A TSO contains physical milliseconds and an 18-bit logical counter. For a positive TTL, compute an inclusive expiration cutoff as follows: ```text reference_ms = physical_time(reference_tso) if ttl_seconds > reference_ms / 1000: cutoff_tso = 0 else: cutoff_tso = ((reference_ms - ttl_seconds * 1000) << 18) | ((1 << 18) - 1) first_retained_tso = nextTso(cutoff_tso) ``` The cutoff includes every logical timestamp in the expiration millisecond. FE additionally bounds the cutoff by `MAX_REAL_TSO` so its successor remains representable. Each statement that needs TTL-aware Binlog reads obtains one reference TSO from the Master FE before acquiring internal table locks: - A Master FE calls `TSOService.getTSO()` locally. - A Follower or Observer calls `FrontendService.getCurrentTso()` using the statement execution timeout. - The statement caches the reference in `StatementContext`; each relation derives its cutoff from that reference and its own TTL. - Allocation or RPC failures fail planning rather than falling back to an FE/BE local clock. `MasterTsoProvider` does not add automatic redirection or RPC retries. The Master TSO service provides the successful-allocation contract. The RPC result carries a status and an optional TSO, and the client checks the status before using the successful result. Scans preserve the existing half-open interval `[start_tso, end_tso)`: | Read mode | Retention behavior | |---|---| | DETAIL / APPEND_ONLY | Clamp the scan start to `max(user_start_tso, first_retained_tso)` | | No explicit scan start | Start at `first_retained_tso` | | MIN_DELTA with an explicit scan start | Reject a start earlier than `first_retained_tso` | The MIN_DELTA check operates on the scan lower bound after any existing Stream-offset conversion. A scan start equal to `cutoff_tso` is expired; a start equal to `nextTso(cutoff_tso)` is valid. The existing scan-range fields carry the bounds to BE, which applies: ```text __DORIS_BINLOG_TSO__ >= effective_start_tso AND __DORIS_BINLOG_TSO__ < end_tso ``` The upper predicate is applied only when an end bound is present. Filtering occurs before DETAIL expansion or MIN_DELTA merging. UPDATE before/after images share a physical change record and commit TSO, so this filtering does not split an UPDATE pair. TTL-aware reads cannot reuse SQL or Query Cache results across expiration boundaries. The query paths include `@incr` and Stream/IVM paths that actually read Binlog. The debug-only `binlog(...)` TVF does not inject the TTL lower bound and can inspect records awaiting physical reclamation. ### Background reference distribution and discovery Query-time TSO acquisition is independent of background cleanup. Queries do not wait for a background refresh or use the last BE heartbeat reference. ```mermaid flowchart TD M[Master TSO service] --> Q[One reference per query statement] Q --> F[TTL scan bounds and expired-offset checks] M --> H[HeartbeatMgr reference refresh] H --> B[BE ClusterInfo: monotonic in-memory reference] H --> D[Cloud catalog discovery] D --> L[Load uncached Tablet metadata and Rowsets] B --> S[Bounded BE TTL scanner] L --> S S --> C[Compaction with frozen TTL and cutoff] C --> V[Validate current TTL at publication or commit] V --> R[Version-covering output Rowset] R --> G[Existing delayed file reclamation] ``` `HeartbeatMgr` submits reference refresh work when at least 15 seconds have elapsed since the last submission and no refresh is pending. A dedicated worker obtains a Master TSO and publishes it in `TMasterInfo`. Existing BE heartbeats distribute it in both deployment modes. The 15-second value is a submission threshold, not a fixed cleanup period. Heartbeat scheduling, worker execution, discovery, and compaction introduce additional delay. Heartbeat sending does not wait for the refresh worker. BE accepts reference advancement only from the current accepted Master epoch and updates `ClusterInfo` monotonically. FE uses CAS to discard refresh results that cross a Master transition. Reference allocation does not require new transactions, Binlog tombstones, or a prior scan for eligible tables. **The background reference is process-level memory state. It is not persisted in TabletMeta, FE Journal/Image, or MS KV.** BE restart resets it to zero, and TTL GC waits for a fresh valid heartbeat. Configuration updates do not reset it. Discovery covers both active and idle tablets: - The BE scanner visits up to 64 registered Row Binlog tablets per round, with a one-second wait between rounds. Each tablet inspection examines at most 256 Rowset metadata entries. - Preparation uses a single worker, a queue limit of 64, and deduplication by tablet ID. Tablet registration uses weak references, and shutdown joins the scanner before stopping the preparation pool. - Expired tablets can trigger Binlog compaction even without new writes or a positive ordinary compaction score. - In cloud mode, `RowBinlogTtlDiscovery` traverses the FE catalog with a 64-tablet limit and a 50 ms traversal budget per invocation, preserving its cursor across rounds. - For each compute group, discovery targets the existing replica-owner BE. `sync_tablet_meta(..., discover_row_binlog_ttl=true)` queues work that can load an uncached tablet and refresh its configuration and Rowsets. Discovery honors the Binlog feature switch, global and tablet-level automatic-compaction switches, and cloud compaction ownership/standby rules. In discovery mode, `synced_tablets` reports successful preparation-task enqueueing, not completed synchronization or reclamation. Failed attempts can be revisited in later sweeps. ### Compaction and configuration-change correctness A non-empty Rowset is eligible for TTL removal only when: ```text commit_tso is present AND commit_tso.start_tso > 0 AND commit_tso.end_tso >= commit_tso.start_tso AND commit_tso.end_tso <= gc_cutoff_tso ``` Missing or incomplete TSO ranges do not authorize deletion. This stage removes fully expired Rowsets; it does not individually filter expired rows from a partially expired Rowset. Compaction freezes the cutoff and TTL during preparation. Shared-nothing compaction also freezes the confirmed visible version and excludes Rowsets beyond it from TTL deletion. Cloud compaction uses the existing committed-Rowset view and MS job-conflict handling. Keep version-covering inputs separate from data inputs. When every data input expires, emit an empty output Rowset covering the original version range. Reclamation then follows the existing stale/recycle lifecycle. Any compaction that actually removes TTL data must revalidate the retention policy before publishing its result: | Deployment | Commit check | |---|---| | Shared-nothing | Under the Tablet header write lock, verify that ROW TTL remains enabled and its duration matches the prepared duration | | Cloud | Send the prepared TTL in `TabletCompactionJobPB.row_binlog_ttl_seconds`; MS checks the current configuration in the commit transaction | A mismatch rejects the deletion result. This protects against an outdated deletion policy after the relevant Tablet/MS configuration has changed; it is not an atomic configuration switch across every replica. ### Configuration propagation, metadata, and lifecycle Database/table properties continue to use existing FE Journal/Image persistence. Binlog configuration remains part of existing Tablet metadata. For shared-nothing deployments, configuration changes use `UPDATE_TABLET_META_INFO` Agent Tasks, including visible Row Binlog indexes and all replicas. Tasks are tracked, configuration is persisted on BE, and FE waits for completion. A failed DDL may still have partially updated replicas. For cloud deployments, FE updates MS in batches controlled by `cloud_txn_tablet_batch_size` (default 50), then asynchronously notifies BEs to refresh cached metadata. Ordinary cache refresh and TTL discovery share an RPC but have different behavior: ordinary refresh skips uncached tablets; TTL discovery can load them. MS commit-time TTL validation remains necessary even if a BE cache has not received the new configuration. The main protocol additions are: | Interface | Purpose | |---|---| | `FrontendService.getCurrentTso()` | Obtain a query reference from Master FE | | `TMasterInfo.row_binlog_ttl_reference_tso`, field 15 | Distribute the cleanup reference through heartbeats | | `TabletMetaInfoPB.binlog_config`, field 17 | Update cloud Tablet Binlog configuration | | `PSyncTabletMetaRequest.discover_row_binlog_ttl`, field 2 | Request asynchronous discovery, including uncached tablets | | `TabletCompactionJobPB.row_binlog_ttl_seconds`, field 31 | Validate the prepared retention duration at cloud compaction commit | No TTL-enabled flag or background-reference field is added to the persisted BinlogConfig/TabletMeta schema. Existing Rowset commit-TSO ranges remain data metadata, not a GC watermark. After BE restart, configuration is restored or loaded normally, but cleanup waits for a new heartbeat reference. Clone uses the destination BE's reference rather than copying one from the source replica. Existing SQL RESTORE restrictions for ROW Binlog tables remain in place. This initial development design assumes there are no deployed TTL tables and that TTL tables are not created during a mixed-version upgrade. It does not introduce migration of intermediate development metadata or capability negotiation; existing released protocol field numbers and types are preserved. ### Consistency limits and future work With the same TTL, a query reference no older than the compaction reference, and no historical physical gaps, data removed by that compaction is already outside the query's visible retention window. A delayed heartbeat generally delays reclamation instead of widening the deletion range. The initial design has the following limits: - Heartbeats, configuration changes, cache refreshes, and compaction do not advance atomically across replicas. - Increasing TTL does not restore physically removed history. MIN_DELTA expiration checks use the current retention cutoff and cannot identify every historical gap caused by prior reclamation. - The design does not track consumer progress or wait for all consumers to acknowledge records. - There is no active-query safe point. Open readers retain existing lifetime protection, but late fragments and cross-replica retries are not covered by a query-wide history-retention guarantee. - Whole-Rowset expiration and asynchronous file deletion mean physical storage may retain data beyond its logical TTL. Two mechanisms are reserved for future work and are not part of this issue's initial implementation: | Direction | Additional guarantee | Required work | |---|---|---| | Persisted GC watermark | A durable, non-decreasing logical deletion boundary that remains valid after TTL increases | Persist and publish a table-level watermark to query FEs before authorizing physical removal | | Active-query safe point | Protect history needed by late fragments, retries, and long-running queries | Cross-FE registration, leases, renewal, cleanup, and a deletion cutoff strictly below the minimum active inclusive scan start | ### Use case ### 1. Retain recent changes for incremental consumers A table receives continuous events, while downstream jobs normally consume changes every few minutes. Configure a one-day window so jobs can process recent history without retaining all past changes indefinitely. ```sql SELECT event_id, event_time, payload FROM event_log@incr("incrementType" = "DETAIL") ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__; ``` Without an explicit start, the query returns changes within the current TTL window. Old Binlog records expire while ordinary queries against `event_log` continue to return the business rows. ### 2. Detect an incremental consumer that has fallen behind A MIN_DELTA consumer pauses longer than the retention window and resumes from an old position. Silently computing a delta from only the remaining history could produce an incomplete result. When its converted scan start precedes `nextTso(cutoff)`, planning fails with: ```text Row binlog offset has expired according to binlog.ttl_seconds ``` The consumer can then explicitly rebuild from a current snapshot or otherwise reinitialize its state. Automatic resnapshotting is outside this proposal. This check detects an expired current-window start, not every possible historical physical gap. ### 3. Support incremental materialized view maintenance An IVM workload reads changes between refreshes. A retention window can be chosen to cover the expected refresh delay and operational recovery period. IVM paths that read Binlog use the same statement-level Master reference and TTL boundary. An expired MIN_DELTA start must be reported rather than treated as a complete delta. Applications must still account for refresh lag and the initial implementation's lack of active-query protection. ### 4. Reclaim logs from an idle table A table receives a one-time batch and then has no further writes. Its logs should not remain indefinitely merely because no new load triggers ordinary compaction. Heartbeat references continue to advance, and the BE scanner discovers expired Rowsets and schedules Binlog compaction. Source-table rows remain available. Actual file deletion occurs later through the existing storage lifecycle. ### 5. Reclaim cloud Binlog after BE restart or cache eviction A cloud Binlog tablet is absent from BE caches after restart or eviction, and users do not query it again. FE catalog discovery finds the tablet and queues preparation on an eligible owner BE. Once that BE has a valid heartbeat reference, preparation loads metadata and Rowsets and can schedule cleanup without requiring a user read or write. ### 6. Apply different retention windows across workloads Use database defaults for newly created tables and table-level overrides for workloads with longer recovery requirements. An existing table can be extended from one to three days: ```sql ALTER TABLE event_log SET ( "binlog.ttl_seconds" = "259200" ); ``` New queries use the effective table policy. Compactions prepared under an old duration are rejected if they attempt TTL deletion after the relevant current configuration has changed. An extension can expose records that still exist physically, but cannot recreate already reclaimed history. ALTER DATABASE does not retroactively change existing tables. ### 7. Separate logical expiration from cleanup during maintenance An operator temporarily disables automatic compaction. TTL-aware queries should still enforce the retention window, while automatic physical cleanup is paused. After automatic compaction is re-enabled, background discovery resumes cleanup. The debug `binlog(...)` TVF can distinguish physically retained records from the records visible through TTL-aware incremental queries. ### Related issues _No response_ ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
