Merge pull request #26438 from wdberkeley/v25.1.x-iceberg-dbx-backport

piyushredpanda · web-flow · commit f0e74e1eb224 · 2025-06-11T22:29:16.000-07:00
Backport #26426 to v25.1.x
diff --git a/src/v/datalake/catalog_schema_manager.cc b/src/v/datalake/catalog_schema_manager.cc
@@ -20,6 +20,21 @@
 namespace datalake {
 
 namespace {
+
+// NB: Use the concrete type not the alias iceberg::table_properties_t
+// so changes to the alias can't silently destroy the deep copy semantics
+// of this function.
+chunked_hash_map<ss::sstring, ss::sstring, sstring_hash, sstring_eq>
+copy_properties(
+  const chunked_hash_map<ss::sstring, ss::sstring, sstring_hash, sstring_eq>&
+    props) {
+    chunked_hash_map<ss::sstring, ss::sstring, sstring_hash, sstring_eq> result;
+    for (const auto& [key, value] : props) {
+        result.emplace(key, value);
+    }
+    return result;
+}
+
 schema_manager::errc log_and_convert_catalog_err(
   iceberg::catalog::errc e, std::string_view log_msg) {
     switch (e) {
@@ -146,6 +161,7 @@ simple_schema_manager::ensure_table_schema(
           table_location_prefix_(),
           fmt::join(table_id.ns, "/"),
           table_id.table)),
+        .properties = std::nullopt,
       });
 
     co_return std::nullopt;
@@ -164,6 +180,7 @@ simple_schema_manager::get_table_info(
       .schema = it->second.schema.copy(),
       .partition_spec = it->second.partition_spec.copy(),
       .location = it->second.location,
+      .properties = it->second.properties.transform(copy_properties),
     };
 }
 
@@ -281,6 +298,7 @@ catalog_schema_manager::get_table_info(
       .schema = cur_schema->copy(),
       .partition_spec = cur_spec->copy(),
       .location = table.location,
+      .properties = table.properties.transform(copy_properties),
     };
 }
 
