Experiment in decoding postgres tuples

postgres_ffi: expose HeapTupleHeaderData
2026-07-08 22:50:37 +00:00 · 2025-05-27 12:16:11 +02:00 · 2025-05-27 12:16:04 +02:00
17 changed files with 279 additions and 634 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -701,7 +701,7 @@ dependencies = [
 "http 1.1.0",
 "http-body 1.0.0",
 "http-body-util",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-util",
 "itoa",
 "matchit",
@@ -2330,7 +2330,7 @@ dependencies = [
 "futures-core",
 "futures-sink",
 "http-body-util",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-util",
 "pin-project",
 "rand 0.8.5",
@@ -2883,9 +2883,9 @@ dependencies = [

 [[package]]
 name = "httparse"
-version = "1.10.1"
+version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
+checksum = "d897f394bad6a705d5f4104762e116a75639e470d80901eed05a860a95cb1904"

 [[package]]
 name = "httpdate"
@@ -2935,9 +2935,9 @@ dependencies = [

 [[package]]
 name = "hyper"
-version = "1.6.0"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cc2b571658e38e0c01b1fdca3bbbe93c00d3d71693ff2770043f8c29bc7d6f80"
+checksum = "50dfd22e0e76d0f662d429a5f80fcaf3855009297eab6a0a9f8543834744ba05"
 dependencies = [
 "bytes",
 "futures-channel",
@@ -2977,7 +2977,7 @@ checksum = "a0bea761b46ae2b24eb4aef630d8d1c398157b6fc29e6350ecf090a0b70c952c"
 dependencies = [
 "futures-util",
 "http 1.1.0",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-util",
 "rustls 0.22.4",
 "rustls-pki-types",
@@ -2992,7 +2992,7 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3203a961e5c83b6f5498933e78b6b263e208c197b63e9c6c53cc82ffd3f63793"
 dependencies = [
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-util",
 "pin-project-lite",
 "tokio",
@@ -3001,20 +3001,20 @@ dependencies = [

 [[package]]
 name = "hyper-util"
-version = "0.1.12"
+version = "0.1.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cf9f1e950e0d9d1d3c47184416723cf29c0d1f93bd8cccf37e4beb6b44f31710"
+checksum = "cde7055719c54e36e95e8719f95883f22072a48ede39db7fc17a4e1d5281e9b9"
 dependencies = [
 "bytes",
 "futures-channel",
 "futures-util",
 "http 1.1.0",
 "http-body 1.0.0",
- "hyper 1.6.0",
- "libc",
+ "hyper 1.4.1",
 "pin-project-lite",
 "socket2",
 "tokio",
+ "tower 0.4.13",
 "tower-service",
 "tracing",
 ]
@@ -4236,8 +4236,6 @@ name = "pagebench"
 version = "0.1.0"
 dependencies = [
 "anyhow",
- "async-trait",
- "bytes",
 "camino",
 "clap",
 "futures",
@@ -4246,13 +4244,10 @@ dependencies = [
 "humantime-serde",
 "pageserver_api",
 "pageserver_client",
- "pageserver_client_grpc",
- "pageserver_page_api",
 "rand 0.8.5",
 "reqwest",
 "serde",
 "serde_json",
- "thiserror 1.0.69",
 "tokio",
 "tokio-util",
 "tracing",
@@ -4437,29 +4432,6 @@ dependencies = [
 "workspace_hack",
 ]

-[[package]]
-name = "pageserver_client_grpc"
-version = "0.1.0"
-dependencies = [
- "bytes",
- "futures",
- "http 1.1.0",
- "hyper 1.6.0",
- "hyper-util",
- "metrics",
- "pageserver_page_api",
- "priority-queue",
- "rand 0.8.5",
- "thiserror 1.0.69",
- "tokio",
- "tokio-util",
- "tonic 0.13.1",
- "tower 0.4.13",
- "tracing",
- "utils",
- "uuid",
-]
-
 [[package]]
 name = "pageserver_compaction"
 version = "0.1.0"
@@ -5036,17 +5008,6 @@ dependencies = [
 "elliptic-curve 0.13.8",
 ]

-[[package]]
-name = "priority-queue"
-version = "2.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef08705fa1589a1a59aa924ad77d14722cb0cd97b67dd5004ed5f4a4873fce8d"
-dependencies = [
- "autocfg",
- "equivalent",
- "indexmap 2.9.0",
-]
-
 [[package]]
 name = "proc-macro2"
 version = "1.0.94"
@@ -5247,7 +5208,7 @@ dependencies = [
 "humantime",
 "humantime-serde",
 "hyper 0.14.30",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-util",
 "indexmap 2.9.0",
 "ipnet",
@@ -5643,7 +5604,7 @@ dependencies = [
 "http-body-util",
 "http-types",
 "humantime-serde",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "itertools 0.10.5",
 "metrics",
 "once_cell",
@@ -5683,7 +5644,7 @@ dependencies = [
 "http 1.1.0",
 "http-body 1.0.0",
 "http-body-util",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-rustls 0.26.0",
 "hyper-util",
 "ipnet",
@@ -5740,7 +5701,7 @@ dependencies = [
 "futures",
 "getrandom 0.2.11",
 "http 1.1.0",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "parking_lot 0.11.2",
 "reqwest",
 "reqwest-middleware",
@@ -6681,12 +6642,12 @@ dependencies = [

 [[package]]
 name = "socket2"
-version = "0.5.10"
+version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678"
+checksum = "7b5fac59a5cb5dd637972e5fca70daf0523c9067fcdc4842f053dae04a18f8e9"
 dependencies = [
 "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.48.0",
 ]

 [[package]]
@@ -6752,7 +6713,7 @@ dependencies = [
 "http-body-util",
 "http-utils",
 "humantime",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-util",
 "metrics",
 "once_cell",
@@ -7577,12 +7538,11 @@ dependencies = [
 "axum",
 "base64 0.22.1",
 "bytes",
- "flate2",
 "h2 0.4.4",
 "http 1.1.0",
 "http-body 1.0.0",
 "http-body-util",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-timeout",
 "hyper-util",
 "percent-encoding",
@@ -7639,7 +7599,6 @@ dependencies = [
 "tokio",
 "tower-layer",
 "tower-service",
- "tracing",
 ]

 [[package]]
@@ -8632,7 +8591,7 @@ dependencies = [
 "hex",
 "hmac",
 "hyper 0.14.30",
- "hyper 1.6.0",
+ "hyper 1.4.1",
 "hyper-util",
 "indexmap 2.9.0",
 "itertools 0.12.1",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -8,7 +8,6 @@ members = [
    "pageserver/compaction",
    "pageserver/ctl",
    "pageserver/client",
-    "pageserver/client_grpc",
    "pageserver/pagebench",
    "pageserver/page_api",
    "proxy",
@@ -200,7 +199,7 @@ tokio-tar = "0.3"
 tokio-util = { version = "0.7.10", features = ["io", "rt"] }
 toml = "0.8"
 toml_edit = "0.22"
-tonic = { version = "0.13.1", default-features = false, features = ["gzip", "channel", "codegen", "prost", "router", "server", "tls-ring", "tls-native-roots"] }
+tonic = { version = "0.13.1", default-features = false, features = ["channel", "codegen", "prost", "router", "server", "tls-ring", "tls-native-roots"] }
 tonic-reflection = { version = "0.13.1", features = ["server"] }
 tower = { version = "0.5.2", default-features = false }
 tower-http = { version = "0.6.2", features = ["auth", "request-id", "trace"] }
@@ -255,7 +254,6 @@ metrics = { version = "0.1", path = "./libs/metrics/" }
 pageserver = { path = "./pageserver" }
 pageserver_api = { version = "0.1", path = "./libs/pageserver_api/" }
 pageserver_client = { path = "./pageserver/client" }
-pageserver_client_grpc = { path = "./pageserver/client_grpc" }
 pageserver_compaction = { version = "0.1", path = "./pageserver/compaction/" }
 pageserver_page_api = { path = "./pageserver/page_api" }
 postgres_backend = { version = "0.1", path = "./libs/postgres_backend/" }
--- a/libs/postgres_ffi/bindgen_deps.h
+++ b/libs/postgres_ffi/bindgen_deps.h
@@ -12,3 +12,5 @@
 #include "storage/bufpage.h"
 #include "storage/off.h"
 #include "access/multixact.h"
+#include "postgres.h"
+#include "access/htup.h"
--- a/libs/postgres_ffi/build.rs
+++ b/libs/postgres_ffi/build.rs
@@ -29,6 +29,11 @@ impl ParseCallbacks for PostgresFfiCallbacks {
            "CheckPoint",
            "FullTransactionId",
            "ControlFileData",
+            "PageXLogRecPtr",
+            "HeapTupleHeaderData",
+            "ItemPointerData",
+            "BlockIdData",
+            "OffsetNumber",
        ];

        if serde_list.contains(&derive_info.name) {
@@ -125,6 +130,11 @@ fn main() -> anyhow::Result<()> {
            .allowlist_var("PG_CONTROL_FILE_SIZE")
            .allowlist_var("PG_CONTROLFILEDATA_OFFSETOF_CRC")
            .allowlist_type("PageHeaderData")
+            .allowlist_type("HeapTupleHeaderData")
+            .allowlist_type("ItemPointerData")
+            .allowlist_type("TransactionId")
+            .allowlist_type("LocationIndex")
+            .allowlist_type("Datum")
            .allowlist_type("DBState")
            .allowlist_type("RelMapFile")
            .allowlist_type("RepOriginId")
--- a/libs/postgres_ffi/src/lib.rs
+++ b/libs/postgres_ffi/src/lib.rs
@@ -223,9 +223,10 @@ pub mod walrecord;

 // Export some widely used datatypes that are unlikely to change across Postgres versions
 pub use v14::bindings::{
-    BlockNumber, CheckPoint, ControlFileData, MultiXactId, OffsetNumber, Oid, PageHeaderData,
-    RepOriginId, TimeLineID, TimestampTz, TransactionId, XLogRecPtr, XLogRecord, XLogSegNo, uint32,
-    uint64,
+    BlockNumber, CheckPoint, ControlFileData, HeapTupleHeaderData, ItemIdData, ItemPointerData,
+    LocationIndex, MultiXactId, OffsetNumber, Oid, PageHeaderData, PageXLogRecPtr, RepOriginId,
+    TimeLineID, TimestampTz, TransactionId, XLogRecPtr, XLogRecord, XLogSegNo, uint8, uint16,
+    uint32, uint64,
 };
 // Likewise for these, although the assumption that these don't change is a little more iffy.
 pub use v14::bindings::{MultiXactOffset, MultiXactStatus};
--- a/pageserver/client_grpc/Cargo.toml
+++ b/pageserver/client_grpc/Cargo.toml
@@ -1,25 +0,0 @@
-[package]
-name = "pageserver_client_grpc"
-version = "0.1.0"
-edition = "2024"
-
-[dependencies]
-bytes.workspace = true
-futures.workspace = true
-http.workspace = true
-thiserror.workspace = true
-tonic.workspace = true
-tracing.workspace = true
-tokio = { version = "1.43.1", features = ["full", "macros", "net", "io-util", "rt", "rt-multi-thread"] }
-uuid = { version = "1", features = ["v4"] }
-tower = {  version = "0.4", features = ["timeout", "util"] }
-rand = "0.8"
-tokio-util = { version = "0.7", features = ["compat"] }
-hyper-util = "0.1.9"
-hyper = "1.6.0"
-metrics.workspace = true
-priority-queue = "2.3.1"
-
-
-pageserver_page_api.workspace = true
-utils.workspace = true
--- a/pageserver/client_grpc/src/lib.rs
+++ b/pageserver/client_grpc/src/lib.rs
@@ -1,193 +0,0 @@
-//
-// Pageserver gRPC client library
-//
-// This library provides a gRPC client for the pageserver for the
-// communicator project.
-//
-// This library is a work in progress.
-//
-// TODO: This should properly use the shard map
-//
-
-use std::collections::HashMap;
-
-use bytes::Bytes;
-use futures::{StreamExt};
-use thiserror::Error;
-use tonic::metadata::AsciiMetadataValue;
-
-use pageserver_page_api::model::*;
-use pageserver_page_api::proto;
-
-use pageserver_page_api::proto::PageServiceClient;
-use utils::shard::ShardIndex;
-
-use std::fmt::Debug;
-use tracing::error;
-
-use tokio::sync::RwLock;
-
-use tonic::transport::{Channel, Endpoint};
-#[derive(Error, Debug)]
-pub enum PageserverClientError {
-    #[error("could not connect to service: {0}")]
-    ConnectError(#[from] tonic::transport::Error),
-    #[error("could not perform request: {0}`")]
-    RequestError(#[from] tonic::Status),
-    #[error("protocol error: {0}")]
-    ProtocolError(#[from] ProtocolError),
-    #[error("could not perform request: {0}`")]
-    InvalidUri(#[from] http::uri::InvalidUri),
-    #[error("could not perform request: {0}`")]
-    Other(String),
-}
-
-pub struct PageserverClient {
-    _tenant_id: String,
-    _timeline_id: String,
-    _auth_token: Option<String>,
-    shard_map: HashMap<ShardIndex, String>,
-    channels: tokio::sync::RwLock<HashMap<ShardIndex, Channel>>,
-    auth_interceptor: AuthInterceptor,
-}
-
-impl PageserverClient {
-    /// TODO: this doesn't currently react to changes in the shard map.
-    pub fn new(
-        tenant_id: &str,
-        timeline_id: &str,
-        auth_token: &Option<String>,
-        shard_map: HashMap<ShardIndex, String>,
-    ) -> Self {
-        Self {
-            _tenant_id: tenant_id.to_string(),
-            _timeline_id: timeline_id.to_string(),
-            _auth_token: auth_token.clone(),
-            shard_map,
-            channels: RwLock::new(HashMap::new()),
-            auth_interceptor: AuthInterceptor::new(tenant_id, timeline_id, auth_token.as_deref()),
-        }
-    }
-    //
-    // TODO: This opens a new gRPC stream for every request, which is extremely inefficient
-    pub async fn get_page(
-        &self,
-        request: &GetPageRequest,
-    ) -> Result<Vec<Bytes>, PageserverClientError> {
-        // FIXME: calculate the shard number correctly
-        let shard = ShardIndex::unsharded();
-        let chan = self.get_client(shard).await;
-
-        let mut client =
-            PageServiceClient::with_interceptor(chan, self.auth_interceptor.for_shard(shard));
-
-        let request = proto::GetPageRequest::from(request);
-        let request_stream = futures::stream::once(std::future::ready(request));
-
-        let mut response_stream = client
-            .get_pages(tonic::Request::new(request_stream))
-            .await?
-            .into_inner();
-
-        let Some(response) = response_stream.next().await else {
-            return Err(PageserverClientError::Other(
-                "no response received for getpage request".to_string(),
-            ));
-        };
-
-        match response {
-            Err(status) => {
-                return Err(PageserverClientError::RequestError(status));
-            }
-            Ok(resp) => {
-                let response: GetPageResponse = resp.try_into().unwrap();
-                return Ok(response.page_images.to_vec());
-            }
-        }
-    }
-
-
-    //
-    // TODO: this should use a connection pool with concurrency limits,
-    // not a single connection to the shard.
-    //
-    async fn get_client(&self, shard: ShardIndex) -> Channel {
-        // Get channel from the hashmap
-        let mut channels = self.channels.write();
-        if let Some(channel) = channels.await.get(&shard) {
-            return channel.clone();
-        }
-        // Create a new channel if it doesn't exist
-        let shard_url = self
-            .shard_map
-            .get(&shard)
-            .expect("shard not found in shard map");
-
-        let attempt = Endpoint::from_shared(shard_url.clone())
-            .expect("invalid endpoint")
-            .connect()
-            .await;
-
-        match attempt {
-            Ok(channel) => {
-                channels = self.channels.write();
-                channels.await.insert(shard, channel.clone());
-                channel.clone()
-            }
-            Err(e) => {
-                // TODO: handle this more gracefully, e.g. with a connection pool retry
-                panic!("Failed to connect to shard {shard}: {e}");
-            }
-        }
-    }
-}
-
-/// Inject tenant_id, timeline_id and authentication token to all pageserver requests.
-#[derive(Clone)]
-struct AuthInterceptor {
-    tenant_id: AsciiMetadataValue,
-    shard_id: Option<AsciiMetadataValue>,
-    timeline_id: AsciiMetadataValue,
-    auth_header: Option<AsciiMetadataValue>, // including "Bearer " prefix
-}
-
-impl AuthInterceptor {
-    fn new(tenant_id: &str, timeline_id: &str, auth_token: Option<&str>) -> Self {
-        Self {
-            tenant_id: tenant_id.parse().expect("could not parse tenant id"),
-            shard_id: None,
-            timeline_id: timeline_id.parse().expect("could not parse timeline id"),
-            auth_header: auth_token
-                .map(|t| format!("Bearer {t}"))
-                .map(|t| t.parse().expect("could not parse auth token")),
-        }
-    }
-
-    fn for_shard(&self, shard_id: ShardIndex) -> Self {
-        let mut with_shard = self.clone();
-        with_shard.shard_id = Some(
-            shard_id
-                .to_string()
-                .parse()
-                .expect("could not parse shard id"),
-        );
-        with_shard
-    }
-}
-
-impl tonic::service::Interceptor for AuthInterceptor {
-    fn call(&mut self, mut req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
-        req.metadata_mut()
-            .insert("neon-tenant-id", self.tenant_id.clone());
-        if let Some(shard_id) = &self.shard_id {
-            req.metadata_mut().insert("neon-shard-id", shard_id.clone());
-        }
-        req.metadata_mut()
-            .insert("neon-timeline-id", self.timeline_id.clone());
-        if let Some(auth_header) = &self.auth_header {
-            req.metadata_mut()
-                .insert("authorization", auth_header.clone());
-        }
-        Ok(req)
-    }
-}
--- a/pageserver/page_api/proto/page_service.proto
+++ b/pageserver/page_api/proto/page_service.proto
@@ -54,9 +54,9 @@ service PageService {
  // RPCs use regular unary requests, since they are not as frequent and
  // performance-critical, and this simplifies implementation.
  //
-  // NB: a gRPC status response (e.g. errors) will terminate the stream. The
-  // stream may be shared by multiple Postgres backends, so we avoid this by
-  // sending them as GetPageResponse.status_code instead.
+  // NB: a status response (e.g. errors) will terminate the stream. The stream
+  // may be shared by e.g. multiple Postgres backends, so we should avoid this.
+  // Most errors are therefore sent as GetPageResponse.status instead.
  rpc GetPages (stream GetPageRequest) returns (stream GetPageResponse);

  // Returns the size of a relation, as # of blocks.
@@ -159,8 +159,8 @@ message GetPageRequest {
 // A GetPageRequest class. Primarily intended for observability, but may also be
 // used for prioritization in the future.
 enum GetPageClass {
-  // Unknown class. For backwards compatibility: used when an older client version sends a class
-  // that a newer server version has removed.
+  // Unknown class. For forwards compatibility: used when the client sends a
+  // class that the server doesn't know about.
  GET_PAGE_CLASS_UNKNOWN = 0;
  // A normal request. This is the default.
  GET_PAGE_CLASS_NORMAL = 1;
@@ -180,37 +180,31 @@ message GetPageResponse {
  // The original request's ID.
  uint64 request_id = 1;
  // The response status code.
-  GetPageStatusCode status_code = 2;
+  GetPageStatus status = 2;
  // A string describing the status, if any.
  string reason = 3;
-  // The 8KB page images, in the same order as the request. Empty if status_code != OK.
+  // The 8KB page images, in the same order as the request. Empty if status != OK.
  repeated bytes page_image = 4;
 }

-// A GetPageResponse status code.
-//
-// These are effectively equivalent to gRPC statuses. However, we use a bidirectional stream
-// (potentially shared by many backends), and a gRPC status response would terminate the stream so
-// we send GetPageResponse messages with these codes instead.
-enum GetPageStatusCode {
-  // Unknown status. For forwards compatibility: used when an older client version receives a new
-  // status code from a newer server version.
-  GET_PAGE_STATUS_CODE_UNKNOWN = 0;
+// A GetPageResponse status code. Since we use a bidirectional stream, we don't
+// want to send errors as gRPC statuses, since this would terminate the stream.
+enum GetPageStatus {
+  // Unknown status. For forwards compatibility: used when the server sends a
+  // status code that the client doesn't know about.
+  GET_PAGE_STATUS_UNKNOWN = 0;
  // The request was successful.
-  GET_PAGE_STATUS_CODE_OK = 1;
+  GET_PAGE_STATUS_OK = 1;
  // The page did not exist. The tenant/timeline/shard has already been
  // validated during stream setup.
-  GET_PAGE_STATUS_CODE_NOT_FOUND = 2;
+  GET_PAGE_STATUS_NOT_FOUND = 2;
  // The request was invalid.
-  GET_PAGE_STATUS_CODE_INVALID_REQUEST = 3;
-  // The request failed due to an internal server error.
-  GET_PAGE_STATUS_CODE_INTERNAL_ERROR = 4;
+  GET_PAGE_STATUS_INVALID = 3;
  // The tenant is rate limited. Slow down and retry later.
-  GET_PAGE_STATUS_CODE_SLOW_DOWN = 5;
-  // NB: shutdown errors are emitted as a gRPC Unavailable status.
-  //
-  // TODO: consider adding a GET_PAGE_STATUS_CODE_LAYER_DOWNLOAD in the case of a layer download.
-  // This could free up the server task to process other requests while the download is in progress.
+  GET_PAGE_STATUS_SLOW_DOWN = 4;
+  // TODO: consider adding a GET_PAGE_STATUS_LAYER_DOWNLOAD in the case of a
+  // layer download. This could free up the server task to process other
+  // requests while the layer download is in progress.
 }

 // Fetches the size of a relation at a given LSN, as # of blocks. Only valid on
--- a/pageserver/page_api/src/lib.rs
+++ b/pageserver/page_api/src/lib.rs
@@ -18,6 +18,6 @@ pub mod proto {
    pub use page_service_server::{PageService, PageServiceServer};
 }

-pub mod model;
+mod model;

 pub use model::*;
--- a/pageserver/page_api/src/model.rs
+++ b/pageserver/page_api/src/model.rs
@@ -35,12 +35,6 @@ impl ProtocolError {
    }
 }

-impl From<ProtocolError> for tonic::Status {
-    fn from(err: ProtocolError) -> Self {
-        tonic::Status::invalid_argument(format!("{err}"))
-    }
-}
-
 /// The LSN a request should read at.
 #[derive(Clone, Copy, Debug)]
 pub struct ReadLsn {
@@ -102,15 +96,6 @@ impl TryFrom<ReadLsn> for proto::ReadLsn {
    }
 }

-impl From<&ReadLsn> for proto::ReadLsn {
-    fn from(value: &ReadLsn) -> proto::ReadLsn {
-        proto::ReadLsn {
-            request_lsn: value.request_lsn.into(),
-            not_modified_since_lsn: value.not_modified_since_lsn.unwrap_or_default().0,
-        }
-    }
-}
-
 // RelTag is defined in pageserver_api::reltag.
 pub type RelTag = pageserver_api::reltag::RelTag;

@@ -141,16 +126,6 @@ impl From<RelTag> for proto::RelTag {
    }
 }

-impl From<&RelTag> for proto::RelTag {
-    fn from(value: &RelTag) -> proto::RelTag {
-        proto::RelTag {
-            spc_oid: value.spcnode,
-            db_oid: value.dbnode,
-            rel_number: value.relnode,
-            fork_number: value.forknum as u32,
-        }
-    }
-}
 /// Checks whether a relation exists, at the given LSN. Only valid on shard 0, other shards error.
 #[derive(Clone, Copy, Debug)]
 pub struct CheckRelExistsRequest {
@@ -330,17 +305,6 @@ impl TryFrom<proto::GetPageRequest> for GetPageRequest {
    }
 }

-impl From<&GetPageRequest> for proto::GetPageRequest {
-    fn from(request: &GetPageRequest) -> proto::GetPageRequest {
-        proto::GetPageRequest {
-            request_id: request.request_id,
-            request_class: request.request_class.into(),
-            read_lsn: Some(request.read_lsn.try_into().unwrap()),
-            rel: Some(request.rel.into()),
-            block_number: request.block_numbers.clone().into_vec(),
-        }
-    }
-}
 impl TryFrom<GetPageRequest> for proto::GetPageRequest {
    type Error = ProtocolError;

@@ -364,7 +328,7 @@ pub type RequestID = u64;
 /// A GetPage request class.
 #[derive(Clone, Copy, Debug)]
 pub enum GetPageClass {
-    /// Unknown class. For backwards compatibility: used when an older client version sends a class
+    /// Unknown status. For backwards compatibility: used when an older client version sends a class
    /// that a newer server version has removed.
    Unknown,
    /// A normal request. This is the default.
@@ -422,7 +386,7 @@ pub struct GetPageResponse {
    /// The original request's ID.
    pub request_id: RequestID,
    /// The response status code.
-    pub status_code: GetPageStatusCode,
+    pub status: GetPageStatus,
    /// A string describing the status, if any.
    pub reason: Option<String>,
    /// The 8KB page images, in the same order as the request. Empty if status != OK.
@@ -433,7 +397,7 @@ impl From<proto::GetPageResponse> for GetPageResponse {
    fn from(pb: proto::GetPageResponse) -> Self {
        Self {
            request_id: pb.request_id,
-            status_code: pb.status_code.into(),
+            status: pb.status.into(),
            reason: Some(pb.reason).filter(|r| !r.is_empty()),
            page_images: pb.page_image.into(),
        }
@@ -444,20 +408,16 @@ impl From<GetPageResponse> for proto::GetPageResponse {
    fn from(response: GetPageResponse) -> Self {
        Self {
            request_id: response.request_id,
-            status_code: response.status_code.into(),
+            status: response.status.into(),
            reason: response.reason.unwrap_or_default(),
            page_image: response.page_images.into_vec(),
        }
    }
 }

-/// A GetPage response status code.
-///
-/// These are effectively equivalent to gRPC statuses. However, we use a bidirectional stream
-/// (potentially shared by many backends), and a gRPC status response would terminate the stream so
-/// we send GetPageResponse messages with these codes instead.
+/// A GetPage response status.
 #[derive(Clone, Copy, Debug)]
-pub enum GetPageStatusCode {
+pub enum GetPageStatus {
    /// Unknown status. For forwards compatibility: used when an older client version receives a new
    /// status code from a newer server version.
    Unknown,
@@ -467,50 +427,46 @@ pub enum GetPageStatusCode {
    /// setup.
    NotFound,
    /// The request was invalid.
-    InvalidRequest,
-    /// The request failed due to an internal server error.
-    InternalError,
+    Invalid,
    /// The tenant is rate limited. Slow down and retry later.
    SlowDown,
 }

-impl From<proto::GetPageStatusCode> for GetPageStatusCode {
-    fn from(pb: proto::GetPageStatusCode) -> Self {
+impl From<proto::GetPageStatus> for GetPageStatus {
+    fn from(pb: proto::GetPageStatus) -> Self {
        match pb {
-            proto::GetPageStatusCode::Unknown => Self::Unknown,
-            proto::GetPageStatusCode::Ok => Self::Ok,
-            proto::GetPageStatusCode::NotFound => Self::NotFound,
-            proto::GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
-            proto::GetPageStatusCode::InternalError => Self::InternalError,
-            proto::GetPageStatusCode::SlowDown => Self::SlowDown,
+            proto::GetPageStatus::Unknown => Self::Unknown,
+            proto::GetPageStatus::Ok => Self::Ok,
+            proto::GetPageStatus::NotFound => Self::NotFound,
+            proto::GetPageStatus::Invalid => Self::Invalid,
+            proto::GetPageStatus::SlowDown => Self::SlowDown,
        }
    }
 }

-impl From<i32> for GetPageStatusCode {
-    fn from(status_code: i32) -> Self {
-        proto::GetPageStatusCode::try_from(status_code)
-            .unwrap_or(proto::GetPageStatusCode::Unknown)
+impl From<i32> for GetPageStatus {
+    fn from(status: i32) -> Self {
+        proto::GetPageStatus::try_from(status)
+            .unwrap_or(proto::GetPageStatus::Unknown)
            .into()
    }
 }

-impl From<GetPageStatusCode> for proto::GetPageStatusCode {
-    fn from(status_code: GetPageStatusCode) -> Self {
-        match status_code {
-            GetPageStatusCode::Unknown => Self::Unknown,
-            GetPageStatusCode::Ok => Self::Ok,
-            GetPageStatusCode::NotFound => Self::NotFound,
-            GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
-            GetPageStatusCode::InternalError => Self::InternalError,
-            GetPageStatusCode::SlowDown => Self::SlowDown,
+impl From<GetPageStatus> for proto::GetPageStatus {
+    fn from(status: GetPageStatus) -> Self {
+        match status {
+            GetPageStatus::Unknown => Self::Unknown,
+            GetPageStatus::Ok => Self::Ok,
+            GetPageStatus::NotFound => Self::NotFound,
+            GetPageStatus::Invalid => Self::Invalid,
+            GetPageStatus::SlowDown => Self::SlowDown,
        }
    }
 }

-impl From<GetPageStatusCode> for i32 {
-    fn from(status_code: GetPageStatusCode) -> Self {
-        proto::GetPageStatusCode::from(status_code).into()
+impl From<GetPageStatus> for i32 {
+    fn from(status: GetPageStatus) -> Self {
+        proto::GetPageStatus::from(status).into()
    }
 }

--- a/pageserver/pagebench/Cargo.toml
+++ b/pageserver/pagebench/Cargo.toml
@@ -10,24 +10,19 @@ license.workspace = true
 anyhow.workspace = true
 camino.workspace = true
 clap.workspace = true
-thiserror.workspace = true
 futures.workspace = true
 hdrhistogram.workspace = true
 humantime.workspace = true
 humantime-serde.workspace = true
 rand.workspace = true
 reqwest.workspace=true
-bytes.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tracing.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
-async-trait = "0.1"

 pageserver_client.workspace = true
-pageserver_client_grpc.workspace = true
 pageserver_api.workspace = true
-pageserver_page_api.workspace = true
 utils = { path = "../../libs/utils/" }
 workspace_hack = { version = "0.1", path = "../../workspace_hack" }
--- a/pageserver/pagebench/src/cmd/getpage_latest_lsn.rs
+++ b/pageserver/pagebench/src/cmd/getpage_latest_lsn.rs
@@ -11,32 +11,20 @@ use camino::Utf8PathBuf;
 use pageserver_api::key::Key;
 use pageserver_api::keyspace::KeySpaceAccum;
 use pageserver_api::models::{PagestreamGetPageRequest, PagestreamRequest};
-use pageserver_page_api::model::{GetPageClass};
-use pageserver_client::page_service::PagestreamClient;
 use pageserver_api::shard::TenantShardId;
 use rand::prelude::*;
 use tokio::task::JoinSet;
 use tokio_util::sync::CancellationToken;
 use tracing::info;
 use utils::id::TenantTimelineId;
-use utils::id::TenantId;
-use utils::id::TimelineId;
 use utils::lsn::Lsn;

-
-
-use utils::shard::ShardIndex;
-use futures::{future::BoxFuture, stream::FuturesOrdered, FutureExt, StreamExt};
-
 use crate::util::tokio_thread_local_stats::AllThreadLocalStats;
 use crate::util::{request_stats, tokio_thread_local_stats};

 /// GetPage@LatestLSN, uniformly distributed across the compute-accessible keyspace.
 #[derive(clap::Parser)]
 pub(crate) struct Args {
-
-    #[clap(long, default_value = "false")]
-    grpc: bool,
    #[clap(long, default_value = "http://localhost:9898")]
    mgmt_api_endpoint: String,
    #[clap(long, default_value = "postgres://postgres@localhost:64000")]
@@ -130,7 +118,6 @@ struct Output {
    total: request_stats::Output,
 }

-
 tokio_thread_local_stats::declare!(STATS: request_stats::Stats);

 pub(crate) fn main(args: Args) -> anyhow::Result<()> {
@@ -316,20 +303,7 @@ async fn main_impl(
                .unwrap();

        Box::pin(async move {
-            if args.grpc {
-                let grpc = GrpcProtocol::new(
-                    args.page_service_connstring.clone(),
-                    worker_id.timeline.tenant_id,
-                    worker_id.timeline.timeline_id).await;
-                client_proto(args, grpc, worker_id, ss, cancel, rps_period, ranges, weights).await
-            } else {
-                let pg =  PgProtocol::new(
-                    args.page_service_connstring.clone(),
-                    worker_id.timeline.tenant_id,
-                    worker_id.timeline.timeline_id).await;
-                client_proto(args, pg, worker_id, ss, cancel, rps_period, ranges, weights).await
-            }
-
+            client_libpq(args, worker_id, ss, cancel, rps_period, ranges, weights).await
        })
    };

@@ -380,212 +354,9 @@ async fn main_impl(

    anyhow::Ok(())
 }
-// src/protocol.rs
-use async_trait::async_trait;
-use rand::distributions::weighted::WeightedIndex;

-// — your existing imports for PagestreamClient, PageserverClientError, KeyRange, etc. —
-
-/// Common interface for both Pg and Grpc versions.
-#[async_trait]
-trait Protocol: Send {
-    /// Constructor/factory.
-    async fn new(
-        conn_string: String,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> Self
-    where
-        Self: Sized;
-
-    /// Fire off a “get page” request and store the start time.
-    async fn add_to_inflight(
-        &mut self,
-        start: Instant,
-        args: &Args,
-        ranges: Vec<KeyRange>,
-        weights: WeightedIndex<i128>,
-    );
-
-    /// Wait for the next response and return its start time.
-    async fn get_start_time(&mut self) -> Instant;
-
-    /// How many in-flight requests do we have?
-    fn len(&self) -> usize;
-}
-
-///////////////////////////////////////////////////////////////////////////////
-// PgProtocol
-///////////////////////////////////////////////////////////////////////////////
-
-struct PgProtocol {
-    libpq_pagestream: PagestreamClient,
-    libpq_vector: VecDeque<Instant>,
-}
-
-#[async_trait]
-impl Protocol for PgProtocol {
-    async fn new(
-        conn_string: String,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> Self {
-        let client = pageserver_client::page_service::Client::new(conn_string)
-            .await
-            .unwrap()
-            .pagestream(tenant_id, timeline_id)
-            .await
-            .unwrap();
-        Self {
-            libpq_pagestream: client,
-            libpq_vector: VecDeque::new(),
-        }
-    }
-
-    async fn add_to_inflight(
-        &mut self,
-        start: Instant,
-        args: &Args,
-        ranges: Vec<KeyRange>,
-        weights: WeightedIndex<i128>,
-    ) {
-        // build your PagestreamGetPageRequest exactly as before…
-        let req = {
-            let mut rng = rand::thread_rng();
-            let r = &ranges[weights.sample(&mut rng)];
-            let key: i128 = rng.gen_range(r.start..r.end);
-            let key = Key::from_i128(key);
-            assert!(key.is_rel_block_key());
-            let (rel_tag, block_no) = key.to_rel_block().unwrap();
-            PagestreamGetPageRequest {
-                hdr: PagestreamRequest {
-                    reqid: 0,
-                    request_lsn: if rng.gen_bool(args.req_latest_probability) {
-                        Lsn::MAX
-                    } else {
-                        r.timeline_lsn
-                    },
-                    not_modified_since: r.timeline_lsn,
-                },
-                rel: rel_tag,
-                blkno: block_no,
-            }
-        };
-
-        let _ = self.libpq_pagestream.getpage_send(req).await;
-        self.libpq_vector.push_back(start);
-    }
-
-    async fn get_start_time(&mut self) -> Instant {
-        let start = self.libpq_vector.pop_front().unwrap();
-        let _ = self.libpq_pagestream.getpage_recv().await;
-        start
-    }
-
-    fn len(&self) -> usize {
-        self.libpq_vector.len()
-    }
-}
-
-///////////////////////////////////////////////////////////////////////////////
-// GrpcProtocol
-///////////////////////////////////////////////////////////////////////////////
-type GetPageFut = BoxFuture<'static, (Instant, Option<pageserver_client_grpc::PageserverClientError>)>;
-struct GrpcProtocol {
-    grpc_page_client: Arc<pageserver_client_grpc::PageserverClient>,
-    grpc_vector: FuturesOrdered<GetPageFut>,
-}
-
-#[async_trait]
-impl Protocol for GrpcProtocol {
-    async fn new(
-        conn_string: String,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> Self {
-        let shard_map = std::collections::HashMap::from([(
-            ShardIndex::unsharded(),
-            conn_string.clone(),
-        )]);
-        let client = pageserver_client_grpc::PageserverClient::new(
-            &tenant_id.to_string(),
-            &timeline_id.to_string(),
-            &None,
-            shard_map,
-        );
-        Self {
-            grpc_page_client: Arc::new(client),
-            grpc_vector: FuturesOrdered::new(),
-        }
-    }
-
-    async fn add_to_inflight(
-        &mut self,
-        start: Instant,
-        args: &Args,
-        ranges: Vec<KeyRange>,
-        weights: WeightedIndex<i128>,
-    ) {
-        // build your GetPageRequest exactly as before…
-        let req = {
-            let mut rng = rand::thread_rng();
-            let r = &ranges[weights.sample(&mut rng)];
-            let key: i128 = rng.gen_range(r.start..r.end);
-            let key = Key::from_i128(key);
-            assert!(key.is_rel_block_key());
-            let (rel_tag, block_no) = key.to_rel_block().unwrap();
-            pageserver_page_api::model::GetPageRequest {
-                request_id: 0,
-                request_class: GetPageClass::Normal,
-                read_lsn: pageserver_page_api::model::ReadLsn {
-                    request_lsn: if rng.gen_bool(args.req_latest_probability) {
-                        Lsn::MAX
-                    } else {
-                        r.timeline_lsn
-                    },
-                    not_modified_since_lsn: Some(r.timeline_lsn),
-                },
-                rel: pageserver_page_api::model::RelTag {
-                    spcnode: rel_tag.spcnode,
-                    dbnode: rel_tag.dbnode,
-                    relnode: rel_tag.relnode,
-                    forknum: rel_tag.forknum,
-                },
-                block_numbers: vec![block_no].into(),
-            }
-        };
-
-        let client_clone = self.grpc_page_client.clone();
-        let getpage_fut : GetPageFut = async move {
-            let result = client_clone.get_page(&req).await;
-            match result {
-                Ok(_) => {
-                    (start, None)
-                }
-                Err(e) => {
-                    (start, Some(e))
-                }
-            }
-        }.boxed();
-        self.grpc_vector.push_back(getpage_fut);
-    }
-
-    async fn get_start_time(&mut self) -> Instant {
-        let (start, err) = self.grpc_vector.next().await.unwrap();
-        if let Some(e) = err {
-            tracing::error!("getpage request failed: {e}");
-        }
-        start
-    }
-
-    fn len(&self) -> usize {
-        self.grpc_vector.len()
-    }
-}
-
-async fn client_proto(
+async fn client_libpq(
    args: &Args,
-    mut protocol: impl Protocol,
    worker_id: WorkerId,
    shared_state: Arc<SharedState>,
    cancel: CancellationToken,
@@ -593,11 +364,18 @@ async fn client_proto(
    ranges: Vec<KeyRange>,
    weights: rand::distributions::weighted::WeightedIndex<i128>,
 ) {
-
+    let client = pageserver_client::page_service::Client::new(args.page_service_connstring.clone())
+        .await
+        .unwrap();
+    let mut client = client
+        .pagestream(worker_id.timeline.tenant_id, worker_id.timeline.timeline_id)
+        .await
+        .unwrap();

    shared_state.start_work_barrier.wait().await;
    let client_start = Instant::now();
    let mut ticks_processed = 0;
+    let mut inflight = VecDeque::new();
    while !cancel.is_cancelled() {
        // Detect if a request took longer than the RPS rate
        if let Some(period) = &rps_period {
@@ -612,12 +390,37 @@ async fn client_proto(
            ticks_processed = periods_passed_until_now;
        }

-        while protocol.len() < args.queue_depth.get() {
+        while inflight.len() < args.queue_depth.get() {
            let start = Instant::now();
-            protocol.add_to_inflight(start, args, ranges.clone(), weights.clone()).await;
+            let req = {
+                let mut rng = rand::thread_rng();
+                let r = &ranges[weights.sample(&mut rng)];
+                let key: i128 = rng.gen_range(r.start..r.end);
+                let key = Key::from_i128(key);
+                assert!(key.is_rel_block_key());
+                let (rel_tag, block_no) = key
+                    .to_rel_block()
+                    .expect("we filter non-rel-block keys out above");
+                PagestreamGetPageRequest {
+                    hdr: PagestreamRequest {
+                        reqid: 0,
+                        request_lsn: if rng.gen_bool(args.req_latest_probability) {
+                            Lsn::MAX
+                        } else {
+                            r.timeline_lsn
+                        },
+                        not_modified_since: r.timeline_lsn,
+                    },
+                    rel: rel_tag,
+                    blkno: block_no,
+                }
+            };
+            client.getpage_send(req).await.unwrap();
+            inflight.push_back(start);
        }

-        let start = protocol.get_start_time().await;
+        let start = inflight.pop_front().unwrap();
+        client.getpage_recv().await.unwrap();
        let end = Instant::now();
        shared_state.live_stats.request_done();
        ticks_processed += 1;
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -7520,11 +7520,15 @@ fn is_send() {

 #[cfg(test)]
 mod tests {
+    use std::io::Cursor;
+    use std::path::PathBuf;
    use std::sync::Arc;

    use pageserver_api::key::Key;
    use pageserver_api::value::Value;
+    use postgres_ffi::v17::bindings::PageHeaderData;
    use std::iter::Iterator;
+    use tokio_epoll_uring::BoundedBuf;
    use tracing::Instrument;
    use utils::id::TimelineId;
    use utils::lsn::Lsn;
@@ -7836,4 +7840,126 @@ mod tests {

        layers.get_from_desc(&desc)
    }
+
+    fn fixture_path(relative: &str) -> PathBuf {
+        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(relative)
+    }
+
+    use postgres_ffi::ItemIdData;
+    use postgres_ffi::LocationIndex;
+    use postgres_ffi::PageXLogRecPtr;
+    use postgres_ffi::TransactionId;
+    /// We already have a PageHeaderData via bindgen, but it's not serializable and
+    /// contains a non-serializable __IncompleteArrayField.
+    /// FIXME: we should be able to exclude pd_linp from the bindgen, but blocklist_item()
+    /// wasn't working for some reason.
+    use serde::{Deserialize, Serialize};
+
+    #[repr(C)]
+    #[derive(Debug, Default, Serialize, Deserialize)]
+    pub struct SerPageHeaderData {
+        pub pd_lsn: PageXLogRecPtr,
+        pub pd_checksum: postgres_ffi::uint16,
+        pub pd_flags: postgres_ffi::uint16,
+        pub pd_lower: LocationIndex,
+        pub pd_upper: LocationIndex,
+        pub pd_special: LocationIndex,
+        pub pd_pagesize_version: postgres_ffi::uint16,
+        pub pd_prune_xid: TransactionId,
+    }
+
+    #[repr(C)]
+    #[repr(align(4))]
+    #[derive(Debug, Copy, Clone, Deserialize)]
+    pub struct SerItemIdData {
+        pub raw: [u8; 4],
+    }
+
+    /// Postgres HeapTupleHeaderData type is actually some kind of placeholder,
+    /// so we have a hand-rolled struct to describe the actual serialized header of a tuple
+    #[repr(C)]
+    #[derive(Debug, Default, Serialize, Deserialize)]
+    struct SerMinimalTupleData {
+        t_xmin: postgres_ffi::uint32, /* actual length of minimal tuple */
+        t_xmax: postgres_ffi::uint32, /* actual length of minimal tuple */
+        t_cid_or_t_xvac: postgres_ffi::uint32,
+        t_ctid: postgres_ffi::ItemPointerData,
+        t_infomask2: postgres_ffi::uint16, /* number of attributes + various flags */
+        t_infomask: postgres_ffi::uint16,  /* various flag bits, see below */
+
+        t_hoff: postgres_ffi::uint8, /* sizeof header incl. bitmap, padding */
+
+                                     /* ^ - 23 bytes - ^ */
+    }
+
+    #[test]
+    fn decode_tuples() {
+        use utils::bin_ser::LeSer;
+
+        //assert_eq!(std::mem::size_of::<SerMinimalTupleData>(), 23);
+
+        // A simple example with three valid tuples, each tuple is (id, string, string, timestamp)
+        let input_raw = std::fs::read(fixture_path("test_data/example_page.bin")).unwrap();
+        // 00000000  00 00 00 00 a0 3d 53 01  00 00 00 00 24 00 70 1f  |.....=S.....$.p.|
+        // 00000010  00 20 04 20 00 00 00 00  d0 9f 60 00 a0 9f 60 00  |. . ......`...`.|
+        // 00000020  70 9f 60 00 00 00 00 00  00 00 00 00 00 00 00 00  |p.`.............|
+        // 00000030  00 00 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // *
+        // 00001f70  e7 02 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // 00001f80  03 00 04 00 02 08 18 00  03 00 00 00 0b 6b 65 79  |.............key|
+        // 00001f90  31 0f 76 61 6c 75 65 31  9f 2a e0 69 79 d8 02 00  |1.value1.*.iy...|
+        // 00001fa0  e6 02 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // 00001fb0  02 00 04 00 02 08 18 00  02 00 00 00 0b 6b 65 79  |.............key|
+        // 00001fc0  33 0f 76 61 6c 75 65 33  54 b9 a3 69 79 d8 02 00  |3.value3T..iy...|
+        // 00001fd0  e5 02 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
+        // 00001fe0  01 00 04 00 02 08 18 00  01 00 00 00 0b 6b 65 79  |.............key|
+        // 00001ff0  32 0f 76 61 6c 75 65 32  dd 0e 6b 68 79 d8 02 00  |2.value2..khy...|
+        // 00002000
+
+        // The attributes for this example:
+        // attrelid |   attname   | atttypid | attlen | attnum | attcacheoff | atttypmod | attndims | attbyval | attalign | attstorage | attcompression | attnotnull | atthasdef | atthasmissing | attidentity | attgenerated | attisdropped | attislocal | attinhcount | attcollation | attstattarget | attacl | attoptions | attfdwoptions | attmissingval
+        // ----------+-------------+----------+--------+--------+-------------+-----------+----------+----------+----------+------------+----------------+------------+-----------+---------------+-------------+--------------+--------------+------------+-------------+--------------+---------------+--------+------------+---------------+---------------
+        //     16385 | id          |       23 |      4 |      1 |          -1 |        -1 |        0 | t        | i        | p          |                | t          | t         | f             |             |              | f            | t          |           0 |            0 |               |        |            |               |
+        //     16385 | key         |     1043 |     -1 |      2 |          -1 |       259 |        0 | f        | i        | x          |                | f          | f         | f             |             |              | f            | t          |           0 |          100 |               |        |            |               |
+        //     16385 | value       |       25 |     -1 |      3 |          -1 |        -1 |        0 | f        | i        | x          |                | f          | f         | f             |             |              | f            | t          |           0 |          100 |               |        |            |               |
+        //     16385 | modified_at |     1184 |      8 |      4 |          -1 |        -1 |        0 | t        | d        | p          |                | f          | t         | f             |             |              | f            | t          |           0 |            0 |               |        |            |               |
+
+        // TODO: test cases for schema changes:
+        // 1. baseline: data that exactly matches attributes
+        // 2. add one column, write a row, read all rows
+        // 3. drop one column, write a row, read all rows.
+        // 4. vacuum full, write a row, read all rows.
+
+        //        let mut cursor = Cursor::new(input_raw);
+        let header =
+            SerPageHeaderData::des(&input_raw[0..std::mem::size_of::<SerPageHeaderData>()])
+                .unwrap();
+
+        let item_count = (header.pd_lower as usize - std::mem::size_of::<SerPageHeaderData>())
+            / std::mem::size_of::<ItemIdData>();
+
+        assert_eq!(item_count, 3);
+
+        for item_counter in 0..item_count {
+            let item_base = std::mem::size_of::<SerPageHeaderData>()
+                + item_counter * std::mem::size_of::<ItemIdData>();
+            let item = SerItemIdData::des(
+                &input_raw[item_base..item_base + std::mem::size_of::<ItemIdData>()],
+            )
+            .unwrap();
+
+            let item = unsafe { std::mem::transmute::<[u8; 4], ItemIdData>(item.raw) };
+
+            println!("item: {} {}", item.lp_off(), item.lp_len());
+
+            let tuple_header_bytes =
+                &input_raw[item.lp_off() as usize..item.lp_off() as usize + 23];
+            let tuple_header = SerMinimalTupleData::des(tuple_header_bytes).unwrap();
+
+            println!("tuple_header: insert offset {}", tuple_header.t_hoff);
+
+            let tuple_raw = &input_raw[item.lp_off() as usize + tuple_header.t_hoff as usize
+                ..item.lp_off() as usize + item.lp_len() as usize];
+        }
+    }
 }
--- a/pageserver/test_data/example_page.bin
+++ b/pageserver/test_data/example_page.bin
--- a/test_runner/regress/test_storage_controller.py
+++ b/test_runner/regress/test_storage_controller.py
@@ -4158,12 +4158,17 @@ def test_storcon_create_delete_sk_down(
        env.storage_controller.stop()
        env.storage_controller.start()

-    with env.endpoints.create("main", tenant_id=tenant_id) as ep:
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    with env.endpoints.create("main", tenant_id=tenant_id, config_lines=config_lines) as ep:
        # endpoint should start.
        ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])
        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")

-    with env.endpoints.create("child_of_main", tenant_id=tenant_id) as ep:
+    with env.endpoints.create(
+        "child_of_main", tenant_id=tenant_id, config_lines=config_lines
+    ) as ep:
        # endpoint should start.
        ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])
        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
@@ -4244,12 +4249,17 @@ def test_storcon_few_sk(

    env.safekeepers[0].assert_log_contains(f"creating new timeline {tenant_id}/{timeline_id}")

-    with env.endpoints.create("main", tenant_id=tenant_id) as ep:
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    with env.endpoints.create("main", tenant_id=tenant_id, config_lines=config_lines) as ep:
        # endpoint should start.
        ep.start(safekeeper_generation=1, safekeepers=safekeeper_list)
        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")

-    with env.endpoints.create("child_of_main", tenant_id=tenant_id) as ep:
+    with env.endpoints.create(
+        "child_of_main", tenant_id=tenant_id, config_lines=config_lines
+    ) as ep:
        # endpoint should start.
        ep.start(safekeeper_generation=1, safekeepers=safekeeper_list)
        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
--- a/test_runner/regress/test_wal_acceptor.py
+++ b/test_runner/regress/test_wal_acceptor.py
@@ -2012,7 +2012,10 @@ def test_explicit_timeline_creation(neon_env_builder: NeonEnvBuilder):
    tenant_id = env.initial_tenant
    timeline_id = env.initial_timeline

-    ep = env.endpoints.create("main")
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    ep = env.endpoints.create("main", config_lines=config_lines)

    # expected to fail because timeline is not created on safekeepers
    with pytest.raises(Exception, match=r".*timed out.*"):
@@ -2040,7 +2043,10 @@ def test_explicit_timeline_creation_storcon(neon_env_builder: NeonEnvBuilder):
    }
    env = neon_env_builder.init_start()

-    ep = env.endpoints.create("main")
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    ep = env.endpoints.create("main", config_lines=config_lines)

    # endpoint should start.
    ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])
--- a/test_runner/regress/test_wal_acceptor_async.py
+++ b/test_runner/regress/test_wal_acceptor_async.py
@@ -637,7 +637,10 @@ async def quorum_sanity_single(
    # create timeline on `members_sks`
    Safekeeper.create_timeline(tenant_id, timeline_id, env.pageservers[0], mconf, members_sks)

-    ep = env.endpoints.create(branch_name)
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    ep = env.endpoints.create(branch_name, config_lines=config_lines)
    ep.start(safekeeper_generation=1, safekeepers=compute_sks_ids)
    ep.safe_psql("create table t(key int, value text)")
Author	SHA1	Message	Date
John Spray	52e98de0bc	Experiment in decoding postgres tuples	2025-05-27 12:16:11 +02:00
John Spray	53286d8880	postgres_ffi: expose HeapTupleHeaderData	2025-05-27 12:16:04 +02:00