feat(inkless:retention): track earliest batch timestamp per log - #704
Merged
Conversation
…hmark Add a schema-agnostic write-path benchmark (@tag("benchmark"), excluded from CI) that drives only the commit/delete Job APIs and never references the new column, so the same test compiles and runs both before and after the column exists. It is committed first, ahead of the feature, on purpose: its parent is the pre-column state, so the maintenance-cost A/B is reproducible by checkout alone -- run it on this commit for the "before" numbers and on the feature commit for the "after" -- with no cherry-pick or separate worktree. Shapes: steady-state append (recompute guard skips), commit-into-empty, and delete-advance.
Add logs.earliest_batch_timestamp: the effective timestamp (see batch_timestamp) of the batch at log_start, i.e. the oldest retained batch of the partition. This lets a future retention check decide "time retention deletes nothing" from a single column instead of scanning the whole partition, for the common steady-state case where the oldest batch is still within retention. No reader uses the column yet; this only adds and maintains it. The column is maintained under the existing logs FOR UPDATE in every write path that changes the oldest batch, sourced from the batches row (so it matches what a coalesced row stores), never from the request: - delete_records_v1 recomputes it when a delete advances log_start, or NULLs it when the log becomes empty; - commit_file_v1 and commit_file_v2 populate it when a partition in the commit has no value yet. This covers empty->non-empty (the just committed batch is the oldest) and lazily backfills pre-existing NULL logs; an IS NULL guard keeps steady-state appends free; - prune_batches_below_highest_tiered_offset_v1 (cross-tier pruning also removes the oldest batches and advances log_start) recomputes it when the prune deleted something, or NULLs it when the log becomes empty. NULL means "unknown, must scan", so there is no backfill over the batches table at migration time. Both commit functions are maintained because batch coalescing defaults off, making commit_file_v1 the default production commit path. Regenerates jOOQ sources (new column + timestamp_nullable_t domain, and the schema-version stamp bump) and updates existing tests for the new LogsRecord constructor arity. Adds LogsEarliestBatchTimestampTest, which covers the invariant across every write path (commit into empty, CreateTime vs LogAppendTime, delete/prune advance recompute, delete-all and prune-all reset to NULL, no-op delete/prune, and commit backfill of a pre-existing NULL log); it lives with the feature because it references the new column and asserts the maintenance this commit introduces.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a new persisted optimization field in the Inkless Postgres control-plane schema: logs.earliest_batch_timestamp, intended to support a future retention fast-path by tracking the effective timestamp of the oldest retained batch. It adds a DB migration to create and maintain the column across relevant write paths, updates jOOQ-generated schema bindings, and extends the test suite (plus an opt-in benchmark) to validate and quantify the behavior.
Changes:
- Add
logs.earliest_batch_timestamp(nullable) and maintain it incommit_file_v1/v2,delete_records_v1, andprune_batches_below_highest_tiered_offset_v1. - Add correctness tests covering all maintenance paths and edge cases (empty log, append, delete/prune advancing log start, backfill from NULL).
- Regenerate jOOQ bindings for schema version 20 and update existing tests to include the new column in
LogsRecordexpectations.
Reviewed changes
Copilot reviewed 7 out of 125 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/TopicsAndPartitionsCreateJobTest.java | Updates expected LogsRecord values to include earliest_batch_timestamp (NULL for fresh logs). |
| storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/LogsEarliestBatchTimestampTest.java | New test suite validating earliest_batch_timestamp maintenance across commit/delete/prune paths. |
| storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/EarliestBatchTimestampMaintenanceBenchmarkTest.java | New opt-in benchmark (tagged benchmark) to measure write-path overhead of maintaining the column. |
| storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/DeleteTopicJobTest.java | Updates expected LogsRecord to include populated earliest_batch_timestamp where applicable. |
| storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/DeleteRecordsJobTest.java | Updates expected LogsRecord to include recomputed/null earliest_batch_timestamp after deletes. |
| storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/CommitFileJobTest.java | Updates expected LogsRecord to include earliest_batch_timestamp for CREATE_TIME vs LOG_APPEND_TIME commits. |
| storage/inkless/src/main/resources/db/migration/V20__Logs_earliest_batch_timestamp.sql | Adds new column + domain and updates DB write-path functions to maintain earliest_batch_timestamp. |
| storage/inkless/src/main/jooq/org/jooq/generated/UDTs.java | jOOQ regen: bump schema version to 20. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/RepairDisklessLogResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/RepairDisklessLogRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/RepairDisklessLogResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/RepairDisklessLogRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/PruneBatchesBelowHighestTieredOffsetResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/PruneBatchesBelowHighestTieredOffsetRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/ListOffsetsResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/ListOffsetsRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogProducerStateV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/FindBatchesResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/FindBatchesRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/EnforceRetentionResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/EnforceRetentionRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/DeleteRecordsResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/DeleteRecordsRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/CommitBatchResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/CommitBatchRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/BatchMetadataV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/BatchInfoV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/AdvanceCrossTierLogStartResponseV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/records/AdvanceCrossTierLogStartRequestV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/PruneBatchesBelowHighestTieredOffsetResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/PruneBatchesBelowHighestTieredOffsetRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/RepairDisklessLogResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/RepairDisklessLogRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/PruneBatchesBelowHighestTieredOffsetResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/PruneBatchesBelowHighestTieredOffsetRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/ListOffsetsResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/ListOffsetsRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogProducerStateV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/FindBatchesResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/FindBatchesRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/EnforceRetentionResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/EnforceRetentionRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/DeleteRecordsResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/DeleteRecordsRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/CommitBatchResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/CommitBatchRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/BatchMetadataV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/BatchInfoV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/AdvanceCrossTierLogStartResponseV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/AdvanceCrossTierLogStartRequestV1Path.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/ListOffsetsResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/ListOffsetsRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogProducerStateV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/FindBatchesResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/FindBatchesRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/EnforceRetentionResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/EnforceRetentionRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/DeleteRecordsResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/DeleteRecordsRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/CommitBatchResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/CommitBatchRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/BatchMetadataV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/BatchInfoV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/AdvanceCrossTierLogStartResponseV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/udt/AdvanceCrossTierLogStartRequestV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/RepairDisklessLogV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/RepairDisklessLogV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/PruneBatchesBelowHighestTieredOffsetV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/ProducerStateRecord.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/LogsRecord.java | jOOQ regen: adds earliest_batch_timestamp accessor + constructor param. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/ListOffsetsV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/InitDisklessLogV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FindBatchesV2Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FindBatchesV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FilesRecord.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/EnforceRetentionV2Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/EnforceRetentionV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/DeleteRecordsV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/CommitFileV2Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/CommitFileV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/BatchesRecord.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/records/AdvanceCrossTierLogStartV1Record.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/PruneBatchesBelowHighestTieredOffsetV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/ProducerState.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/Logs.java | jOOQ regen: adds EARLIEST_BATCH_TIMESTAMP field + schema version 20 metadata. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/ListOffsetsV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/InitDisklessLogV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/FindBatchesV2.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/FindBatchesV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/Files.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/EnforceRetentionV2.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/EnforceRetentionV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/DeleteRecordsV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/CommitFileV2.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/CommitFileV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/Batches.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/tables/AdvanceCrossTierLogStartV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/Tables.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/routines/MarkFileToDeleteV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/routines/FlushCommitRunV2.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteTopicV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteFilesV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteBatchV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/routines/BatchTimestamp.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/Routines.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/Keys.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/Indexes.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/PruneBatchesBelowHighestTieredOffsetErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/ListOffsetsResponseErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/InitDisklessLogErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/FindBatchesResponseErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/FileStateT.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/FileReasonT.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/EnforceRetentionResponseErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/DeleteRecordsResponseErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/CommitBatchResponseErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/enums/AdvanceCrossTierLogStartResponseErrorV1.java | jOOQ regen: schema version 20 metadata update. |
| storage/inkless/src/main/jooq/org/jooq/generated/Domains.java | jOOQ regen: adds timestamp_nullable_t domain binding + schema version 20 metadata. |
| storage/inkless/src/main/jooq/org/jooq/generated/DefaultSchema.java | jOOQ regen: registers new domain + schema version 20 metadata. |
jeqo
marked this pull request as ready for review
July 16, 2026 14:32
ivanyu
approved these changes
Jul 17, 2026
ivanyu
left a comment
Member
There was a problem hiding this comment.
LGTM. Do you think it makes sense to mirror this to the in-memory implementation?
Contributor
Author
jeqo
added a commit
that referenced
this pull request
Jul 20, 2026
* test(inkless:retention): add earliest_batch_timestamp write-path benchmark Add a schema-agnostic write-path benchmark (@tag("benchmark"), excluded from CI) that drives only the commit/delete Job APIs and never references the new column, so the same test compiles and runs both before and after the column exists. It is committed first, ahead of the feature, on purpose: its parent is the pre-column state, so the maintenance-cost A/B is reproducible by checkout alone -- run it on this commit for the "before" numbers and on the feature commit for the "after" -- with no cherry-pick or separate worktree. Shapes: steady-state append (recompute guard skips), commit-into-empty, and delete-advance. * feat(inkless:retention): track earliest batch timestamp per log Add logs.earliest_batch_timestamp: the effective timestamp (see batch_timestamp) of the batch at log_start, i.e. the oldest retained batch of the partition. This lets a future retention check decide "time retention deletes nothing" from a single column instead of scanning the whole partition, for the common steady-state case where the oldest batch is still within retention. No reader uses the column yet; this only adds and maintains it. The column is maintained under the existing logs FOR UPDATE in every write path that changes the oldest batch, sourced from the batches row (so it matches what a coalesced row stores), never from the request: - delete_records_v1 recomputes it when a delete advances log_start, or NULLs it when the log becomes empty; - commit_file_v1 and commit_file_v2 populate it when a partition in the commit has no value yet. This covers empty->non-empty (the just committed batch is the oldest) and lazily backfills pre-existing NULL logs; an IS NULL guard keeps steady-state appends free; - prune_batches_below_highest_tiered_offset_v1 (cross-tier pruning also removes the oldest batches and advances log_start) recomputes it when the prune deleted something, or NULLs it when the log becomes empty. NULL means "unknown, must scan", so there is no backfill over the batches table at migration time. Both commit functions are maintained because batch coalescing defaults off, making commit_file_v1 the default production commit path. Regenerates jOOQ sources (new column + timestamp_nullable_t domain, and the schema-version stamp bump) and updates existing tests for the new LogsRecord constructor arity. Adds LogsEarliestBatchTimestampTest, which covers the invariant across every write path (commit into empty, CreateTime vs LogAppendTime, delete/prune advance recompute, delete-all and prune-all reset to NULL, no-op delete/prune, and commit backfill of a pre-existing NULL log); it lives with the feature because it references the new column and asserts the maintenance this commit introduces.
jeqo
added a commit
that referenced
this pull request
Jul 20, 2026
* test(inkless:retention): add earliest_batch_timestamp write-path benchmark Add a schema-agnostic write-path benchmark (@tag("benchmark"), excluded from CI) that drives only the commit/delete Job APIs and never references the new column, so the same test compiles and runs both before and after the column exists. It is committed first, ahead of the feature, on purpose: its parent is the pre-column state, so the maintenance-cost A/B is reproducible by checkout alone -- run it on this commit for the "before" numbers and on the feature commit for the "after" -- with no cherry-pick or separate worktree. Shapes: steady-state append (recompute guard skips), commit-into-empty, and delete-advance. * feat(inkless:retention): track earliest batch timestamp per log Add logs.earliest_batch_timestamp: the effective timestamp (see batch_timestamp) of the batch at log_start, i.e. the oldest retained batch of the partition. This lets a future retention check decide "time retention deletes nothing" from a single column instead of scanning the whole partition, for the common steady-state case where the oldest batch is still within retention. No reader uses the column yet; this only adds and maintains it. The column is maintained under the existing logs FOR UPDATE in every write path that changes the oldest batch, sourced from the batches row (so it matches what a coalesced row stores), never from the request: - delete_records_v1 recomputes it when a delete advances log_start, or NULLs it when the log becomes empty; - commit_file_v1 and commit_file_v2 populate it when a partition in the commit has no value yet. This covers empty->non-empty (the just committed batch is the oldest) and lazily backfills pre-existing NULL logs; an IS NULL guard keeps steady-state appends free; - prune_batches_below_highest_tiered_offset_v1 (cross-tier pruning also removes the oldest batches and advances log_start) recomputes it when the prune deleted something, or NULLs it when the log becomes empty. NULL means "unknown, must scan", so there is no backfill over the batches table at migration time. Both commit functions are maintained because batch coalescing defaults off, making commit_file_v1 the default production commit path. Regenerates jOOQ sources (new column + timestamp_nullable_t domain, and the schema-version stamp bump) and updates existing tests for the new LogsRecord constructor arity. Adds LogsEarliestBatchTimestampTest, which covers the invariant across every write path (commit into empty, CreateTime vs LogAppendTime, delete/prune advance recompute, delete-all and prune-all reset to NULL, no-op delete/prune, and commit backfill of a pre-existing NULL log); it lives with the feature because it references the new column and asserts the maintenance this commit introduces.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds
logs.earliest_batch_timestampand maintains it in the control-plane write paths, so a future retention check can decide "time retention deletes nothing" from one column instead of the current O(depth) reverse-aggregate scan over a partition's wholebatcheshistory. This PR only adds and maintains the column (plus tests and a benchmark); no reader consumes it yet, so retention behavior is unchanged. The reader-side short-circuit is a follow-up.The column
logs.earliest_batch_timestampis the effective timestamp (perbatch_timestamp:batch_max_timestampfor CreateTime,log_append_timestampfor LogAppendTime) of the batch atlog_start_offset-- the oldest retained batch.NULLmeans "unknown, must scan"; pre-existing and freshly created logs startNULL(no migration-time backfill) and are populated lazily as writes touch each log.Maintenance
Maintained server-side under the existing
logs FOR UPDATE, sourced from thebatchesrow (so it matches coalesced rows), via a single indexed lookup of the oldest batch:delete_records_v1: recompute when a delete advanceslog_start(orNULLwhen the log empties); no advance leaves it untouched.commit_file_v1andcommit_file_v2: populate only when the partition has no value yet (IS NULL) -- covers empty->non-empty and lazily backfills pre-existingNULLlogs. Appends add newer batches at the head, so a populated log is untouched and steady-state appends stay free. Both are maintained because batch coalescing defaults off (batch.coalescing.enabled=false), makingcommit_file_v1the default path.prune_batches_below_highest_tiered_offset_v1: like delete, cross-tier prune removes the oldest batches and advanceslog_start, so it recomputes when it deleted something (orNULLwhen empty).Changes
V20: adds the nullabletimestamp_nullable_tdomain and theearliest_batch_timestampcolumn, andCREATE OR REPLACEs the four functions above in place (signatures unchanged; no new/dropped versions).LogsRecordarity (with correct expected timestamps).LogsEarliestBatchTimestampTest: invariant across commit/delete/prune (empty->non-empty, CreateTime vs LogAppendTime, advance recompute, delete-all/prune-all ->NULL, no-op delete/prune, backfill of a pre-existingNULLlog).EarliestBatchTimestampMaintenanceBenchmarkTest(@Tag("benchmark"), excluded from CI).Compatibility / rolling upgrade
Backward compatible; old and new brokers can coexist indefinitely.
ADD COLUMN(no rewrite/backfill), ignored by old broker code.CREATE OR REPLACEbodies only), so no cross-version function mismatch.Only the usual migration-ordering discipline applies (schema stays ahead of not-yet-upgraded brokers). Broker-binary rollback is safe; the migration need not be rolled back for a downgrade.
Benchmark
The benchmark is schema-agnostic (only commit/delete Job APIs, never the new column) and is committed first, so its parent is the pre-column baseline: the A/B is reproducible by checkout -- run
./gradlew :storage:inkless:benchmarkTest --tests "*EarliestBatchTimestampMaintenanceBenchmarkTest"on the benchmark commit ("before") and the feature commit ("after"), then diffms/call.Representative run (16 partitions/call, 500 calls/shape, wall-clock vs a local Postgres container; indicative):
The gap is within run-to-run variance (delete-advance spanned 1.9-2.4 across branch runs; "after" came out equal-or-lower), so there is no measurable regression: the steady-state append hot path is untouched and the recompute paths stay flat rather than scaling with log depth.
Scope / follow-ups
Testing
./gradlew :storage:inkless:test --tests "io.aiven.inkless.control_plane.postgres.*"is green; the benchmark runs via the opt-inbenchmarkTesttask.