diff --git a/src/v/datalake/catalog_schema_manager.h b/src/v/datalake/catalog_schema_manager.h
@@ -40,6 +40,7 @@ class schema_manager {
         iceberg::schema schema;
         iceberg::partition_spec partition_spec;
         iceberg::uri location;
+        std::optional<iceberg::table_properties_t> properties;
 
         // Fills the field IDs of the given type with those in the current
         // schema. Returns true on success.
diff --git a/src/v/datalake/datalake_manager.cc b/src/v/datalake/datalake_manager.cc
@@ -33,7 +33,6 @@
 
 constexpr std::chrono::milliseconds translation_jitter{500};
 constexpr std::chrono::milliseconds translation_jitter_base{5000};
-static constexpr std::string_view iceberg_data_path_prefix = "data";
 
 namespace datalake {
 
@@ -702,7 +701,6 @@ datalake_manager::handle_translator_state_change(const model::ntp& ntp) {
         std::move(record_translator),
         std::move(table_creator),
         _location_provider,
-        remote_path{iceberg_data_path_prefix},
         *reservations,
         _topic_table,
         _features,
diff --git a/src/v/datalake/partitioning_writer.cc b/src/v/datalake/partitioning_writer.cc
@@ -136,7 +136,7 @@ partitioning_writer::finish() && {
 
         files.push_back(partitioned_file{
           .local_file = std::move(file_res.value()),
-          .table_location = remote_prefix_,
+          .data_location = remote_prefix_,
           .schema_id = schema_id_,
           .partition_spec_id = spec_.spec_id,
           .partition_key = std::move(pk),
diff --git a/src/v/datalake/partitioning_writer.h b/src/v/datalake/partitioning_writer.h
@@ -58,7 +58,7 @@ class partitioning_writer {
 
     struct partitioned_file {
         local_file_metadata local_file;
-        remote_path table_location;
+        remote_path data_location;
         iceberg::schema::id_t schema_id;
         iceberg::partition_spec::id_t partition_spec_id;
         iceberg::partition_key partition_key;
diff --git a/src/v/datalake/record_multiplexer.cc b/src/v/datalake/record_multiplexer.cc
@@ -28,6 +28,22 @@
 namespace datalake {
 
 namespace {
+
+// Get the data location for the table. Some catalogs require using the property
+// `write.data.path`. Otherwise, it defaults to <table location>/data.
+iceberg::uri get_data_location(const schema_manager::table_info& table_info) {
+    static constexpr std::string_view write_data_path_prop = "write.data.path";
+
+    if (table_info.properties.has_value()) {
+        auto it = table_info.properties->find(write_data_path_prop);
+        if (it != table_info.properties->end()) {
+            return iceberg::uri(it->second);
+        }
+    }
+
+    return iceberg::uri(fmt::format("{}/data", table_info.location));
+}
+
 template<typename Func>
 requires requires(Func f, model::record_batch batch) {
     { f(std::move(batch)) } -> std::same_as<ss::future<ss::stop_iteration>>;
@@ -258,9 +274,9 @@ ss::future<ss::stop_iteration> record_multiplexer::do_multiplex(
                 co_return ss::stop_iteration::yes;
             }
 
-            auto table_remote_path = _location_provider.from_uri(
-              load_res.value().location);
-            if (!table_remote_path) {
+            auto data_location = get_data_location(load_res.value());
+            auto data_remote_path = _location_provider.from_uri(data_location);
+            if (!data_remote_path) {
                 vlog(
                   _log.warn,
                   "Error getting location prefix for {} while creating writer "
@@ -278,7 +294,7 @@ ss::future<ss::stop_iteration> record_multiplexer::do_multiplex(
                 load_res.value().schema.schema_id,
                 std::move(record_type.type),
                 std::move(load_res.value().partition_spec),
-                std::move(table_remote_path.value())));
+                std::move(data_remote_path.value())));
             writer_iter = iter;
         }
 
@@ -476,9 +492,9 @@ record_multiplexer::handle_invalid_record(
                 co_return writer_error::unknown_error;
             }
 
-            auto table_remote_path = _location_provider.from_uri(
-              load_res.value().location);
-            if (!table_remote_path) {
+            auto data_location = get_data_location(load_res.value());
+            auto data_remote_path = _location_provider.from_uri(data_location);
+            if (!data_remote_path) {
                 vlog(
                   _log.warn,
                   "Error getting location prefix for {} while creating writer "
@@ -493,7 +509,7 @@ record_multiplexer::handle_invalid_record(
               load_res.value().schema.schema_id,
               std::move(record_type.type),
               std::move(load_res.value().partition_spec),
-              std::move(table_remote_path.value()));
+              std::move(data_remote_path.value()));
         }
 
         int64_t estimated_size = (key ? key->size_bytes() : 0)
diff --git a/src/v/datalake/tests/translation_task_test.cc b/src/v/datalake/tests/translation_task_test.cc
@@ -199,7 +199,6 @@ TEST_F(TranslateTaskTest, TestHappyPathTranslation) {
     auto result = std::move(task)
                     .finish(
                       translation_task::custom_partitioning_enabled::yes,
-                      datalake::remote_path("test/location/1"),
                       test_rcn,
                       as)
                     .get();
@@ -230,7 +229,6 @@ TEST_F(TranslateTaskTest, TestDataFileMissing) {
     auto result = std::move(task)
                     .finish(
                       translation_task::custom_partitioning_enabled::yes,
-                      datalake::remote_path("test/location/1"),
                       test_rcn,
                       as)
                     .get();
@@ -265,7 +263,6 @@ TEST_F(TranslateTaskTest, TestUploadError) {
     auto result = std::move(task)
                     .finish(
                       translation_task::custom_partitioning_enabled::yes,
-                      datalake::remote_path("test/location/1"),
                       test_rcn,
                       as)
                     .get();
diff --git a/src/v/datalake/translation/deps.cc b/src/v/datalake/translation/deps.cc
@@ -493,7 +493,6 @@ class partition_translation_context : public translation_context {
       std::unique_ptr<record_translator> record_translator,
       std::unique_ptr<table_creator> table_creator,
       location_provider location_provider,
-      remote_path upload_path_prefix,
       scheduling::reservations_tracker& reservations,
       ss::sharded<cluster::topic_table>* topics,
       ss::sharded<features::feature_table>* features,
@@ -507,7 +506,6 @@ class partition_translation_context : public translation_context {
       , _record_translator(std::move(record_translator))
       , _table_creator(std::move(table_creator))
       , _location_provider(std::move(location_provider))
-      , _upload_path_prefix(std::move(upload_path_prefix))
       , _reservations(reservations)
       , _topics(topics->local())
       , _features(features->local())
@@ -605,8 +603,8 @@ class partition_translation_context : public translation_context {
         }
         vlog(datalake_log.debug, "[{}] finishing translation", _ntp);
         auto task = std::exchange(_in_progress_translation, std::nullopt);
-        auto result = co_await std::move(task.value())
-                        .finish(_cp_enabled, _upload_path_prefix, rcn, as);
+        auto result
+          = co_await std::move(task.value()).finish(_cp_enabled, rcn, as);
 
         if (result.has_error()) {
             co_return map_error_code(result.error());
@@ -668,7 +666,6 @@ class partition_translation_context : public translation_context {
     std::unique_ptr<record_translator> _record_translator;
     std::unique_ptr<table_creator> _table_creator;
     location_provider _location_provider;
-    remote_path _upload_path_prefix;
     scheduling::reservations_tracker& _reservations;
     cluster::topic_table& _topics;
     features::feature_table& _features;
@@ -690,7 +687,6 @@ translation_context::make_default_translation_context(
   std::unique_ptr<record_translator> record_translator,
   std::unique_ptr<table_creator> table_creator,
   location_provider location_provider,
-  remote_path upload_path_prefix,
   scheduling::reservations_tracker& reservations,
   ss::sharded<cluster::topic_table>* topics,
   ss::sharded<features::feature_table>* features,
@@ -705,7 +701,6 @@ translation_context::make_default_translation_context(
       std::move(record_translator),
       std::move(table_creator),
       std::move(location_provider),
-      std::move(upload_path_prefix),
       reservations,
       topics,
       features,
diff --git a/src/v/datalake/translation/deps.h b/src/v/datalake/translation/deps.h
@@ -294,7 +294,6 @@ class translation_context {
       std::unique_ptr<record_translator>,
       std::unique_ptr<table_creator>,
       location_provider,
-      remote_path,
       scheduling::reservations_tracker&,
       ss::sharded<cluster::topic_table>*,
       ss::sharded<features::feature_table>*,
diff --git a/src/v/datalake/translation_task.cc b/src/v/datalake/translation_task.cc
@@ -65,11 +65,10 @@ ss::future<checked<remote_path, translation_task::errc>> execute_single_upload(
   prefix_logger& logger,
   cloud_data_io& _cloud_io,
   const partitioning_writer::partitioned_file& file,
-  const remote_path& remote_path_prefix,
   retry_chain_node& parent_rcn,
   lazy_abort_source& lazy_as) {
     auto file_remote_path = remote_path{
-      file.table_location / remote_path_prefix / file.partition_key_path
+      file.data_location / file.partition_key_path
       / file.local_file.path().filename()};
 
     // (Approximate because I'm ignoring backoff) ~5 retries at 1
@@ -126,7 +125,6 @@ upload_files(
   cloud_data_io& _cloud_io,
   const chunked_vector<partitioning_writer::partitioned_file>& files,
   translation_task::custom_partitioning_enabled is_custom_partitioning_enabled,
-  const remote_path& remote_path_prefix,
   retry_chain_node& rcn,
   lazy_abort_source& lazy_as) {
     chunked_vector<coordinator::data_file> ret;
@@ -135,7 +133,7 @@ upload_files(
     std::optional<translation_task::errc> upload_error;
     for (auto& file : files) {
         auto r = co_await execute_single_upload(
-          logger, _cloud_io, file, remote_path_prefix, rcn, lazy_as);
+          logger, _cloud_io, file, rcn, lazy_as);
 
         if (r.has_error()) {
             vlog(
@@ -280,7 +278,6 @@ ss::future<
   checked<coordinator::translated_offset_range, translation_task::errc>>
 translation_task::finish(
   custom_partitioning_enabled is_custom_partitioning_enabled,
-  const remote_path& remote_path_prefix,
   retry_chain_node& rcn,
   ss::abort_source& as) && {
     auto mux_result = co_await std::move(_multiplexer).finish();
@@ -331,7 +328,6 @@ translation_task::finish(
           *_cloud_io,
           write_result.data_files,
           is_custom_partitioning_enabled,
-          remote_path_prefix,
           rcn,
           lazy_as);
         if (upload_res.has_error()) {
@@ -347,7 +343,6 @@ translation_task::finish(
           *_cloud_io,
           write_result.dlq_files,
           is_custom_partitioning_enabled,
-          remote_path_prefix,
           rcn,
           lazy_as);
         if (dlq_upload_res.has_error()) {
diff --git a/src/v/datalake/translation_task.h b/src/v/datalake/translation_task.h
@@ -88,7 +88,6 @@ class translation_task {
      */
     ss::future<checked<coordinator::translated_offset_range, errc>> finish(
       custom_partitioning_enabled is_custom_partitioning_enabled,
-      const remote_path& remote_path_prefix,
       retry_chain_node& parent_rcn,
       ss::abort_source&) &&;
 
diff --git a/src/v/iceberg/BUILD b/src/v/iceberg/BUILD
@@ -821,6 +821,7 @@ redpanda_cc_library(
         ":uri",
         "//src/v/container:fragmented_vector",
         "//src/v/model",
+        "//src/v/utils:absl_sstring_hash",
         "//src/v/utils:named_type",
         "//src/v/utils:uuid",
     ],
diff --git a/src/v/iceberg/merge_append_action.cc b/src/v/iceberg/merge_append_action.cc
@@ -27,18 +27,38 @@
 namespace iceberg {
 
 namespace {
+
+// Derive the metadata location from table properties.
+// Some catalogs require respecting the property `write.metadata.path`,
+// but the default location is <table location>/metadata.
+uri get_metadata_location(const table_metadata& table) {
+    static constexpr std::string_view write_metadata_path_prop
+      = "write.metadata.path";
+
+    if (table.properties.has_value()) {
+        auto it = table.properties->find(write_metadata_path_prop);
+        if (it != table.properties->end()) {
+            return uri(it->second);
+        }
+    }
+
+    return uri(fmt::format("{}/metadata", table.location));
+}
+
 uri get_manifest_path(
-  const uri& location, const uuid_t& commit_uuid, size_t num) {
+  const table_metadata& table, const uuid_t& commit_uuid, size_t num) {
+    auto metadata_location = get_metadata_location(table);
     return uri(
-      fmt::format("{}/metadata/{}-m{}.avro", location, commit_uuid, num));
+      fmt::format("{}/{}-m{}.avro", metadata_location, commit_uuid, num));
 }
 uri get_manifest_list_path(
-  const uri& location,
+  const table_metadata& table,
   snapshot_id snap_id,
   const uuid_t& commit_uuid,
   size_t num) {
+    auto metadata_location = get_metadata_location(table);
     return uri{fmt::format(
-      "{}/metadata/snap-{}-{}-{}.avro", location, snap_id(), commit_uuid, num)};
+      "{}/snap-{}-{}-{}.avro", metadata_location, snap_id(), commit_uuid, num)};
 }
 
 action::errc to_action_errc(metadata_io::errc e) {
@@ -248,7 +268,7 @@ ss::future<action::action_outcome> merge_append_action::build_updates() && {
     // naming uniqueness. Retries for us are expected to take the form of an
     // entirely new transaction.
     const auto new_mlist_path = get_manifest_list_path(
-      table_.location, new_snap_id, commit_uuid_, 0);
+      table_, new_snap_id, commit_uuid_, 0);
 
     vlog(
       log.info,
@@ -516,7 +536,7 @@ merge_append_action::merge_mfiles(
     }
 
     const auto merged_manifest_path = get_manifest_path(
-      table_.location, ctx.commit_uuid, generate_manifest_num());
+      table_, ctx.commit_uuid, generate_manifest_num());
     const auto mfile_up_res = co_await upload_as_manifest(
       merged_manifest_path, *schema, pspec, std::move(merged_entries));
     if (mfile_up_res.has_error()) {
diff --git a/src/v/iceberg/table_metadata.h b/src/v/iceberg/table_metadata.h
@@ -17,13 +17,17 @@
 #include "iceberg/snapshot.h"
 #include "iceberg/transform.h"
 #include "iceberg/uri.h"
+#include "utils/absl_sstring_hash.h"
 #include "utils/named_type.h"
 #include "utils/uuid.h"
 
 #include <optional>
 
 namespace iceberg {
 
+using table_properties_t
+  = chunked_hash_map<ss::sstring, ss::sstring, sstring_hash, sstring_eq>;
+
 enum class sort_direction {
     asc,
     desc,
@@ -67,7 +71,7 @@ struct table_metadata {
     partition_spec::id_t default_spec_id;
     partition_field::id_t last_partition_id;
 
-    std::optional<chunked_hash_map<ss::sstring, ss::sstring>> properties;
+    std::optional<table_properties_t> properties;
     std::optional<snapshot_id> current_snapshot_id;
     std::optional<chunked_vector<snapshot>> snapshots;
 
diff --git a/src/v/iceberg/table_metadata_json.cc b/src/v/iceberg/table_metadata_json.cc
diff --git a/src/v/iceberg/tests/merge_append_action_test.cc b/src/v/iceberg/tests/merge_append_action_test.cc
diff --git a/src/v/iceberg/tests/rest_catalog_test.cc b/src/v/iceberg/tests/rest_catalog_test.cc
