Skip to content

fix: check data batch nullability match with schema in write process #122

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
zjw1111 merged 5 commits into from
Feb 6, 2026
Merged

fix: check data batch nullability match with schema in write process #122

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9ef9111
fix: check data batch nullability match with schema in write process
lszskye Feb 6, 2026
1eb6106
fix
lszskye Feb 6, 2026
b5e5bc4
fix0306
lszskye Feb 6, 2026
aac5bdf
fix0417
lszskye Feb 6, 2026
2044efb
fix0611
lszskye Feb 6, 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
2 changes: 2 additions & 0 deletions include/paimon/file_store_write.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,8 @@ class PAIMON_EXPORT FileStoreWrite {
virtual ~FileStoreWrite() = default;

/// Support write an input `RecordBatch` to internal buffer or file.
/// @note If a field in table schema is marked as non-nullable (`nullable = false`),
/// the corresponding array in `batch` must have zero null entries.
virtual Status Write(std::unique_ptr<RecordBatch>&& batch) = 0;

/// Generate a list of commit messages with the latest generated data file meta
Expand Down
13 changes: 11 additions & 2 deletions include/paimon/record_batch.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,8 +93,17 @@ class PAIMON_EXPORT RecordBatch {
/// various properties such as data, row kinds, partition information, and bucket id.
class PAIMON_EXPORT RecordBatchBuilder {
public:
/// Constructs a `RecordBatchBuilder` with Arrow data.
/// @param data Arrow array containing the record data.
/// Constructs a `RecordBatchBuilder` with Arrow data
///
/// @note The `data` must conform to table schema:
/// - Each array in `data` corresponds to a field in table schema.
/// - If a field in table schema is marked as non-nullable (`nullable = false`),
/// the corresponding array in `data` must have zero null entries.
///
/// @note Consistency between `data` and table schema will be validated during the write
/// process.
///
/// @param data ArrowArray struct containing the columnar data (via C Data Interface)
explicit RecordBatchBuilder(::ArrowArray* data);

~RecordBatchBuilder();
Expand Down
53 changes: 50 additions & 3 deletions src/paimon/common/utils/arrow/arrow_utils.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ class ArrowUtils {
~ArrowUtils() = delete;

static Result<std::shared_ptr<arrow::Schema>> DataTypeToSchema(
const std::shared_ptr<::arrow::DataType>& data_type) {
const std::shared_ptr<arrow::DataType>& data_type) {
if (data_type->id() != arrow::Type::STRUCT) {
return Status::Invalid(fmt::format("Expected struct data type, actual data type: {}",
data_type->ToString()));
Expand All @@ -40,8 +40,7 @@ class ArrowUtils {
}

static Result<std::vector<int32_t>> CreateProjection(
const std::shared_ptr<::arrow::Schema>& file_schema,
const arrow::FieldVector& read_fields) {
const std::shared_ptr<arrow::Schema>& file_schema, const arrow::FieldVector& read_fields) {
std::vector<int32_t> target_to_src_mapping;
target_to_src_mapping.reserve(read_fields.size());
for (const auto& field : read_fields) {
Expand All @@ -54,6 +53,54 @@ class ArrowUtils {
}
return target_to_src_mapping;
}

static Status CheckNullabilityMatch(const std::shared_ptr<arrow::Schema>& schema,
const std::shared_ptr<arrow::Array>& data) {
auto struct_array = arrow::internal::checked_pointer_cast<arrow::StructArray>(data);
if (struct_array->num_fields() != schema->num_fields()) {
return Status::Invalid(fmt::format(
"CheckNullabilityMatch failed, data field count {} mismatch schema field count {}",
struct_array->num_fields(), schema->num_fields()));
}
Comment thread
lszskye marked this conversation as resolved.
for (int32_t i = 0; i < schema->num_fields(); i++) {
PAIMON_RETURN_NOT_OK(
InnerCheckNullabilityMatch(schema->field(i), struct_array->field(i)));
}
return Status::OK();
}

private:
static Status InnerCheckNullabilityMatch(const std::shared_ptr<arrow::Field>& field,
const std::shared_ptr<arrow::Array>& data) {
if (PAIMON_UNLIKELY(!field->nullable() && data->null_count() != 0)) {
return Status::Invalid(fmt::format(
"CheckNullabilityMatch failed, field {} not nullable while data have null value",
field->name()));
}
auto type = field->type();
if (type->id() == arrow::Type::STRUCT) {
auto struct_type =
arrow::internal::checked_pointer_cast<arrow::StructType>(field->type());
auto struct_array = arrow::internal::checked_pointer_cast<arrow::StructArray>(data);
for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
PAIMON_RETURN_NOT_OK(
InnerCheckNullabilityMatch(struct_type->field(i), struct_array->field(i)));
}
} else if (type->id() == arrow::Type::LIST) {
auto list_type = arrow::internal::checked_pointer_cast<arrow::ListType>(field->type());
auto list_array = arrow::internal::checked_pointer_cast<arrow::ListArray>(data);
PAIMON_RETURN_NOT_OK(
InnerCheckNullabilityMatch(list_type->value_field(), list_array->values()));
} else if (type->id() == arrow::Type::MAP) {
auto map_type = arrow::internal::checked_pointer_cast<arrow::MapType>(field->type());
auto map_array = arrow::internal::checked_pointer_cast<arrow::MapArray>(data);
PAIMON_RETURN_NOT_OK(
InnerCheckNullabilityMatch(map_type->key_field(), map_array->keys()));
PAIMON_RETURN_NOT_OK(
InnerCheckNullabilityMatch(map_type->item_field(), map_array->items()));
}
return Status::OK();
}
};

} // namespace paimon
231 changes: 230 additions & 1 deletion src/paimon/common/utils/arrow/arrow_utils_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,10 @@
#include "paimon/common/utils/arrow/arrow_utils.h"

#include "arrow/api.h"
#include "arrow/ipc/api.h"
#include "gtest/gtest.h"
#include "paimon/common/types/data_field.h"
#include "paimon/testing/utils/testharness.h"

namespace paimon::test {

TEST(ArrowUtilsTest, TestCreateProjection) {
Expand Down Expand Up @@ -107,4 +108,232 @@ TEST(ArrowUtilsTest, TestCreateProjection) {
}
}

TEST(ArrowUtilsTest, TestCheckNullableMatchSimple) {
auto field = arrow::field("column1", arrow::int32(), /*nullable=*/false);
auto schema = arrow::schema({field});
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({field}), R"([
[20],
[null],
[10]
])")
.ValueOrDie();

ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field column1 not nullable while data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({field}), R"([
[20],
[10]
])")
.ValueOrDie();

ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
}
}

