Skip to content

[refactor](arrow) Make format convertors explicit and schema-owned #68381

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 2 commits into from
Sep 23, 2026
Merged

[refactor](arrow) Make format convertors explicit and schema-owned #68381

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c070334
[refactor](arrow) Make format convertors explicit and schema-owned
Gabriel39 Sep 22, 2026
c552895
[fix](regression) Declare the Python timezone UDTF string output
Gabriel39 Sep 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 71 additions & 0 deletions be/src/core/data_type_serde/data_type_array_serde.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,77 @@ Status DataTypeArraySerDe::write_column_to_arrow(const IColumn& column, const Nu
return Status::OK();
}

namespace {

template <typename WriteNested>
Status write_array_column_to_target(const IColumn& column, const NullMap* null_map,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
WriteNested&& write_nested) {
const auto& array_column = assert_cast<const ColumnArray&>(column);
const auto& offsets = array_column.get_offsets();
const auto& nested_data = array_column.get_data();
auto& builder = assert_cast<arrow::ListBuilder&>(*array_builder);
auto* nested_builder = builder.value_builder();
for (size_t array_idx = start; array_idx < end; ++array_idx) {
if (null_map != nullptr && (*null_map)[array_idx]) {
RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, *array_builder));
continue;
}
RETURN_IF_ERROR(checkArrowStatus(builder.Append(), column, *array_builder));
RETURN_IF_ERROR(write_nested(nested_data, nested_builder, offsets[array_idx - 1],
offsets[array_idx]));
}
return Status::OK();
}

} // namespace

Status DataTypeArraySerDe::write_column_to_paimon_arrow(
const std::shared_ptr<const IDataType>& type, const IColumn& column,
const NullMap* null_map, const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
const cctz::time_zone& ctz) const {
// Reject an incompatible target before casting its nested schema or builder.
if (field->type()->id() != arrow::Type::LIST ||

@github-actions github-actions Bot Sep 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Validate the nested binding before dispatching to the child SerDe

This only checks that the outer target is a LIST. The Paimon/Iceberg converters bypass the recursive plain-binding validator, so a same-outer-kind mismatch reaches the child writer. For example, a Doris array<decimal(10,3)> value 1.234 has raw coefficient 1234; if the pinned target is list<decimal(10,2)>, the Decimal128 writer accepts the same builder and persists that coefficient as 12.34. array<int> versus list<string> instead reaches the wrong builder cast, and Map/Struct have the same gap. Please validate the full recursive binding (including decimal parameters, timestamp units, and struct field identity/order) before appending the parent builder.

array_builder->type()->id() != arrow::Type::LIST) {
return Status::InvalidArgument("Paimon array writer requires an Arrow list field");
}
const auto& array_type = assert_cast<const DataTypeArray&>(*type);
const auto& list_type = assert_cast<const arrow::ListType&>(*field->type());
const auto& nested_field = list_type.value_field();
return write_array_column_to_target(
column, null_map, array_builder, start, end,
[&](const IColumn& nested_data, arrow::ArrayBuilder* nested_builder,
int64_t nested_start, int64_t nested_end) {
return nested_serde->write_column_to_paimon_arrow(
array_type.get_nested_type(), nested_data, nullptr, nested_field,
nested_builder, nested_start, nested_end, ctz);
});
}

Status DataTypeArraySerDe::write_column_to_iceberg_arrow(
const std::shared_ptr<const IDataType>& type, const IColumn& column,
const NullMap* null_map, const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
const cctz::time_zone& ctz) const {
// Reject an incompatible target before casting its nested schema or builder.
if (field->type()->id() != arrow::Type::LIST ||
array_builder->type()->id() != arrow::Type::LIST) {
return Status::InvalidArgument("Iceberg array writer requires an Arrow list field");
}
const auto& array_type = assert_cast<const DataTypeArray&>(*type);
const auto& list_type = assert_cast<const arrow::ListType&>(*field->type());
const auto& nested_field = list_type.value_field();
return write_array_column_to_target(
column, null_map, array_builder, start, end,
[&](const IColumn& nested_data, arrow::ArrayBuilder* nested_builder,
int64_t nested_start, int64_t nested_end) {
return nested_serde->write_column_to_iceberg_arrow(
array_type.get_nested_type(), nested_data, nullptr, nested_field,
nested_builder, nested_start, nested_end, ctz);
});
}

