serde/parquet: account stats collector memory in the writer

column_writer::memory_usage() drives the translator's memory reservation
but counted only buffered page data. Each leaf column also retains the
row group's min/max bounds untruncated, which for byte-array columns
grows with the value width.

Author: andrwng

src/v/serde/parquet/column_stats_collector.h

@@ -106,10 +106,25 @@ class column_stats_collector {
 
     int64_t null_count() const { return _null_count; }
 
+    // Byte-array bounds are retained untruncated.
+    int64_t memory_usage() const {
+        return bound_bytes(_min) + bound_bytes(_max);
+    }
+
     bound_ref_type min() { return normalize(_min, true); }
     bound_ref_type max() { return normalize(_max, false); }
 
 private:
+    static int64_t bound_bytes(const std::optional<value_type>& bound) {
+        if constexpr (std::is_trivially_copyable_v<value_type>) {
+            return 0;
+        } else {
+            return bound.has_value()
+                     ? static_cast<int64_t>(bound->val.size_bytes())
+                     : 0;
+        }
+    }
+
     bound_ref_type normalize(bound_ref_type v, bool min) {
         if constexpr (std::is_floating_point_v<decltype(v->val)>) {
             // min floats are always written as -0 and max as 0

src/v/serde/parquet/column_writer.cc

@@ -314,7 +314,14 @@ class buffered_column_writer final : public column_writer::impl {
     }
 
     int64_t memory_usage() const override {
-        return _total_memory_usage + current_page_memory_usage();
+        // Memory that flushing will release.
+        //
+        // NOTE: the translator flushes to recover from memory pressure, so a
+        // term it cannot free does not belong here: _file_stats lives until
+        // close and is not accounted here.
+        return _total_memory_usage + current_page_memory_usage()
+               + _current_page_stats.memory_usage()
+               + _flushed_stats.memory_usage();
     }
 
     int64_t current_page_memory_usage() const override {

src/v/serde/parquet/tests/column_stats_collector_test.cc

@@ -18,6 +18,7 @@
 #include <cmath>
 #include <limits>
 #include <optional>
+#include <string>
 
 using namespace serde::parquet;
 using testing::DoubleEq;
@@ -64,6 +65,31 @@ TEST(ColumnStatsCollector, FloatingPoint) {
     EXPECT_THAT(collector.max(), DoubleBound(IsPositiveZero()));
 }
 
+TEST(ColumnStatsCollector, MemoryUsage) {
+    // Fixed-width columns hold their bounds inline: no heap, regardless of the
+    // values recorded.
+    column_stats_collector<int64_value, ordering::int64> numeric;
+    EXPECT_EQ(numeric.memory_usage(), 0);
+    numeric.record_value(int64_value{42});
+    numeric.record_value(int64_value{7});
+    EXPECT_EQ(numeric.memory_usage(), 0);
+
+    // Byte-array columns retain the untruncated min and max values, so the
+    // footprint tracks the size of those bounds.
+    column_stats_collector<byte_array_value, ordering::byte_array> binary;
+    EXPECT_EQ(binary.memory_usage(), 0);
+    auto cat = byte_array_value{iobuf::from("cat")};
+    binary.record_value(cat);
+    // min == max == "cat": 3 bytes each.
+    EXPECT_EQ(binary.memory_usage(), 6);
+    auto big = byte_array_value{iobuf::from(std::string(4096, 'z'))};
+    binary.record_value(big);
+    // min stays "cat" (3 bytes), max is now the 4096-byte value.
+    EXPECT_EQ(binary.memory_usage(), 3 + 4096);
+    binary.reset();
+    EXPECT_EQ(binary.memory_usage(), 0);
+}
+
 TEST(ColumnStatsCollector, Binary) {
     column_stats_collector<byte_array_value, ordering::byte_array> collector;
     auto empty = byte_array_value{iobuf::from("")};

src/v/serde/parquet/tests/writer_test.cc

@@ -259,6 +259,19 @@ TEST(ParquetWriter, FlushesRowGroupWhenSizeExceeded) {
     w.close().get();
 }
 
+TEST(ParquetWriter, BufferedMemoryZeroAfterFlush) {
+    iobuf file;
+    writer w({.schema = simple_schema()}, make_iobuf_ref_output_stream(file));
+    w.init().get();
+    for (size_t i = 0; i < 3; ++i) {
+        w.write_row(make_row(/*data_size=*/64)).get();
+    }
+    EXPECT_GT(w.stats().buffered_size, 0);
+    w.flush_row_group().get();
+    EXPECT_EQ(w.stats().buffered_size, 0);
+    w.close().get();
+}
+
 TEST(ParquetWriter, ColumnMemoryEstimateCoversActual) {
 #ifdef SEASTAR_DEFAULT_ALLOCATOR
     GTEST_SKIP() << "memory::stats() reports fixed values under the system "