Merge pull request #23519 from rockwotj/parquet-metadata

Author: rockwotj

.bazelrc

@@ -21,6 +21,9 @@ build:system-clang-17 --action_env=CXX=clang++-17 --host_action_env=CXX=clang++-
 build:system-clang-18 --config=system-clang
 build:system-clang-18 --action_env=CC=clang-18 --host_action_env=CC=clang-18
 build:system-clang-18 --action_env=CXX=clang++-18 --host_action_env=CXX=clang++-18
+build:system-clang-19 --config=system-clang
+build:system-clang-19 --action_env=CC=clang-19 --host_action_env=CC=clang-19
+build:system-clang-19 --action_env=CXX=clang++-19 --host_action_env=CXX=clang++-19
 
 # https://github.com/bazelbuild/rules_foreign_cc/issues/1065
 # https://github.com/bazelbuild/rules_foreign_cc/issues/1186#issuecomment-2053550487

MODULE.bazel

@@ -189,13 +189,15 @@ use_repo(
     "com_github_hamba_avro_v2",
     "com_github_hashicorp_go_multierror",
     "com_github_kballard_go_shellquote",
+    "com_github_kr_pretty",
     "com_github_lestrrat_go_jwx",
     "com_github_linkedin_goavro_v2",
     "com_github_lorenzosaino_go_sysctl",
     "com_github_mattn_go_isatty",
     "com_github_moby_term",
     "com_github_opencontainers_go_digest",
     "com_github_opencontainers_image_spec",
+    "com_github_parquet_go_parquet_go",
     "com_github_pkg_browser",
     "com_github_pkg_errors",
     "com_github_prometheus_client_model",
@@ -206,6 +208,7 @@ use_repo(
     "com_github_safchain_ethtool",
     "com_github_santhosh_tekuri_jsonschema_v6",
     "com_github_schollz_progressbar_v3",
+    "com_github_segmentio_encoding",
     "com_github_spf13_afero",
     "com_github_spf13_cobra",
     "com_github_spf13_pflag",

bazel/thirdparty/go.work

@@ -4,7 +4,7 @@ go 1.23.0
 // Everything we build with Bazel goes here.
 use (
 	src/go/rpk
-	src/go/kreq-gen
+	src/v/test_utils/go
 	src/transform-sdk/go/transform
 	src/transform-sdk/go/transform/internal/testdata
 	src/transform-sdk/tests

src/CMakeLists.txt

@@ -7,5 +7,4 @@ find_package(Boost REQUIRED
 find_package(absl REQUIRED)
 find_program(GO_PROGRAM go REQUIRED)
 add_subdirectory(v)
-add_subdirectory(go/kreq-gen)
 add_subdirectory(transform-sdk/go/transform/internal/testdata)

src/go/kreq-gen/go.mod

@@ -91,7 +91,7 @@ class bytes {
     bool empty() const noexcept { return data_.empty(); }
 
     void resize(size_type size) { data_.resize(size); }
-
+    void reserve(size_type size) { data_.reserve(size); }
     void push_back(value_type v) { data_.push_back(v); }
 
     friend bool operator==(const bytes&, const bytes&) = default;

src/go/kreq-gen/go.sum

@@ -52,21 +52,24 @@ class fragmented_vector {
 
     // calculate the maximum number of elements per fragment while
     // keeping the element count a power of two
-    static constexpr size_t calc_elems_per_frag(size_t esize) {
-        size_t max = fragment_size_bytes / esize;
+    static consteval size_t calc_elems_per_frag() {
+        size_t max = fragment_size_bytes / sizeof(T);
         if constexpr (is_chunked_vector) {
-            max = max_allocation_size / esize;
+            max = max_allocation_size / sizeof(T);
         }
-        assert(max > 0);
         return std::bit_floor(max);
     }
 
-    static constexpr size_t elems_per_frag = calc_elems_per_frag(sizeof(T));
-
-    static_assert(
-      (elems_per_frag & (elems_per_frag - 1)) == 0,
-      "element count per fragment must be a power of 2");
-    static_assert(elems_per_frag >= 1);
+    /**
+     * The maximum number of bytes per fragment as specified in
+     * as part of the type. Note that for most types, the true
+     * number of bytes in a full fragment may as low as half
+     * of this amount (+1) since the number of elements is restricted
+     * to a power of two.
+     */
+    static consteval size_t calc_max_frag_bytes() {
+        return calc_elems_per_frag() * sizeof(T);
+    }
 
 public:
     using this_type = fragmented_vector<T, fragment_size_bytes>;
@@ -81,19 +84,6 @@ class fragmented_vector {
     using pointer = T*;
     using const_pointer = const T*;
 
-    /**
-     * The maximum number of bytes per fragment as specified in
-     * as part of the type. Note that for most types, the true
-     * number of bytes in a full fragment may as low as half
-     * of this amount (+1) since the number of elements is restricted
-     * to a power of two.
-     */
-    static constexpr size_t max_frag_bytes = elems_per_frag * sizeof(T);
-
-    static_assert(
-      max_frag_bytes <= max_allocation_size,
-      "max size of a fragment must be <= 128KiB");
-
     fragmented_vector() noexcept = default;
     explicit fragmented_vector(allocator_type alloc)
       : _frags(alloc) {}
@@ -180,7 +170,7 @@ class fragmented_vector {
         --_size;
         if (_frags.back().empty()) {
             _frags.pop_back();
-            _capacity -= std::min(elems_per_frag, _capacity);
+            _capacity -= std::min(calc_elems_per_frag(), _capacity);
         }
         update_generation();
     }
@@ -203,7 +193,7 @@ class fragmented_vector {
         while (n >= _frags.back().size()) {
             n -= _frags.back().size();
             _frags.pop_back();
-            _capacity -= elems_per_frag;
+            _capacity -= calc_elems_per_frag();
         }
 
         for (size_t i = 0; i < n; ++i) {
@@ -213,18 +203,22 @@ class fragmented_vector {
     }
 
     const_reference at(size_t index) const {
+        static constexpr size_t elems_per_frag = calc_elems_per_frag();
         return _frags.at(index / elems_per_frag).at(index % elems_per_frag);
     }
 
     reference at(size_t index) {
+        static constexpr size_t elems_per_frag = calc_elems_per_frag();
         return _frags.at(index / elems_per_frag).at(index % elems_per_frag);
     }
 
     const_reference operator[](size_t index) const {
+        static constexpr size_t elems_per_frag = calc_elems_per_frag();
         return _frags[index / elems_per_frag][index % elems_per_frag];
     }
 
     reference operator[](size_t index) {
+        static constexpr size_t elems_per_frag = calc_elems_per_frag();
         return _frags[index / elems_per_frag][index % elems_per_frag];
     }
 
@@ -275,6 +269,7 @@ class fragmented_vector {
         // For fixed size fragments we noop, as we already reserve the full size
         // of vector
         if constexpr (is_chunked_vector) {
+            static constexpr size_t elems_per_frag = calc_elems_per_frag();
             if (new_cap > _capacity) {
                 if (_frags.empty()) {
                     auto& frag = _frags.emplace_back();
@@ -310,7 +305,11 @@ class fragmented_vector {
     /**
      * Returns the (maximum) number of elements in each fragment of this vector.
      */
-    static constexpr size_t elements_per_fragment() { return elems_per_frag; }
+    static constexpr size_t elements_per_fragment() {
+        return calc_elems_per_frag();
+    }
+
+    static constexpr size_t max_frag_bytes() { return calc_max_frag_bytes(); }
 
     /**
      * Remove all elements from the vector.
@@ -499,6 +498,10 @@ class fragmented_vector {
     }
 
     void add_capacity() {
+        static constexpr size_t elems_per_frag = calc_elems_per_frag();
+        static_assert(
+          calc_max_frag_bytes() <= max_allocation_size,
+          "max size of a fragment must be <= 128KiB");
         if constexpr (is_chunked_vector) {
             if (
               _frags.size() == 1 && _frags.back().capacity() < elems_per_frag) {
@@ -510,7 +513,8 @@ class fragmented_vector {
             } else if (_frags.empty()) {
                 // At least one element or 32 bytes worth of elements for small
                 // items.
-                constexpr size_t initial_cap = std::max(1UL, 32UL / sizeof(T));
+                static constexpr size_t initial_cap = std::max(
+                  1UL, 32UL / sizeof(T));
                 _capacity = initial_cap;
                 _frags.emplace_back(_frags.get_allocator()).reserve(_capacity);
                 return;

src/v/bytes/bytes.h

@@ -60,15 +60,15 @@ class fragmented_vector_validator {
             calc_cap += f.capacity();
 
             if (i + 1 < v._frags.size()) {
-                if (f.size() < v.elems_per_frag) {
+                if (f.size() < v.elements_per_fragment()) {
                     return AssertionFailure() << fmt::format(
                              "fragment {} is undersized ({} < {})",
                              i,
                              f.size(),
-                             v.elems_per_frag);
+                             v.elements_per_fragment());
                 }
             }
-            if (f.capacity() > std::decay_t<decltype(v)>::max_frag_bytes) {
+            if (f.capacity() > std::decay_t<decltype(v)>::max_frag_bytes()) {
                 return AssertionFailure() << fmt::format(
                          "fragment {} capacity over max_frag_bytes ({})",
                          i,

src/v/container/fragmented_vector.h

@@ -8,11 +8,11 @@ redpanda_cc_btest(
         "protocol_test.cc",
     ],
     data = [
-        "//src/go/kreq-gen:kafka_request_generator",
+        "//src/v/test_utils/go/kreq-gen:kafka_request_generator",
     ],
     # could also use runfiles here, which would be easier once cmake is gone
     # because then we could use bazel_tools runfiles library
-    env = {"GENERATOR_BIN": "$(rootpath //src/go/kreq-gen:kafka_request_generator)"},
+    env = {"GENERATOR_BIN": "$(rootpath //src/v/test_utils/go/kreq-gen:kafka_request_generator)"},
     deps = [
         "//src/v/kafka/protocol",
         "//src/v/test_utils:seastar_boost",

src/v/container/tests/fragmented_vector_test.cc

@@ -1,4 +1,4 @@
-set(KAFKA_REQUEST_GENERATOR "${CMAKE_BINARY_DIR}/src/go/kreq-gen/kafka-request-generator")
+set(KAFKA_REQUEST_GENERATOR "${CMAKE_BINARY_DIR}/src/v/test_utils/go/kreq-gen/kafka-request-generator")
 
 rp_test(
   UNIT_TEST

src/v/kafka/protocol/tests/BUILD

@@ -619,6 +619,7 @@ metadata_memory_estimator(size_t request_size, connection_context& conn_ctx) {
     // generally ~8000 bytes). Finally, we add max_frag_bytes to account for the
     // worse-cast overshoot during vector re-allocation.
     return default_memory_estimate(request_size) + size_estimate
-           + large_fragment_vector<metadata_response_partition>::max_frag_bytes;
+           + large_fragment_vector<
+             metadata_response_partition>::max_frag_bytes();
 }
 } // namespace kafka

src/v/kafka/protocol/tests/CMakeLists.txt

@@ -0,0 +1,45 @@
+load("//bazel:build.bzl", "redpanda_cc_library")
+
+package(
+    default_visibility = ["//src/v/serde/parquet:__subpackages__"],
+)
+
+redpanda_cc_library(
+    name = "schema",
+    srcs = [
+        "flattened_schema.cc",
+    ],
+    hdrs = [
+        "flattened_schema.h",
+        "schema.h",
+    ],
+    include_prefix = "serde/parquet",
+    deps = [
+        "//src/v/base",
+        "//src/v/container:fragmented_vector",
+        "//src/v/utils:uuid",
+        "@seastar",
+    ],
+)
+
+redpanda_cc_library(
+    name = "metadata",
+    srcs = [
+        "metadata.cc",
+    ],
+    hdrs = [
+        "metadata.h",
+    ],
+    include_prefix = "serde/parquet",
+    deps = [
+        ":schema",
+        "//src/v/base",
+        "//src/v/bytes:iobuf",
+        "//src/v/container:fragmented_vector",
+        "//src/v/serde/thrift:compact",
+        "//src/v/utils:uuid",
+        "//src/v/utils:vint",
+        "@abseil-cpp//absl/container:flat_hash_map",
+        "@seastar",
+    ],
+)

src/v/kafka/server/handlers/metadata.cc

@@ -0,0 +1,11 @@
+
+v_cc_library(
+  NAME serde_parquet
+  SRCS
+    metadata.cc
+  DEPS
+    Seastar::seastar
+    v::bytes
+    v::container
+    v::utils
+  )

src/v/serde/parquet/BUILD

@@ -0,0 +1,20 @@
+# Parquet Library
+
+Here lies a library to be able to write parquet files for Redpanda's Iceberg integration.
+Due to Redpanda's usage of Seastar off the shelf preexisting parquet libraries do not meet
+our strict requirements, imposed by our userland task scheduler and virtual memory avoiding 
+allocator.
+
+
+### Metadata
+
+Parquet metadata is serialized using [Apache Thrift's compact wire format][thrift-compact-format].
+
+We use metadata that is the logical representation of what our application needs, then we write out
+the wire format with all the deprecated and legacy types to be compatible with legacy query systems.
+
+The physical format of serialized parquet metadata is documented [here][parquet-thrift].
+
+
+[parquet-thrift]: https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift
+[thrift-compact-format]: https://github.com/apache/thrift/blob/master/doc/specs/thrift-compact-protocol.md

src/v/serde/parquet/CMakeLists.txt

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2024 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#include "serde/parquet/flattened_schema.h"
+
+namespace serde::parquet {
+
+chunked_vector<flattened_schema> flatten(const schema_element& root) {
+    chunked_vector<flattened_schema> flattened;
+    root.for_each([&flattened](const schema_element& elem) {
+        flattened.emplace_back(
+          elem.type,
+          elem.repetition_type,
+          elem.name,
+          elem.children.size(),
+          elem.field_id,
+          elem.logical_type);
+    });
+    return flattened;
+}
+
+} // namespace serde::parquet