TEST(ArrowUtilsTest, TestCheckNullableMatchWithStruct) {
auto child1 = arrow::field("child1", arrow::int32(), /*nullable=*/false);
auto child2 = arrow::field("child2", arrow::float64(), /*nullable=*/true);
auto struct_field =
arrow::field("parent", arrow::struct_({child1, child2}), /*nullable=*/false);
auto schema = arrow::schema({struct_field});
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({struct_field}), R"([
[null]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field parent not nullable while data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({struct_field}), R"([
[[1, null]],
[[null, 10.0]]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field child1 not nullable while data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({struct_field}), R"([
[[1, null]],
[[2, 10.0]]
])")
.ValueOrDie();
ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
}
}

TEST(ArrowUtilsTest, TestCheckNullableMatchWithList) {
auto value_field = arrow::field("value", arrow::int32(), /*nullable=*/false);
auto list_field = arrow::field("list_column", arrow::list(value_field), /*nullable=*/false);
auto schema = arrow::schema({list_field});

{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({list_field}), R"([
[[1, 2, null, 4, 5]],
[null]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field list_column not nullable while "
"data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({list_field}), R"([
[[1, 2, null, 4, 5]]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field value not nullable while data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({list_field}), R"([
[[1, 2, 3, 4, 5]]
])")
.ValueOrDie();
ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
}
}

