diff --git a/include/paimon/file_store_write.h b/include/paimon/file_store_write.h index 8233c42a3..b791777da 100644 --- a/include/paimon/file_store_write.h +++ b/include/paimon/file_store_write.h @@ -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&& batch) = 0; /// Generate a list of commit messages with the latest generated data file meta diff --git a/include/paimon/record_batch.h b/include/paimon/record_batch.h index 0cfe87157..8fa7188de 100644 --- a/include/paimon/record_batch.h +++ b/include/paimon/record_batch.h @@ -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(); diff --git a/src/paimon/common/utils/arrow/arrow_utils.h b/src/paimon/common/utils/arrow/arrow_utils.h index 2a634dc4a..2fc1d1d8d 100644 --- a/src/paimon/common/utils/arrow/arrow_utils.h +++ b/src/paimon/common/utils/arrow/arrow_utils.h @@ -30,7 +30,7 @@ class ArrowUtils { ~ArrowUtils() = delete; static Result> DataTypeToSchema( - const std::shared_ptr<::arrow::DataType>& data_type) { + const std::shared_ptr& data_type) { if (data_type->id() != arrow::Type::STRUCT) { return Status::Invalid(fmt::format("Expected struct data type, actual data type: {}", data_type->ToString())); @@ -40,8 +40,7 @@ class ArrowUtils { } static Result> CreateProjection( - const std::shared_ptr<::arrow::Schema>& file_schema, - const arrow::FieldVector& read_fields) { + const std::shared_ptr& file_schema, const arrow::FieldVector& read_fields) { std::vector target_to_src_mapping; target_to_src_mapping.reserve(read_fields.size()); for (const auto& field : read_fields) { @@ -54,6 +53,54 @@ class ArrowUtils { } return target_to_src_mapping; } + + static Status CheckNullabilityMatch(const std::shared_ptr& schema, + const std::shared_ptr& data) { + auto struct_array = arrow::internal::checked_pointer_cast(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())); + } + 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& field, + const std::shared_ptr& 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(field->type()); + auto struct_array = arrow::internal::checked_pointer_cast(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(field->type()); + auto list_array = arrow::internal::checked_pointer_cast(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(field->type()); + auto map_array = arrow::internal::checked_pointer_cast(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 diff --git a/src/paimon/common/utils/arrow/arrow_utils_test.cpp b/src/paimon/common/utils/arrow/arrow_utils_test.cpp index 11c0e717d..6dd730b4c 100644 --- a/src/paimon/common/utils/arrow/arrow_utils_test.cpp +++ b/src/paimon/common/utils/arrow/arrow_utils_test.cpp @@ -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) { @@ -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 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 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 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 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 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 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 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 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(key_field, value_field); + auto map_field = arrow::field("map_column", map_type, /*nullable=*/false); + auto schema = arrow::schema({map_field}); + + { + std::shared_ptr 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 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 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 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 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 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 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 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 diff --git a/src/paimon/core/operation/abstract_file_store_write.cpp b/src/paimon/core/operation/abstract_file_store_write.cpp index 669ef806e..f6d4537bf 100644 --- a/src/paimon/core/operation/abstract_file_store_write.cpp +++ b/src/paimon/core/operation/abstract_file_store_write.cpp @@ -105,6 +105,13 @@ Status AbstractFileStoreWrite::Write(std::unique_ptr&& batch) { options_.GetBucket())); } } + // check nullability + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr 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 writer, diff --git a/src/paimon/core/table/sink/commit_message_test.cpp b/src/paimon/core/table/sink/commit_message_test.cpp index dfe0f4338..f797da4b5 100644 --- a/src/paimon/core/table/sink/commit_message_test.cpp +++ b/src/paimon/core/table/sink/commit_message_test.cpp @@ -1120,13 +1120,8 @@ TEST(CommitMessageTest, TestCompatibleWithComplexDataType) { TEST(CommitMessageTest, TestSerialize) { arrow::FieldVector fields = { - arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), - arrow::field("f2", arrow::int8()), arrow::field("f3", arrow::int16()), - arrow::field("f4", arrow::int16()), arrow::field("f5", arrow::int32()), - arrow::field("f6", arrow::int32()), arrow::field("f7", arrow::int64()), - arrow::field("f8", arrow::int64()), arrow::field("f9", arrow::float32()), - arrow::field("f10", arrow::float64()), arrow::field("f11", arrow::utf8()), - arrow::field("f12", arrow::binary()), arrow::field("non-partition-field", arrow::int32())}; + arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), + arrow::field("f3", arrow::int16()), arrow::field("non-partition-field", arrow::utf8())}; arrow::Schema typed_schema(fields); ::ArrowSchema schema; @@ -1157,14 +1152,29 @@ TEST(CommitMessageTest, TestSerialize) { FileStoreWrite::Create(std::move(write_context))); for (size_t i = 0; i < 10240; i++) { - auto array = std::make_shared(); - arrow::StringBuilder builder; + auto f0_array = std::make_shared(); + auto f1_array = std::make_shared(); + auto f3_array = std::make_shared(); + auto non_partition_array = std::make_shared(); + arrow::BooleanBuilder f0_builder; + arrow::Int8Builder f1_builder; + arrow::Int16Builder f3_builder; + arrow::StringBuilder non_partition_builder; for (size_t j = 0; j < 100; j++) { - ASSERT_TRUE(builder.Append(std::to_string(j)).ok()); + ASSERT_TRUE(f0_builder.Append(true).ok()); + ASSERT_TRUE(f1_builder.Append(j).ok()); + ASSERT_TRUE(f3_builder.Append(1).ok()); + ASSERT_TRUE(non_partition_builder.Append(std::to_string(j)).ok()); } - ASSERT_TRUE(builder.Finish(&array).ok()); + ASSERT_TRUE(f0_builder.Finish(&f0_array).ok()); + ASSERT_TRUE(f1_builder.Finish(&f1_array).ok()); + ASSERT_TRUE(f3_builder.Finish(&f3_array).ok()); + ASSERT_TRUE(non_partition_builder.Finish(&non_partition_array).ok()); + auto struct_array = + arrow::StructArray::Make({f0_array, f1_array, f3_array, non_partition_array}, fields) + .ValueOrDie(); ::ArrowArray arrow_array; - ASSERT_TRUE(arrow::ExportArray(*array, &arrow_array).ok()); + ASSERT_TRUE(arrow::ExportArray(*struct_array, &arrow_array).ok()); RecordBatchBuilder batch_builder(&arrow_array); ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, batch_builder.SetPartition({{"f0", "true"}, {"f3", "1"}}).Finish()); diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index 125c50e6b..e6e7ae95f 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -593,22 +593,6 @@ TEST_P(BlobTableInteTest, TestOnlySomeColumns) { "Can't infer struct array length with 0 child arrays"); } -TEST_P(BlobTableInteTest, TestNullValues) { - CreateTable(); - std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); - auto schema = arrow::schema(fields_); - - // write field: f0, f1 - std::vector write_cols1 = {"f0", "f1"}; - auto src_array1 = std::dynamic_pointer_cast( - arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([ - [1, null] - ])") - .ValueOrDie()); - ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, write_cols1, {src_array1}), - "BlobFormatWriter only support non-null blob."); -} - TEST_P(BlobTableInteTest, TestMultipleAppendsDifferentFirstRowIds) { CreateTable(); std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); diff --git a/test/inte/write_inte_test.cpp b/test/inte/write_inte_test.cpp index 22e0cd271..0db0b2a46 100644 --- a/test/inte/write_inte_test.cpp +++ b/test/inte/write_inte_test.cpp @@ -3893,4 +3893,50 @@ TEST_P(WriteInteTest, TestAppendTableWithDateFieldAsPartitionField) { ASSERT_TRUE(success); } +TEST_P(WriteInteTest, TestNullabilityCheck) { + auto dir = UniqueTestDirectory::Create(); + arrow::FieldVector fields = {arrow::field("f0", arrow::utf8(), /*nullable=*/true), + arrow::field("f1", arrow::int32(), /*nullable=*/false)}; + auto schema = arrow::schema(fields); + auto file_format = GetParam(); + std::map options = { + {Options::FILE_FORMAT, StringUtils::ToUpperCase(file_format)}, + {Options::TARGET_FILE_SIZE, "1024"}, + {Options::BUCKET, "-1"}, + }; + ASSERT_OK_AND_ASSIGN( + auto helper, TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{}, + /*primary_keys=*/{}, options, /*is_streaming_mode=*/false)); + int64_t commit_identifier = 0; + + // write invalid data + std::string data = R"([ + ["banana", 2], + ["dog", 1], + [null, 14], + ["mouse", null] + ])"; + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), data, + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_NOK_WITH_MSG( + helper->WriteAndCommit(std::move(batch), commit_identifier++, + /*expected_commit_messages=*/std::nullopt), + "CheckNullabilityMatch failed, field f1 not nullable while data have null value"); + + // write valid data + data = R"([ + ["banana", 2], + ["dog", 1], + [null, 14], + ["mouse", 100] + ])"; + ASSERT_OK_AND_ASSIGN(batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), data, + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK_AND_ASSIGN(auto commit_msgs, + helper->WriteAndCommit(std::move(batch), commit_identifier++, + /*expected_commit_messages=*/std::nullopt)); +} + } // namespace paimon::test