Status DataTypeArraySerDe::read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array,
int64_t start, int64_t end,
const cctz::time_zone& ctz) const {
Expand Down
10 changes: 10 additions & 0 deletions be/src/core/data_type_serde/data_type_array_serde.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,16 @@ class DataTypeArraySerDe : public DataTypeSerDe {
Status write_column_to_arrow(const IColumn& column, const NullMap* null_map,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
const cctz::time_zone& ctz) const override;
Status write_column_to_paimon_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status write_column_to_iceberg_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status read_column_from_orc(IColumn& column, const OrcDecodedColumnView& view) const override;
Expand Down
103 changes: 103 additions & 0 deletions be/src/core/data_type_serde/data_type_map_serde.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
#include "core/column/column.h"
#include "core/column/column_const.h"
#include "core/column/column_map.h"
#include "core/data_type/data_type_map.h"
#include "core/data_type_serde/arrow_validation.h"
#include "core/data_type_serde/complex_type_deserialize_util.h"
#include "core/data_type_serde/orc_serde_utils.h"
Expand Down Expand Up @@ -440,6 +441,108 @@ Status DataTypeMapSerDe::write_column_to_arrow(const IColumn& column, const Null
return Status::OK();
}

namespace {

template <typename WriteKey, typename WriteValue>
Status write_map_column_to_target(const IColumn& column, const NullMap* null_map,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
WriteKey&& write_key, WriteValue&& write_value) {
auto& builder = assert_cast<arrow::MapBuilder&>(*array_builder);
const auto& map_column = assert_cast<const ColumnMap&>(column);
const IColumn& nested_keys_column = map_column.get_keys();
const IColumn& nested_values_column = map_column.get_values();
DCHECK(nested_keys_column.is_nullable());
DCHECK(nested_values_column.is_nullable());
const auto* keys_nullmap_data =
check_and_get_column<ColumnNullable>(nested_keys_column)->get_null_map_data().data();
const auto& offsets = map_column.get_offsets();
auto* key_builder = builder.key_builder();
auto* value_builder = builder.item_builder();

for (size_t row = start; row < end; ++row) {
if (null_map != nullptr && (*null_map)[row]) {
RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, *array_builder));
continue;
}
if (simd::contain_one(keys_nullmap_data + offsets[row - 1],
offsets[row] - offsets[row - 1])) {
return Status::Error(ErrorCode::INVALID_ARGUMENT,
"Can not write null value of map key to arrow.");
}
RETURN_IF_ERROR(checkArrowStatus(builder.Append(), column, *array_builder));
RETURN_IF_ERROR(write_key(nested_keys_column, key_builder, offsets[row - 1], offsets[row]));
RETURN_IF_ERROR(
write_value(nested_values_column, value_builder, offsets[row - 1], offsets[row]));
}
return Status::OK();
}

} // namespace

Status DataTypeMapSerDe::write_column_to_paimon_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column,
const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder,
int64_t start, int64_t end,
const cctz::time_zone& ctz) const {
// Reject an incompatible target before casting its nested schema or builder.
if (field->type()->id() != arrow::Type::MAP ||
array_builder->type()->id() != arrow::Type::MAP) {
return Status::InvalidArgument("Paimon map writer requires an Arrow map field");
}
const auto& map_type = assert_cast<const DataTypeMap&>(*type);
const auto& arrow_map_type = assert_cast<const arrow::MapType&>(*field->type());
const auto& key_field = arrow_map_type.key_field();
const auto& value_field = arrow_map_type.item_field();
return write_map_column_to_target(
column, null_map, array_builder, start, end,
[&](const IColumn& nested_keys, arrow::ArrayBuilder* key_builder, int64_t nested_start,
int64_t nested_end) {
return key_serde->write_column_to_paimon_arrow(map_type.get_key_type(), nested_keys,
nullptr, key_field, key_builder,
nested_start, nested_end, ctz);
},
[&](const IColumn& nested_values, arrow::ArrayBuilder* value_builder,
int64_t nested_start, int64_t nested_end) {
return value_serde->write_column_to_paimon_arrow(
map_type.get_value_type(), nested_values, nullptr, value_field,
value_builder, nested_start, nested_end, ctz);
});
}

Status DataTypeMapSerDe::write_column_to_iceberg_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column,
const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder,
int64_t start, int64_t end,
const cctz::time_zone& ctz) const {
// Reject an incompatible target before casting its nested schema or builder.
if (field->type()->id() != arrow::Type::MAP ||
array_builder->type()->id() != arrow::Type::MAP) {
return Status::InvalidArgument("Iceberg map writer requires an Arrow map field");
}
const auto& map_type = assert_cast<const DataTypeMap&>(*type);
const auto& arrow_map_type = assert_cast<const arrow::MapType&>(*field->type());
const auto& key_field = arrow_map_type.key_field();
const auto& value_field = arrow_map_type.item_field();
return write_map_column_to_target(
column, null_map, array_builder, start, end,
[&](const IColumn& nested_keys, arrow::ArrayBuilder* key_builder, int64_t nested_start,
int64_t nested_end) {
return key_serde->write_column_to_iceberg_arrow(
map_type.get_key_type(), nested_keys, nullptr, key_field, key_builder,
nested_start, nested_end, ctz);
},
[&](const IColumn& nested_values, arrow::ArrayBuilder* value_builder,
int64_t nested_start, int64_t nested_end) {
return value_serde->write_column_to_iceberg_arrow(
map_type.get_value_type(), nested_values, nullptr, value_field,
value_builder, nested_start, nested_end, ctz);
});
}

