[v24.3.x] Use Redpanda implementation of Parquet by default

src/v/datalake/BUILD

@@ -163,37 +163,65 @@ redpanda_cc_library(
     ],
 )
 
-# TODO(iceberg): uncomment once partition_translator
-# and dependencies are bazel friendly.
-#redpanda_cc_library(
-#    name = "manager",
-#    srcs = [
-#        "datalake_manager.cc",
-#    ],
-#    hdrs = [
-#        "datalake_manager.h",
-#    ],
-#    include_prefix = "datalake",
-#    visibility = ["//visibility:public"],
-#    deps = [
-#        ":cloud_data_io",
-#        ":logger",
-#        ":types",
-#        "//src/v/base",
-#        "//src/v/cluster",
-#        "//src/v/config",
-#        "//src/v/container:chunked_hash_map",
-#        "//src/v/datalake/coordinator:frontend",
-#        "//src/v/datalake/translation:partition_translator",
-#        "//src/v/features",
-#        "//src/v/model",
-#        "//src/v/raft",
-#        "//src/v/serde",
-#        "//src/v/ssx:semaphore",
-#        "@fmt",
-#        "@seastar",
-#    ],
-#)
+redpanda_cc_library(
+    name = "manager",
+    srcs = [
+        "datalake_manager.cc",
+    ],
+    hdrs = [
+        "datalake_manager.h",
+    ],
+    include_prefix = "datalake",
+    visibility = ["//visibility:public"],
+    deps = [
+        ":catalog_schema_manager",
+        ":cloud_data_io",
+        ":logger",
+        ":types",
+        "//src/v/base",
+        "//src/v/cluster",
+        "//src/v/config",
+        "//src/v/container:chunked_hash_map",
+        "//src/v/datalake:record_schema_resolver",
+        "//src/v/datalake/coordinator:frontend",
+        "//src/v/datalake/translation:partition_translator",
+        "//src/v/features",
+        "//src/v/model",
+        "//src/v/pandaproxy",
+        "//src/v/raft",
+        "//src/v/schema:registry",
+        "//src/v/serde",
+        "//src/v/ssx:semaphore",
+        "@fmt",
+        "@seastar",
+    ],
+)
+
+redpanda_cc_library(
+    name = "translation_task",
+    srcs = [
+        "translation_task.cc",
+    ],
+    hdrs = [
+        "translation_task.h",
+    ],
+    include_prefix = "datalake",
+    visibility = ["//visibility:public"],
+    deps = [
+        ":cloud_data_io",
+        ":logger",
+        ":record_multiplexer",
+        ":types",
+        ":writer",
+        "//src/v/base",
+        "//src/v/datalake/coordinator:translated_offset_range",
+        "//src/v/model",
+        "//src/v/utils:lazy_abort_source",
+        "//src/v/utils:retry_chain_node",
+        "@fmt",
+        "@seastar",
+    ],
+)
 
 redpanda_cc_library(
     name = "table_definition",

src/v/datalake/coordinator/BUILD

@@ -281,6 +281,11 @@ redpanda_cc_library(
         "translated_offset_range.h",
     ],
     include_prefix = "datalake/coordinator",
+    visibility = [
+        ":__subpackages__",
+        "//src/v/datalake:__pkg__",
+        "//src/v/datalake/translation:__pkg__",
+    ],
     deps = [
         ":data_file",
         "//src/v/base",

src/v/datalake/translation/BUILD

@@ -32,39 +32,44 @@ redpanda_cc_library(
     ],
 )
 
-# TODO(iceberg): depends on batching_parquet_writer
-# which depends on arrow which is bazel compatible.
-# uncomment once the issue is resolved.
-#redpanda_cc_library(
-#    name = "partition_translator",
-#    srcs = [
-#        "partition_translator.cc",
-#    ],
-#    hdrs = [
-#        "partition_translator.h",
-#    ],
-#    include_prefix = "datalake/translation",
-#    visibility = ["//visibility:public"],
-#    deps = [
-#        ":model",
-#        ":stm",
-#        "//src/v/base",
-#        "//src/v/cluster",
-#        "//src/v/datalake:cloud_data_io",
-#        "//src/v/datalake:logger",
-#        "//src/v/datalake:types",
-#        "//src/v/datalake:writer",
-#        "//src/v/datalake/coordinator:frontend",
-#        "//src/v/features",
-#        "//src/v/kafka/data:partition_proxy",
-#        "//src/v/kafka/utils:txn_reader",
-#        "//src/v/model",
-#        "//src/v/random:time_jitter",
-#        "//src/v/resource_mgmt:io_priority",
-#        "//src/v/ssx:future_util",
-#        "//src/v/ssx:semaphore",
-#        "//src/v/utils:prefix_logger",
-#        "//src/v/utils:retry_chain_node",
-#        "@seastar",
-#   ],
-#)
+redpanda_cc_library(
+    name = "partition_translator",
+    srcs = [
+        "partition_translator.cc",
+    ],
+    hdrs = [
+        "partition_translator.h",
+    ],
+    include_prefix = "datalake/translation",
+    visibility = ["//visibility:public"],
+    deps = [
+        ":model",
+        ":stm",
+        "//src/v/base",
+        "//src/v/cluster",
+        "//src/v/datalake:cloud_data_io",
+        "//src/v/datalake:local_parquet_file_writer",
+        "//src/v/datalake:logger",
+        "//src/v/datalake:record_multiplexer",
+        "//src/v/datalake:record_schema_resolver",
+        "//src/v/datalake:record_translator",
+        "//src/v/datalake:serde_parquet_writer",
+        "//src/v/datalake:translation_task",
+        "//src/v/datalake:types",
+        "//src/v/datalake:writer",
+        "//src/v/datalake/coordinator:frontend",
+        "//src/v/datalake/coordinator:translated_offset_range",
+        "//src/v/features",
+        "//src/v/kafka/data:partition_proxy",
+        "//src/v/kafka/utils:txn_reader",
+        "//src/v/model",
+        "//src/v/random:time_jitter",
+        "//src/v/resource_mgmt:io_priority",
+        "//src/v/ssx:future_util",
+        "//src/v/ssx:semaphore",
+        "//src/v/utils:lazy_abort_source",
+        "//src/v/utils:prefix_logger",
+        "//src/v/utils:retry_chain_node",
+        "@seastar",
+    ],
+)

src/v/datalake/translation/partition_translator.cc

@@ -12,7 +12,6 @@
 
 #include "cluster/archival/types.h"
 #include "cluster/partition.h"
-#include "datalake/batching_parquet_writer.h"
 #include "datalake/coordinator/frontend.h"
 #include "datalake/coordinator/translated_offset_range.h"
 #include "datalake/data_writer_interface.h"
@@ -81,8 +80,6 @@ ss::futurize_t<FuncRet> retry_with_backoff(
 
 static constexpr std::chrono::milliseconds translation_jitter{500};
 constexpr ::model::timeout_clock::duration wait_timeout = 5s;
-constexpr size_t max_rows_per_row_group = std::numeric_limits<size_t>::max();
-constexpr size_t max_bytes_per_row_group = std::numeric_limits<size_t>::max();
 
 partition_translator::~partition_translator() = default;
 partition_translator::partition_translator(
@@ -178,28 +175,6 @@ partition_translator::max_offset_for_translation() const {
     return kafka::prev_offset(model::offset_cast(lso.value()));
 }
 
-namespace {
-/**
- * In order to make it easily configurable we add a
- * '__REDPANDA_USE_SERDE_PARQUET_WRITER' environment variable, when set a serde
- * parquet writer will be used.
- * TODO: remove this once serde parquet writer is ready to be used as
- * default
- */
-ss::shared_ptr<parquet_ostream_factory> get_parquet_writer_factory() {
-    auto use_serde_parquet_writer = std::getenv(
-      "__REDPANDA_USE_SERDE_PARQUET_WRITER");
-
-    if (use_serde_parquet_writer != nullptr) {
-        vlog(datalake_log.info, "Using serde parquet writer");
-        return ss::make_shared<serde_parquet_writer_factory>();
-    }
-    return ss::make_shared<batching_parquet_writer_factory>(
-      max_rows_per_row_group,   // max entries per single parquet row group
-      max_bytes_per_row_group); // max bytes per single parquet row group
-}
-} // namespace
-
 ss::future<std::optional<coordinator::translated_offset_range>>
 partition_translator::do_translation_for_range(
   retry_chain_node& parent,
@@ -210,7 +185,7 @@ partition_translator::do_translation_for_range(
     auto writer_factory = std::make_unique<local_parquet_file_writer_factory>(
       local_path{_writer_scratch_space}, // storage temp files are written to
       fmt::format("{}", begin_offset),   // file prefix
-      get_parquet_writer_factory());
+      ss::make_shared<serde_parquet_writer_factory>());
 
     auto task = translation_task{
       **_cloud_io, *_schema_mgr, *_type_resolver, *_record_translator};

tests/rptest/services/redpanda.py

@@ -2561,13 +2561,6 @@ def __init__(self,
         self._seed_servers = self.nodes
 
         self._expect_max_controller_records = 1000
-        #TODO: remove when sered parquet is enabled by default
-        if hasattr(self._context, 'injected_args') \
-            and self._context.injected_args is not None:
-            if 'use_serde_parquet' in  self._context.injected_args  \
-               and self._context.injected_args['use_serde_parquet']:
-                self._environment.update(
-                    {"__REDPANDA_USE_SERDE_PARQUET_WRITER": "1"})
 
     def redpanda_env_preamble(self):
         # Pass environment variables via FOO=BAR shell expressions

tests/rptest/tests/datalake/datalake_e2e_test.py

@@ -84,10 +84,8 @@ def test_e2e_basic(self, cloud_storage_type, query_engine,
 
     @cluster(num_nodes=4)
     @matrix(cloud_storage_type=supported_storage_types(),
-            query_engine=[QueryEngineType.SPARK, QueryEngineType.TRINO],
-            use_serde_parquet=[False, True])
-    def test_avro_schema(self, cloud_storage_type, query_engine,
-                         use_serde_parquet):
+            query_engine=[QueryEngineType.SPARK, QueryEngineType.TRINO])
+    def test_avro_schema(self, cloud_storage_type, query_engine):
         count = 100
         table_name = f"redpanda.{self.topic_name}"
 

tests/rptest/tests/datalake/rest_catalog_connection_test.py

@@ -84,10 +84,8 @@ def start_producer(self, topic_name: str):
         return producer
 
     @cluster(num_nodes=5)
-    @matrix(cloud_storage_type=supported_storage_types(),
-            use_serde_parquet=[False, True])
-    def test_redpanda_connection_to_rest_catalog(self, cloud_storage_type,
-                                                 use_serde_parquet):
+    @matrix(cloud_storage_type=supported_storage_types())
+    def test_redpanda_connection_to_rest_catalog(self, cloud_storage_type):
 
         catalog = self.catalog_service.client()
         namespace = "redpanda"