[fix](iceberg) Write binary columns with proper Arrow types

be/src/core/data_type_serde/data_type_string_serde.cpp

@@ -17,6 +17,9 @@
 
 #include "core/data_type_serde/data_type_string_serde.h"
 
+#include <array>
+#include <cstring>
+
 #include "core/column/column_string.h"
 #include "core/data_type/define_primitive_type.h"
 #include "util/jsonb_document_cast.h"
@@ -25,6 +28,88 @@
 
 namespace doris {
 
+namespace {
+
+int hex_value(char c) {
+ if (c >= '0' && c <= '9') {
+ return c - '0';
+ }
+ if (c >= 'a' && c <= 'f') {
+ return c - 'a' + 10;
+ }
+ if (c >= 'A' && c <= 'F') {
+ return c - 'A' + 10;
+ }
+ return -1;
+}
+
+Status parse_uuid_to_bytes(StringRef uuid, std::array<uint8_t, 16>* bytes) {
+ if (uuid.size != 32 && uuid.size != 36) {
+ return Status::InvalidArgument("Invalid UUID string length: {}", uuid.size);
+ }
+
+ int hex_count = 0;
+ int high_nibble = -1;
+ int byte_index = 0;
+ for (size_t i = 0; i < uuid.size; ++i) {
+ char c = uuid.data[i];
+ if (uuid.size == 36 && (i == 8 || i == 13 || i == 18 || i == 23)) {
+ if (c != '-') {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+ continue;
+ }
+ if (c == '-') {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+
+ int value = hex_value(c);
+ if (value < 0) {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+ if (hex_count % 2 == 0) {
+ high_nibble = value;
+ } else {
+ (*bytes)[byte_index++] = static_cast<uint8_t>((high_nibble << 4) | value);
+ }
+ ++hex_count;
+ }
+
+ if (hex_count != 32 || byte_index != 16) {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+ return Status::OK();
+}
+
+Status append_fixed_size_binary(arrow::FixedSizeBinaryBuilder& builder, const IColumn& column,
+ StringRef string_ref, int byte_width, bool pad_char_value,
+ bool convert_uuid_string) {
+ if (convert_uuid_string && byte_width == 16 &&
+ (string_ref.size == 32 || string_ref.size == 36)) {
+ std::array<uint8_t, 16> bytes;
+ RETURN_IF_ERROR(parse_uuid_to_bytes(string_ref, &bytes));
+ return checkArrowStatus(builder.Append(bytes.data()), column, builder);
+ }
+
+ if (string_ref.size == byte_width) {
+ return checkArrowStatus(builder.Append(reinterpret_cast<const uint8_t*>(string_ref.data)),
+ column, builder);
+ }
+
+ if (pad_char_value && string_ref.size < byte_width) {
+ std::string padded_value(byte_width, '\0');
+ std::memcpy(padded_value.data(), string_ref.data, string_ref.size);
+ return checkArrowStatus(
+ builder.Append(reinterpret_cast<const uint8_t*>(padded_value.data())), column,
+ builder);
+ }
+
+ return Status::InvalidArgument("Fixed size binary column expects {} bytes, got {}", byte_width,
+ string_ref.size);
+}
+
+} // namespace
+
 template <typename ColumnType>
 Status DataTypeStringSerDeBase<ColumnType>::serialize_column_to_json(const IColumn& column,
  int64_t start_idx,
@@ -243,9 +328,42 @@ Status DataTypeStringSerDeBase<ColumnType>::write_column_to_arrow(
  if (array_builder->type()->id() == arrow::Type::LARGE_STRING) {
  auto& builder = assert_cast<arrow::LargeStringBuilder&>(*array_builder);
  return write_column_to_arrow_impl(column, null_map, builder, start, end);
+ } else if (array_builder->type()->id() == arrow::Type::LARGE_BINARY) {
+ auto& builder = assert_cast<arrow::LargeBinaryBuilder&>(*array_builder);
+ const auto& string_column = assert_cast<const ColumnType&>(column);
+ for (size_t string_i = start; string_i < end; ++string_i) {
+ if (null_map && (*null_map)[string_i]) {
+ RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder));
+ continue;
+ }
+ auto string_ref = string_column.get_data_at(string_i);
+ RETURN_IF_ERROR(checkArrowStatus(
+ builder.Append(reinterpret_cast<const uint8_t*>(string_ref.data),
+ cast_set<int64_t, size_t, false>(string_ref.size)),
+ column, builder));
+ }
+ return Status::OK();
  } else if (array_builder->type()->id() == arrow::Type::STRING) {
  auto& builder = assert_cast<arrow::StringBuilder&>(*array_builder);
  return write_column_to_arrow_impl(column, null_map, builder, start, end);
+ } else if (array_builder->type()->id() == arrow::Type::BINARY) {
+ auto& builder = assert_cast<arrow::BinaryBuilder&>(*array_builder);
+ return write_column_to_arrow_impl(column, null_map, builder, start, end);
+ } else if (array_builder->type()->id() == arrow::Type::FIXED_SIZE_BINARY) {
+ auto& builder = assert_cast<arrow::FixedSizeBinaryBuilder&>(*array_builder);
+ const int byte_width =
+ static_cast<const arrow::FixedSizeBinaryType&>(*array_builder->type()).byte_width();
+ const auto& string_column = assert_cast<const ColumnType&>(column);
+ for (size_t string_i = start; string_i < end; ++string_i) {
+ if (null_map && (*null_map)[string_i]) {
+ RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder));
+ continue;
+ }
+ auto string_ref = string_column.get_data_at(string_i);
+ RETURN_IF_ERROR(append_fixed_size_binary(builder, column, string_ref, byte_width,
+ _type == TYPE_CHAR, _type != TYPE_CHAR));
+ }
+ return Status::OK();
  } else {
  return Status::InvalidArgument("Unsupported arrow type for string column: {}",
  array_builder->type()->name());

be/src/core/data_type_serde/data_type_varbinary_serde.cpp

@@ -70,9 +70,44 @@ Status DataTypeVarbinarySerDe::write_column_to_arrow(const IColumn& column, cons
  if (array_builder->type()->id() == arrow::Type::BINARY) {
  auto& builder = assert_cast<arrow::BinaryBuilder&>(*array_builder);
  return lambda_function(builder);
+ } else if (array_builder->type()->id() == arrow::Type::LARGE_BINARY) {
+ auto& builder = assert_cast<arrow::LargeBinaryBuilder&>(*array_builder);
+ const auto& varbinary_column_data = assert_cast<const ColumnVarbinary&>(column).get_data();
+ for (size_t i = start; i < end; ++i) {
+ if (null_map && (*null_map)[i]) {
+ RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder));
+ continue;
+ }
+ const auto& string_view = varbinary_column_data[i];
+ RETURN_IF_ERROR(checkArrowStatus(
+ builder.Append(reinterpret_cast<const uint8_t*>(string_view.data()),
+ cast_set<int64_t, size_t, false>(string_view.size())),
+ column, builder));
+ }
+ return Status::OK();
  } else if (array_builder->type()->id() == arrow::Type::STRING) {
  auto& builder = assert_cast<arrow::StringBuilder&>(*array_builder);
  return lambda_function(builder);
+ } else if (array_builder->type()->id() == arrow::Type::FIXED_SIZE_BINARY) {
+ auto& builder = assert_cast<arrow::FixedSizeBinaryBuilder&>(*array_builder);
+ const int byte_width =
+ static_cast<const arrow::FixedSizeBinaryType&>(*array_builder->type()).byte_width();
+ const auto& varbinary_column_data = assert_cast<const ColumnVarbinary&>(column).get_data();
+ for (size_t i = start; i < end; ++i) {
+ if (null_map && (*null_map)[i]) {
+ RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder));
+ continue;
+ }
+ const auto& string_view = varbinary_column_data[i];
+ if (string_view.size() != byte_width) {
+ return Status::InvalidArgument("Fixed size binary column expects {} bytes, got {}",
+ byte_width, string_view.size());
+ }
+ RETURN_IF_ERROR(checkArrowStatus(
+ builder.Append(reinterpret_cast<const uint8_t*>(string_view.data())), column,
+ builder));
+ }
+ return Status::OK();
  } else {
  return Status::InvalidArgument("Unsupported arrow type for varbinary column: {}",
  array_builder->type()->name());

be/src/format/arrow/arrow_block_convertor.cpp

@@ -26,11 +26,14 @@
 #include <arrow/status.h>
 #include <arrow/type.h>
 #include <arrow/util/decimal.h>
+#include <arrow/util/key_value_metadata.h>
 #include <arrow/visit_type_inline.h>
 #include <arrow/visitor.h>
 #include <cctz/time_zone.h>
 #include <glog/logging.h>
 
+#include <array>
+#include <cstring>
 #include <ctime>
 #include <memory>
 #include <utility>
@@ -39,6 +42,8 @@
 #include "common/status.h"
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
 #include "core/data_type/data_type.h"
 #include "core/data_type/data_type_array.h"
 #include "core/data_type/data_type_nullable.h"
@@ -52,6 +57,115 @@ class Array;
 
 namespace doris {
 
+namespace {
+
+constexpr const char* ICEBERG_ORIGINAL_TYPE_KEY = "originalType";
+constexpr const char* ICEBERG_UUID_TYPE_VALUE = "uuid";
+
+bool is_iceberg_uuid_field(const std::shared_ptr<arrow::Field>& field) {
+ if (field == nullptr || !field->HasMetadata()) {
+ return false;
+ }
+ const auto result = field->metadata()->Get(ICEBERG_ORIGINAL_TYPE_KEY);
+ return result.ok() && result.ValueUnsafe() == ICEBERG_UUID_TYPE_VALUE;
+}
+
+int hex_value(char c) {
+ if (c >= '0' && c <= '9') {
+ return c - '0';
+ }
+ if (c >= 'a' && c <= 'f') {
+ return c - 'a' + 10;
+ }
+ if (c >= 'A' && c <= 'F') {
+ return c - 'A' + 10;
+ }
+ return -1;
+}
+
+Status parse_uuid_to_bytes(StringRef uuid, std::array<uint8_t, 16>* bytes) {
+ if (uuid.size == 16) {
+ std::memcpy(bytes->data(), uuid.data, bytes->size());
+ return Status::OK();
+ }
+ if (uuid.size != 32 && uuid.size != 36) {
+ return Status::InvalidArgument("Invalid UUID string length: {}", uuid.size);
+ }
+
+ int hex_count = 0;
+ int high_nibble = -1;
+ int byte_index = 0;
+ for (size_t i = 0; i < uuid.size; ++i) {
+ char c = uuid.data[i];
+ if (uuid.size == 36 && (i == 8 || i == 13 || i == 18 || i == 23)) {
+ if (c != '-') {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+ continue;
+ }
+ if (c == '-') {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+
+ int value = hex_value(c);
+ if (value < 0) {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+ if (hex_count % 2 == 0) {
+ high_nibble = value;
+ } else {
+ (*bytes)[byte_index++] = static_cast<uint8_t>((high_nibble << 4) | value);
+ }
+ ++hex_count;
+ }
+
+ if (hex_count != 32 || byte_index != 16) {
+ return Status::InvalidArgument("Invalid UUID string format");
+ }
+ return Status::OK();
+}
+
+Status write_iceberg_uuid_string_column_to_arrow(const IColumn& column, const DataTypePtr& type,
+ arrow::ArrayBuilder* array_builder, int64_t start,
+ int64_t end) {
+ if (array_builder->type()->id() != arrow::Type::FIXED_SIZE_BINARY) {
+ return Status::InvalidArgument("Iceberg UUID must be written to fixed size binary");
+ }
+ const int byte_width =
+ static_cast<const arrow::FixedSizeBinaryType&>(*array_builder->type()).byte_width();
+ if (byte_width != 16) {
+ return Status::InvalidArgument("Iceberg UUID expects 16 bytes, got {}", byte_width);
+ }
+
+ auto& builder = assert_cast<arrow::FixedSizeBinaryBuilder&>(*array_builder);
+ const IColumn* data_column = &column;
+ const NullMap* null_map = nullptr;
+ if (type->is_nullable()) {
+ const auto& nullable_column = assert_cast<const ColumnNullable&>(column);
+ data_column = &nullable_column.get_nested_column();
+ null_map = &nullable_column.get_null_map_data();
+ }
+ if (!data_column->is_column_string()) {
+ return Status::InvalidArgument(
+ "Iceberg UUID string conversion expects string column, got {}",
+ data_column->get_name());
+ }
+
+ const auto& string_column = assert_cast<const ColumnString&>(*data_column);
+ for (size_t row = start; row < end; ++row) {
+ if (null_map != nullptr && (*null_map)[row]) {
+ RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder));
+ continue;
+ }
+ std::array<uint8_t, 16> bytes;
+ RETURN_IF_ERROR(parse_uuid_to_bytes(string_column.get_data_at(row), &bytes));
+ RETURN_IF_ERROR(checkArrowStatus(builder.Append(bytes.data()), column, builder));
+ }
+ return Status::OK();
+}
+
+} // namespace
+
 Status FromBlockToRecordBatchConverter::convert(std::shared_ptr<arrow::RecordBatch>* out) {
  int num_fields = _schema->num_fields();
  if (_block.columns() != num_fields) {
@@ -80,8 +194,11 @@ Status FromBlockToRecordBatchConverter::convert(std::shared_ptr<arrow::RecordBat
  _cur_type = _block.get_by_position(idx).type;
  auto column = _cur_col->convert_to_full_column_if_const();
  auto arrow_type = _schema->field(idx)->type();
- if (arrow_type->name() == "utf8" && column->byte_size() >= MAX_ARROW_UTF8) {
+ if (arrow_type->id() == arrow::Type::STRING && column->byte_size() >= MAX_ARROW_UTF8) {
  arrow_type = arrow::large_utf8();
+ } else if (arrow_type->id() == arrow::Type::BINARY &&
+ column->byte_size() >= MAX_ARROW_UTF8) {
+ arrow_type = arrow::large_binary();
  }
  std::unique_ptr<arrow::ArrayBuilder> builder;
  auto arrow_st = arrow::MakeBuilder(_pool, arrow_type, &builder);
@@ -90,9 +207,15 @@ Status FromBlockToRecordBatchConverter::convert(std::shared_ptr<arrow::RecordBat
  }
  _cur_builder = builder.get();
  try {
- RETURN_IF_ERROR(_cur_type->get_serde()->write_column_to_arrow(
- *column, nullptr, _cur_builder, _cur_start, _cur_start + _cur_rows,
- _timezone_obj));
+ if (is_iceberg_uuid_field(_schema->field(idx)) &&
+ is_string_type(remove_nullable(_cur_type)->get_primitive_type())) {
+ RETURN_IF_ERROR(write_iceberg_uuid_string_column_to_arrow(
+ *column, _cur_type, _cur_builder, _cur_start, _cur_start + _cur_rows));
+ } else {
+ RETURN_IF_ERROR(_cur_type->get_serde()->write_column_to_arrow(
+ *column, nullptr, _cur_builder, _cur_start, _cur_start + _cur_rows,
+ _timezone_obj));
+ }
  } catch (std::exception& e) {
  return Status::InternalError(
  "Fail to convert block data to arrow data, type: {}, name: {}, error: {}",

be/src/format/table/iceberg/arrow_schema_util.cpp

@@ -25,6 +25,7 @@ namespace doris::iceberg {
 const char* ArrowSchemaUtil::PARQUET_FIELD_ID = "PARQUET:field_id";
 const char* ArrowSchemaUtil::ORIGINAL_TYPE = "originalType";
 const char* ArrowSchemaUtil::MAP_TYPE_VALUE = "mapType";
+const char* ArrowSchemaUtil::UUID_TYPE_VALUE = "uuid";
 
 Status ArrowSchemaUtil::convert(const Schema* schema, const std::string& timezone,
  std::vector<std::shared_ptr<arrow::Field>>& fields) {
@@ -75,13 +76,25 @@ Status ArrowSchemaUtil::convert_to(const iceberg::NestedField& field,
  break;
  }
 
- case iceberg::TypeID::BINARY:
  case iceberg::TypeID::STRING:
- case iceberg::TypeID::UUID:
- case iceberg::TypeID::FIXED:
  arrow_type = arrow::utf8();
  break;
 
+ case iceberg::TypeID::BINARY:
+ arrow_type = arrow::binary();
+ break;
+
+ case iceberg::TypeID::UUID:
+ metadata[ORIGINAL_TYPE] = UUID_TYPE_VALUE;
+ arrow_type = arrow::fixed_size_binary(16);
+ break;
+
+ case iceberg::TypeID::FIXED: {
+ iceberg::FixedType* fixed_type = static_cast<iceberg::FixedType*>(field.field_type());
+ arrow_type = arrow::fixed_size_binary(fixed_type->get_length());
+ break;
+ }
+
  case iceberg::TypeID::DECIMAL: {
  auto* dt = dynamic_cast<DecimalType*>(field.field_type());
  arrow_type = arrow::decimal(dt->get_precision(), dt->get_scale());