TEST(ArrowUtilsTest, TestCheckNullableMatchWithMap) {
auto key_field = arrow::field("key", arrow::int32(), /*nullable=*/false);
auto value_field = arrow::field("value", arrow::int32(), /*nullable=*/true);
auto map_type = std::make_shared<arrow::MapType>(key_field, value_field);
auto map_field = arrow::field("map_column", map_type, /*nullable=*/false);
auto schema = arrow::schema({map_field});

{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({map_field}), R"([
[null]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field map_column not nullable while "
"data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({map_field}), R"([
[[[1, null]]]
])")
.ValueOrDie();
ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
}
}

TEST(ArrowUtilsTest, TestCheckNullableMatchComplex) {
auto key_field = arrow::field("key", arrow::int32(), /*nullable=*/false);
auto value_field = arrow::field("value", arrow::int32(), /*nullable=*/false);

auto inner_child1 =
arrow::field("inner1",
arrow::map(arrow::utf8(), arrow::field("inner_list", arrow::list(value_field),
/*nullable=*/true)),
/*nullable=*/false);
auto inner_child2 = arrow::field(
"inner2",
arrow::map(arrow::utf8(), arrow::field("inner_map", arrow::map(arrow::utf8(), value_field),
/*nullable=*/true)),
/*nullable=*/false);
auto inner_child3 = arrow::field(
"inner3",
arrow::map(arrow::utf8(),
arrow::field("inner_struct", arrow::struct_({key_field, value_field}),
/*nullable=*/true)),
/*nullable=*/false);

auto schema = arrow::schema({inner_child1, inner_child2, inner_child3});
// test inner1
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({inner_child1, inner_child2, inner_child3}), R"([
[[["outer_key", [1, 2, 3, null]]], [["outer_key", [["key1", 1]]]], [["outer_key", [100, 200]]]]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field value not nullable while data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({inner_child1, inner_child2, inner_child3}), R"([
[[["outer_key", [1, 2, 3]]], [["outer_key", [["key1", 1]]]], [["outer_key", [100, 200]]]],
[[["outer_key", null]], [["outer_key", [["key1", 1]]]], [["outer_key", [100, 200]]]],
[null, [["outer_key", [["key1", 1]]]], [["outer_key", [100, 200]]]]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field inner1 not nullable while data have null value");
}
// test inner2
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({inner_child1, inner_child2, inner_child3}), R"([
[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", [100, 200]]]],
[[["outer_key", null]], [["outer_key", [["key1", null]]]], [["outer_key", [100, 200]]]]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field value not nullable while data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({inner_child1, inner_child2, inner_child3}), R"([
[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", [100, 200]]]],
[[["outer_key", [1, 2, 3]]], null, [["outer_key", [100, 200]]]]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field inner2 not nullable while data have null value");
}
// test inner3
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({inner_child1, inner_child2, inner_child3}), R"([
[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", null]]],
[[["outer_key", null]], [["outer_key", [["key1", 2]]]], [["outer_key", [100, null]]]]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field value not nullable while data have null value");
}
{
std::shared_ptr<arrow::Array> array =
arrow::ipc::internal::json::ArrayFromJSON(
arrow::struct_({inner_child1, inner_child2, inner_child3}), R"([
[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", null]]],
[[["outer_key", null]], [["outer_key", [["key1", 2]]]], null]
])")
.ValueOrDie();
ASSERT_NOK_WITH_MSG(
ArrowUtils::CheckNullabilityMatch(schema, array),
"CheckNullabilityMatch failed, field inner3 not nullable while data have null value");
}
}

} // namespace paimon::test
7 changes: 7 additions & 0 deletions src/paimon/core/operation/abstract_file_store_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,13 @@ Status AbstractFileStoreWrite::Write(std::unique_ptr<RecordBatch>&& batch) {
options_.GetBucket()));
}
}
// check nullability
Comment thread
lxy-9602 marked this conversation as resolved.
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
std::shared_ptr<arrow::Array> data,
arrow::ImportArray(batch->GetData(), arrow::struct_(write_schema_->fields())));
PAIMON_RETURN_NOT_OK(ArrowUtils::CheckNullabilityMatch(write_schema_, data));
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*data, batch->GetData()));

PAIMON_ASSIGN_OR_RAISE(BinaryRow partition,
file_store_path_factory_->ToBinaryRow(batch->GetPartition()))
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<BatchWriter> writer,
Expand Down
Loading
Loading