add transaction_id to record

Author: shortishly

etc/initdb.d/010-schema.sql

@@ -107,6 +107,7 @@ create table if not exists record (
     topition int references topition (id) on delete cascade,
     offset_id bigint not null,
     primary key (topition, offset_id),
+    transaction_id xid8 default pg_current_xact_id() not null,
     attributes smallint,
     producer_id bigint,
     producer_epoch smallint,

tansu-storage/src/pg.rs

@@ -1710,7 +1710,7 @@ impl Storage for Postgres {
         let records = self
             .prepare_query(
                 &c,
-                "record_fetch.sql",
+                "record_fetch_pg.sql",
                 &[
                     &self.cluster,
                     &topition.topic(),

tansu-storage/src/sql/mod.rs

@@ -184,6 +184,7 @@ pub(crate) static SQL: LazyLock<Cache> = LazyLock::new(|| {
             include_sql!("record_delete_by_topic.sql"),
         ),
         ("record_fetch.sql", include_sql!("record_fetch.sql")),
+        ("record_fetch_pg.sql", include_sql!("record_fetch_pg.sql")),
         ("record_insert.sql", include_sql!("record_insert.sql")),
         ("register_broker.sql", include_sql!("register_broker.sql")),
         ("topic_by_cluster.sql", include_sql!("topic_by_cluster.sql")),

tansu-storage/src/sql/record_fetch_pg.sql

@@ -0,0 +1,45 @@
+-- -*- mode: sql; sql-product: postgres; -*-
+-- Copyright ⓒ 2024-2025 Peter Morgan <peter.james.morgan@gmail.com>
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+-- prepare record_fetch (text, text, integer, integer, integer, integer) as
+with sized as (
+select
+
+r.offset_id,
+r.attributes,
+r.timestamp,
+r.k,
+r.v,
+sum(coalesce(length(r.k), 0) + coalesce(length(r.v), 0)) over (order by r.offset_id) as bytes,
+r.producer_id,
+r.producer_epoch
+
+from
+
+cluster c
+join topic t on t.cluster = c.id
+join topition tp on tp.topic = t.id
+join record r on r.topition = tp.id
+
+where
+
+c.name = $1
+and t.name = $2
+and tp.partition = $3
+and r.offset_id >= $4
+and r.offset_id < $6
+and r.transaction_id < pg_snapshot_xmin(pg_current_snapshot()))
+
+select * from sized where bytes < $5;