Status DataTypeMapSerDe::read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array,
int64_t start, int64_t end,
const cctz::time_zone& ctz) const {
Expand Down
10 changes: 10 additions & 0 deletions be/src/core/data_type_serde/data_type_map_serde.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,16 @@ class DataTypeMapSerDe : public DataTypeSerDe {
Status write_column_to_arrow(const IColumn& column, const NullMap* null_map,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
const cctz::time_zone& ctz) const override;
Status write_column_to_paimon_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status write_column_to_iceberg_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status read_column_from_orc(IColumn& column, const OrcDecodedColumnView& view) const override;
Expand Down
23 changes: 23 additions & 0 deletions be/src/core/data_type_serde/data_type_nullable_serde.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type_serde/arrow_validation.h"
#include "core/data_type_serde/data_type_serde.h"
#include "core/data_type_serde/data_type_string_serde.h"
Expand Down Expand Up @@ -385,6 +386,28 @@ Status DataTypeNullableSerDe::write_column_to_arrow(const IColumn& column, const
start, end, ctz);
}

Status DataTypeNullableSerDe::write_column_to_paimon_arrow(
const std::shared_ptr<const IDataType>& type, const IColumn& column, const NullMap*,
const std::shared_ptr<arrow::Field>& field, arrow::ArrayBuilder* array_builder,
int64_t start, int64_t end, const cctz::time_zone& ctz) const {
const auto& nullable_type = assert_cast<const DataTypeNullable&>(*type);
const auto& column_nullable = assert_cast<const ColumnNullable&>(column);
return nested_serde->write_column_to_paimon_arrow(
nullable_type.get_nested_type(), column_nullable.get_nested_column(),
&column_nullable.get_null_map_data(), field, array_builder, start, end, ctz);
}

Status DataTypeNullableSerDe::write_column_to_iceberg_arrow(
const std::shared_ptr<const IDataType>& type, const IColumn& column, const NullMap*,
const std::shared_ptr<arrow::Field>& field, arrow::ArrayBuilder* array_builder,
int64_t start, int64_t end, const cctz::time_zone& ctz) const {
const auto& nullable_type = assert_cast<const DataTypeNullable&>(*type);
const auto& column_nullable = assert_cast<const ColumnNullable&>(column);
return nested_serde->write_column_to_iceberg_arrow(
nullable_type.get_nested_type(), column_nullable.get_nested_column(),
&column_nullable.get_null_map_data(), field, array_builder, start, end, ctz);
}

Status DataTypeNullableSerDe::read_column_from_arrow(IColumn& column,
const arrow::Array* arrow_array, int64_t start,
int64_t end,
Expand Down
10 changes: 10 additions & 0 deletions be/src/core/data_type_serde/data_type_nullable_serde.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,16 @@ class DataTypeNullableSerDe : public DataTypeSerDe {
Status write_column_to_arrow(const IColumn& column, const NullMap* null_map,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
const cctz::time_zone& ctz) const override;
Status write_column_to_paimon_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status write_column_to_iceberg_arrow(const std::shared_ptr<const IDataType>& type,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
int64_t end, const cctz::time_zone& ctz) const override;
Status read_column_from_decoded_values(IColumn& column,
Expand Down
18 changes: 18 additions & 0 deletions be/src/core/data_type_serde/data_type_serde.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
namespace arrow {
class ArrayBuilder;
class Array;
class Field;
} // namespace arrow
namespace cctz {
class time_zone;
Expand Down Expand Up @@ -498,6 +499,23 @@ class DataTypeSerDe {
virtual Status write_column_to_arrow(const IColumn& column, const NullMap* null_map,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const = 0;
// Most scalar types deliberately share their physical Arrow encoding across these protocols.
// Target-specific SerDes override the corresponding method; callers never retry another
// protocol method after an error.
virtual Status write_column_to_paimon_arrow(const std::shared_ptr<const IDataType>&,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>&,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const {
return write_column_to_arrow(column, null_map, array_builder, start, end, ctz);
}
virtual Status write_column_to_iceberg_arrow(const std::shared_ptr<const IDataType>&,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>&,
arrow::ArrayBuilder* array_builder, int64_t start,
int64_t end, const cctz::time_zone& ctz) const {
return write_column_to_arrow(column, null_map, array_builder, start, end, ctz);
}
virtual Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array,
int64_t start, int64_t end,
const cctz::time_zone& ctz) const = 0;
Expand Down
Loading
Loading