perf(inkless:control-plane): index and rewrite list_offsets_v1 timestamp lookups [KC-537] - #799

…amp lookups (#799)

MAX_TIMESTAMP and the concrete-timestamp branch of list_offsets_v1
filter and aggregate on batch_timestamp(timestamp_type,
batch_max_timestamp, log_append_timestamp). No index covered that
expression, so both walked every retained batch of the partition and
evaluated the function per row. Cost grew with retention, and a
partition with a large backlog pushed one ListOffsets past
statement_timeout.

V30 adds an expression index on (topic_id, partition,
batch_timestamp(...)) INCLUDE (base_offset). MAX becomes a backward
range scan. The migration runs ANALYZE so the expression gets
statistics without waiting for autovacuum on a table where 10% churn
can take days.

The index stores the expression as a function call, so batch_timestamp
must stay non-inlinable (LANGUAGE plpgsql): a LANGUAGE sql rewrite
inlines every query to a bare CASE that no longer matches and the
index goes unused silently. The migration header records that, the
measured build rate (about 200k rows/s single-threaded), and that
pre-creating the index CONCURRENTLY before the upgrade is required at
the table sizes V24 designs for.

The index alone does not fix the concrete-timestamp lookup. It picked
one batch with ORDER BY batch_id LIMIT 1, and batches_pkey yields that
order, so the planner could walk the primary key from the smallest
batch_id in the table with the partition and timestamp as filters. It
chose that walk whenever it expected an early hit, and it expects one
in the wrong cases: plpgsql runs the query as a generic plan with a
fixed selectivity for the timestamp bound, and even a custom plan
assumes matches are spread evenly along the walk while timestamps grow
with offset. A consumer seeking to a recent time then scanned nearly
the whole partition, across other partitions' rows.

V31 rewrites each lookup as MIN(base_offset) over the timestamp
predicate, then fetches that batch by offset through the covering
index. No index orders base_offset, so the aggregate has one plan: a
range scan of batches_by_timestamp_idx costing one entry per batch at
or after the target. The result is unchanged because within a
partition batch_id order is offset order; commit_file assigns offsets
and inserts batches under the same logs row lock.

A timestamp at or before the oldest retained batch would make that scan
read every index entry of the partition. V31 therefore reads the oldest
batch first, which has the smallest offset, and skips the scan when it
already qualifies.

Testing

ListOffsetsJobTest gains a case with out-of-order timestamps across
batches, pinning that the lookup returns the first batch by offset and
not the smallest qualifying timestamp. Further cases pin a tie on the
maximum timestamp, a target below the oldest batch, and the clamp to
the log start offset.

ListOffsetsTimestampScanBenchmarkTest (tagged benchmark) measures the
lookups against the retained batch count. The test container can now
stop Flyway at a target version, so the same workload runs against
V29, V30, and V31 in separate databases, with the EXPLAIN shape
following the variant. The harness runs ANALYZE in a committed
transaction: the test pool has autocommit off, and a bare execute is
rolled back on connection return.

The same runs report the cost of the new index. batches_by_timestamp_idx
takes about 59 bytes per row, roughly 46% on top of the existing
indexes of batches. Commit_file writes about 100 bytes more WAL per
batch. Commit latency shows no change beyond run-to-run noise, but the
benchmark inserts one partition in timestamp order, so it does not cover
inserts spread across many partitions.

Assisted-by: Claude Fable 5.1

…amp lookups (#799)

MAX_TIMESTAMP and the concrete-timestamp branch of list_offsets_v1
filter and aggregate on batch_timestamp(timestamp_type,
batch_max_timestamp, log_append_timestamp). No index covered that
expression, so both walked every retained batch of the partition and
evaluated the function per row. Cost grew with retention, and a
partition with a large backlog pushed one ListOffsets past
statement_timeout.

V30 adds an expression index on (topic_id, partition,
batch_timestamp(...)) INCLUDE (base_offset). MAX becomes a backward
range scan. The migration runs ANALYZE so the expression gets
statistics without waiting for autovacuum on a table where 10% churn
can take days.

The index stores the expression as a function call, so batch_timestamp
must stay non-inlinable (LANGUAGE plpgsql): a LANGUAGE sql rewrite
inlines every query to a bare CASE that no longer matches and the
index goes unused silently. The migration header records that, the
measured build rate (about 200k rows/s single-threaded), and that
pre-creating the index CONCURRENTLY before the upgrade is required at
the table sizes V24 designs for.

The index alone does not fix the concrete-timestamp lookup. It picked
one batch with ORDER BY batch_id LIMIT 1, and batches_pkey yields that
order, so the planner could walk the primary key from the smallest
batch_id in the table with the partition and timestamp as filters. It
chose that walk whenever it expected an early hit, and it expects one
in the wrong cases: plpgsql runs the query as a generic plan with a
fixed selectivity for the timestamp bound, and even a custom plan
assumes matches are spread evenly along the walk while timestamps grow
with offset. A consumer seeking to a recent time then scanned nearly
the whole partition, across other partitions' rows.

V31 rewrites each lookup as MIN(base_offset) over the timestamp
predicate, then fetches that batch by offset through the covering
index. No index orders base_offset, so the aggregate has one plan: a
range scan of batches_by_timestamp_idx costing one entry per batch at
or after the target. The result is unchanged because within a
partition batch_id order is offset order; commit_file assigns offsets
and inserts batches under the same logs row lock.

A timestamp at or before the oldest retained batch would make that scan
read every index entry of the partition. V31 therefore reads the oldest
batch first, which has the smallest offset, and skips the scan when it
already qualifies.

Testing

ListOffsetsJobTest gains a case with out-of-order timestamps across
batches, pinning that the lookup returns the first batch by offset and
not the smallest qualifying timestamp. Further cases pin a tie on the
maximum timestamp, a target below the oldest batch, and the clamp to
the log start offset.

ListOffsetsTimestampScanBenchmarkTest (tagged benchmark) measures the
lookups against the retained batch count. The test container can now
stop Flyway at a target version, so the same workload runs against
V29, V30, and V31 in separate databases, with the EXPLAIN shape
following the variant. The harness runs ANALYZE in a committed
transaction: the test pool has autocommit off, and a bare execute is
rolled back on connection return.

The same runs report the cost of the new index. batches_by_timestamp_idx
takes about 59 bytes per row, roughly 46% on top of the existing
indexes of batches. Commit_file writes about 100 bytes more WAL per
batch. Commit latency shows no change beyond run-to-run noise, but the
benchmark inserts one partition in timestamp order, so it does not cover
inserts spread across many partitions.

Assisted-by: Claude Fable 5.1

…amp lookups (#799)

MAX_TIMESTAMP and the concrete-timestamp branch of list_offsets_v1
filter and aggregate on batch_timestamp(timestamp_type,
batch_max_timestamp, log_append_timestamp). No index covered that
expression, so both walked every retained batch of the partition and
evaluated the function per row. Cost grew with retention, and a
partition with a large backlog pushed one ListOffsets past
statement_timeout.

V30 adds an expression index on (topic_id, partition,
batch_timestamp(...)) INCLUDE (base_offset). MAX becomes a backward
range scan. The migration runs ANALYZE so the expression gets
statistics without waiting for autovacuum on a table where 10% churn
can take days.

The index stores the expression as a function call, so batch_timestamp
must stay non-inlinable (LANGUAGE plpgsql): a LANGUAGE sql rewrite
inlines every query to a bare CASE that no longer matches and the
index goes unused silently. The migration header records that, the
measured build rate (about 200k rows/s single-threaded), and that
pre-creating the index CONCURRENTLY before the upgrade is required at
the table sizes V24 designs for.

The index alone does not fix the concrete-timestamp lookup. It picked
one batch with ORDER BY batch_id LIMIT 1, and batches_pkey yields that
order, so the planner could walk the primary key from the smallest
batch_id in the table with the partition and timestamp as filters. It
chose that walk whenever it expected an early hit, and it expects one
in the wrong cases: plpgsql runs the query as a generic plan with a
fixed selectivity for the timestamp bound, and even a custom plan
assumes matches are spread evenly along the walk while timestamps grow
with offset. A consumer seeking to a recent time then scanned nearly
the whole partition, across other partitions' rows.

V31 rewrites each lookup as MIN(base_offset) over the timestamp
predicate, then fetches that batch by offset through the covering
index. No index orders base_offset, so the aggregate has one plan: a
range scan of batches_by_timestamp_idx costing one entry per batch at
or after the target. The result is unchanged because within a
partition batch_id order is offset order; commit_file assigns offsets
and inserts batches under the same logs row lock.

A timestamp at or before the oldest retained batch would make that scan
read every index entry of the partition. V31 therefore reads the oldest
batch first, which has the smallest offset, and skips the scan when it
already qualifies.

Testing

ListOffsetsJobTest gains a case with out-of-order timestamps across
batches, pinning that the lookup returns the first batch by offset and
not the smallest qualifying timestamp. Further cases pin a tie on the
maximum timestamp, a target below the oldest batch, and the clamp to
the log start offset.

ListOffsetsTimestampScanBenchmarkTest (tagged benchmark) measures the
lookups against the retained batch count. The test container can now
stop Flyway at a target version, so the same workload runs against
V29, V30, and V31 in separate databases, with the EXPLAIN shape
following the variant. The harness runs ANALYZE in a committed
transaction: the test pool has autocommit off, and a bare execute is
rolled back on connection return.

The same runs report the cost of the new index. batches_by_timestamp_idx
takes about 59 bytes per row, roughly 46% on top of the existing
indexes of batches. Commit_file writes about 100 bytes more WAL per
batch. Commit latency shows no change beyond run-to-run noise, but the
benchmark inserts one partition in timestamp order, so it does not cover
inserts spread across many partitions.

Assisted-by: Claude Fable 5.1