From 3b56fb36e649893f51a68be7e18d4959da34fe89 Mon Sep 17 00:00:00 2001 From: lxy264173 Date: Thu, 16 Jul 2026 16:22:26 +0800 Subject: [PATCH 1/3] fix(blob): support blob-only writes and reject nested blob fields --- src/paimon/common/data/blob_utils.cpp | 9 +- src/paimon/common/data/blob_utils.h | 3 +- src/paimon/common/data/blob_utils_test.cpp | 8 +- src/paimon/core/append/append_only_writer.cpp | 10 +- .../core/append/append_only_writer_test.cpp | 34 +++++++ .../core/io/rolling_blob_file_writer.cpp | 39 +++++--- src/paimon/core/io/rolling_blob_file_writer.h | 3 +- .../core/schema/arrow_schema_validator.cpp | 48 +++++++--- .../core/schema/arrow_schema_validator.h | 3 +- .../schema/arrow_schema_validator_test.cpp | 49 +++++++++- test/inte/blob_table_inte_test.cpp | 92 ++++++++++++++++--- 11 files changed, 239 insertions(+), 59 deletions(-) diff --git a/src/paimon/common/data/blob_utils.cpp b/src/paimon/common/data/blob_utils.cpp index 7bfeb9223..9f0e734f7 100644 --- a/src/paimon/common/data/blob_utils.cpp +++ b/src/paimon/common/data/blob_utils.cpp @@ -82,13 +82,12 @@ Result BlobUtils::SeparateBlobArray( return Status::Invalid( "SeparateBlobArray expects at least one non-inline blob field, but got none."); } - if (main_fields.empty()) { - return Status::Invalid("SeparateBlobArray expects at least one main field, but got none."); - } SeparatedStructArrays result; - PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(result.main_array, - arrow::StructArray::Make(main_arrays, main_fields)); + if (!main_fields.empty()) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(result.main_array, + arrow::StructArray::Make(main_arrays, main_fields)); + } PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(result.blob_array, arrow::StructArray::Make(blob_arrays, blob_fields)); return result; diff --git a/src/paimon/common/data/blob_utils.h b/src/paimon/common/data/blob_utils.h index 196355260..30a303899 100644 --- a/src/paimon/common/data/blob_utils.h +++ b/src/paimon/common/data/blob_utils.h @@ -51,7 +51,8 @@ class PAIMON_EXPORT BlobUtils { }; struct SeparatedStructArrays { - /// Non-blob fields (includes inline blob fields when inline_fields is provided) + /// Non-blob fields (includes inline blob fields when inline_fields is provided). + /// nullptr when all fields are stored in blob files. std::shared_ptr main_array; /// Blob fields that go to separate .blob files std::shared_ptr blob_array; diff --git a/src/paimon/common/data/blob_utils_test.cpp b/src/paimon/common/data/blob_utils_test.cpp index d42845f10..50557a2f3 100644 --- a/src/paimon/common/data/blob_utils_test.cpp +++ b/src/paimon/common/data/blob_utils_test.cpp @@ -182,11 +182,13 @@ TEST_F(BlobUtilsTest, SeparateBlobArray) { BlobUtils::SeparateBlobArray(struct_array, /*inline_fields=*/{"f2_blob"}), "SeparateBlobArray expects at least one non-inline blob field, but got none."); - // All fields are blob with no inline -> no main field -> should return error + // All fields are blob with no inline -> no main array is needed auto all_blob_struct = arrow::StructArray::Make({blob_array_data}, {blob_field}).ValueOrDie(); auto all_blob_sa = std::dynamic_pointer_cast(all_blob_struct); - ASSERT_NOK_WITH_MSG(BlobUtils::SeparateBlobArray(all_blob_sa, /*inline_fields=*/{}), - "SeparateBlobArray expects at least one main field, but got none."); + ASSERT_OK_AND_ASSIGN(auto all_blob_separated, + BlobUtils::SeparateBlobArray(all_blob_sa, /*inline_fields=*/{})); + ASSERT_EQ(nullptr, all_blob_separated.main_array); + ASSERT_TRUE(all_blob_separated.blob_array->Equals(*all_blob_sa)); } TEST_F(BlobUtilsTest, SeparateBlobArrayWithPartialInline) { diff --git a/src/paimon/core/append/append_only_writer.cpp b/src/paimon/core/append/append_only_writer.cpp index 43d1e2bd4..21d2b8db7 100644 --- a/src/paimon/core/append/append_only_writer.cpp +++ b/src/paimon/core/append/append_only_writer.cpp @@ -238,10 +238,14 @@ AppendOnlyWriter::RollingFileWriterResult AppendOnlyWriter::CreateRollingBlobWri options_.GetBlobTargetFileSize(), single_blob_file_writer_factory); }; + WriterFactory main_writer_factory; + if (schemas.main_schema->num_fields() > 0) { + main_writer_factory = + GetDataFileWriterFactory(schemas.main_schema, schemas.main_schema->field_names()); + } return std::make_unique( - options_.GetTargetFileSize(/*has_primary_key=*/false), - GetDataFileWriterFactory(schemas.main_schema, schemas.main_schema->field_names()), - blob_schema, blob_writer_creator, arrow::struct_(write_schema_->fields()), inline_fields); + options_.GetTargetFileSize(/*has_primary_key=*/false), main_writer_factory, blob_schema, + blob_writer_creator, arrow::struct_(write_schema_->fields()), inline_fields); } Status AppendOnlyWriter::Sync() { diff --git a/src/paimon/core/append/append_only_writer_test.cpp b/src/paimon/core/append/append_only_writer_test.cpp index 2d3d05427..df7d912aa 100644 --- a/src/paimon/core/append/append_only_writer_test.cpp +++ b/src/paimon/core/append/append_only_writer_test.cpp @@ -741,6 +741,40 @@ TEST_F(AppendOnlyWriterTest, TestWriteWithSingleBlobField) { ASSERT_OK(writer->Close()); } +TEST_F(AppendOnlyWriterTest, TestWriteWithOnlyBlobField) { + auto options = + CreateOptions({{Options::FILE_FORMAT, "orc"}, {Options::MANIFEST_FORMAT, "orc"}}); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto path_factory = CreatePathFactory(dir->Str(), "orc", options); + + auto blob_field = BlobUtils::ToArrowField("blob", false); + auto schema = arrow::schema({blob_field}); + ASSERT_OK_AND_ASSIGN(auto writer, + CreateAppendOnlyWriter(options, /*schema_id=*/0, schema, + /*write_cols=*/std::vector{"blob"}, + /*max_sequence_number=*/-1, path_factory, + compact_manager_, memory_pool_)); + + arrow::LargeBinaryBuilder blob_builder; + ASSERT_TRUE(blob_builder.Append("a", 1).ok()); + ASSERT_TRUE(blob_builder.Append("bb", 2).ok()); + auto blob_array = blob_builder.Finish().ValueOrDie(); + + ASSERT_OK(writer->Write(CreateStructBatch(schema, {blob_array}))); + ASSERT_OK_AND_ASSIGN(CommitIncrement inc, writer->PrepareCommit(/*wait_compaction=*/true)); + + const auto& new_files = inc.GetNewFilesIncrement().NewFiles(); + ASSERT_EQ(new_files.size(), 1); + ASSERT_TRUE(BlobUtils::IsBlobFile(new_files[0]->file_name)); + ASSERT_EQ(new_files[0]->row_count, 2); + ASSERT_TRUE(new_files[0]->write_cols.has_value()); + ASSERT_EQ(new_files[0]->write_cols.value(), std::vector({"blob"})); + ASSERT_TRUE( + options.GetFileSystem()->Exists(path_factory->ToPath(new_files[0]->file_name)).value()); + ASSERT_OK(writer->Close()); +} + TEST_F(AppendOnlyWriterTest, TestWriteWithMultipleBlobFields) { auto options = CreateOptions({{Options::FILE_FORMAT, "orc"}, {Options::MANIFEST_FORMAT, "orc"}}); diff --git a/src/paimon/core/io/rolling_blob_file_writer.cpp b/src/paimon/core/io/rolling_blob_file_writer.cpp index c48a8bbeb..a8c9f4f8b 100644 --- a/src/paimon/core/io/rolling_blob_file_writer.cpp +++ b/src/paimon/core/io/rolling_blob_file_writer.cpp @@ -55,7 +55,7 @@ RollingBlobFileWriter::RollingBlobFileWriter( Status RollingBlobFileWriter::Write(::ArrowArray* record) { ScopeGuard guard([this]() -> void { this->Abort(); }); // Open the current writer if write the first record or roll over happen before. - if (PAIMON_UNLIKELY(current_writer_ == nullptr)) { + if (writer_factory_ != nullptr && PAIMON_UNLIKELY(current_writer_ == nullptr)) { PAIMON_RETURN_NOT_OK(OpenCurrentWriter()); } if (PAIMON_UNLIKELY(blob_writer_ == nullptr)) { @@ -69,12 +69,14 @@ Status RollingBlobFileWriter::Write(::ArrowArray* record) { PAIMON_ASSIGN_OR_RAISE(BlobUtils::SeparatedStructArrays separated_arrays, BlobUtils::SeparateBlobArray(struct_array, inline_fields_)); // Write main (non-blob) data - ::ArrowArray c_main_array; - PAIMON_RETURN_NOT_OK_FROM_ARROW( - arrow::ExportArray(*separated_arrays.main_array, &c_main_array)); - ScopeGuard array_lifecycle_guard( - [&c_main_array]() -> void { ArrowArrayRelease(&c_main_array); }); - PAIMON_RETURN_NOT_OK(current_writer_->Write(&c_main_array)); + if (current_writer_ != nullptr) { + ::ArrowArray c_main_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW( + arrow::ExportArray(*separated_arrays.main_array, &c_main_array)); + ScopeGuard array_lifecycle_guard( + [&c_main_array]() -> void { ArrowArrayRelease(&c_main_array); }); + PAIMON_RETURN_NOT_OK(current_writer_->Write(&c_main_array)); + } // Write blob data via MultipleBlobFileWriter (each blob field independently) ::ArrowArray c_blob_array; @@ -84,28 +86,35 @@ Status RollingBlobFileWriter::Write(::ArrowArray* record) { PAIMON_RETURN_NOT_OK(blob_writer_->Write(&c_blob_array)); record_count_ += record_count; - PAIMON_ASSIGN_OR_RAISE(bool need_rolling_file, NeedRollingFile()); - if (need_rolling_file) { - PAIMON_RETURN_NOT_OK(CloseCurrentWriter()); + if (current_writer_ != nullptr) { + PAIMON_ASSIGN_OR_RAISE(bool need_rolling_file, NeedRollingFile()); + if (need_rolling_file) { + PAIMON_RETURN_NOT_OK(CloseCurrentWriter()); + } } guard.Release(); return Status::OK(); } Status RollingBlobFileWriter::CloseCurrentWriter() { - if (current_writer_ == nullptr) { + if (current_writer_ == nullptr && blob_writer_ == nullptr) { return Status::OK(); } if (blob_writer_ == nullptr) { return Status::OK(); } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr main_data_file_meta, CloseMainWriter()); + std::shared_ptr main_data_file_meta; + if (current_writer_ != nullptr) { + PAIMON_ASSIGN_OR_RAISE(main_data_file_meta, CloseMainWriter()); + } PAIMON_ASSIGN_OR_RAISE(std::vector> blob_metas, CloseBlobWriter()); - PAIMON_RETURN_NOT_OK( - ValidateFileConsistency(main_data_file_meta, blob_metas, blob_schema_->num_fields())); - results_.push_back(main_data_file_meta); + if (main_data_file_meta != nullptr) { + PAIMON_RETURN_NOT_OK( + ValidateFileConsistency(main_data_file_meta, blob_metas, blob_schema_->num_fields())); + results_.push_back(main_data_file_meta); + } results_.insert(results_.end(), blob_metas.begin(), blob_metas.end()); current_writer_.reset(); diff --git a/src/paimon/core/io/rolling_blob_file_writer.h b/src/paimon/core/io/rolling_blob_file_writer.h index 324772059..0aa998f3b 100644 --- a/src/paimon/core/io/rolling_blob_file_writer.h +++ b/src/paimon/core/io/rolling_blob_file_writer.h @@ -38,7 +38,8 @@ namespace paimon { /// between them. /// /// Multiple blob fields are supported. Each blob field is written to its own set of blob files -/// independently via MultipleBlobFileWriter. +/// independently via MultipleBlobFileWriter. For blob-only writes, the main writer factory may be +/// nullptr and only blob files are produced. /// /// 
 /// For example,
diff --git a/src/paimon/core/schema/arrow_schema_validator.cpp b/src/paimon/core/schema/arrow_schema_validator.cpp
index c9053475c..cb8989cf0 100644
--- a/src/paimon/core/schema/arrow_schema_validator.cpp
+++ b/src/paimon/core/schema/arrow_schema_validator.cpp
@@ -55,10 +55,19 @@ Status ArrowSchemaValidator::ValidateSchema(const arrow::Schema& schema) {
 
 Status ArrowSchemaValidator::ValidateSchemaWithFieldId(const arrow::Schema& schema) {
     PAIMON_RETURN_NOT_OK(ValidateSchema(schema));
-    auto struct_type = arrow::struct_(schema.fields());
     std::set field_id_set;
-    PAIMON_RETURN_NOT_OK(
-        ValidateDataTypeWithFieldId(struct_type, /*key_value_metadata=*/nullptr, &field_id_set));
+    for (const auto& field : schema.fields()) {
+        PAIMON_ASSIGN_OR_RAISE(DataField data_field,
+                               DataField::ConvertArrowFieldToDataField(field));
+        auto iter = field_id_set.find(data_field.Id());
+        if (iter != field_id_set.end()) {
+            return Status::Invalid(
+                fmt::format("field id must be unique, duplicate field id {}", data_field.Id()));
+        }
+        field_id_set.insert(data_field.Id());
+        PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(field->type(), field->metadata(),
+                                                         /*allow_blob=*/true, &field_id_set));
+    }
     return Status::OK();
 }
 
@@ -93,7 +102,7 @@ Status ArrowSchemaValidator::ValidateNoWhitespaceOnlyFields(const arrow::FieldVe
 
 Status ArrowSchemaValidator::ValidateDataTypeWithFieldId(
     const std::shared_ptr& type,
-    const std::shared_ptr& key_value_metadata,
+    const std::shared_ptr& key_value_metadata, bool allow_blob,
     std::set* field_id_set) {
     const auto kind = type->id();
     switch (kind) {
@@ -114,7 +123,7 @@ Status ArrowSchemaValidator::ValidateDataTypeWithFieldId(
             const auto& value_field =
                 arrow::internal::checked_cast(type.get())->value_field();
             PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(
-                value_field->type(), value_field->metadata(), field_id_set));
+                value_field->type(), value_field->metadata(), /*allow_blob=*/false, field_id_set));
             break;
         }
         case arrow::Type::type::STRUCT: {
@@ -135,7 +144,7 @@ Status ArrowSchemaValidator::ValidateDataTypeWithFieldId(
                 }
                 field_id_set->insert(data_field.Id());
                 PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(
-                    sub_field->type(), sub_field->metadata(), field_id_set));
+                    sub_field->type(), sub_field->metadata(), /*allow_blob=*/false, field_id_set));
             }
             break;
         }
@@ -144,14 +153,17 @@ Status ArrowSchemaValidator::ValidateDataTypeWithFieldId(
                 arrow::internal::checked_cast(type.get())->key_field();
             const auto& item_field =
                 arrow::internal::checked_cast(type.get())->item_field();
-            PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(key_field->type(),
-                                                             key_field->metadata(), field_id_set));
-            PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(item_field->type(),
-                                                             item_field->metadata(), field_id_set));
+            PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(
+                key_field->type(), key_field->metadata(), /*allow_blob=*/false, field_id_set));
+            PAIMON_RETURN_NOT_OK(ValidateDataTypeWithFieldId(
+                item_field->type(), item_field->metadata(), /*allow_blob=*/false, field_id_set));
             break;
         }
         case arrow::Type::type::LARGE_BINARY: {
             if (BlobUtils::IsBlobMetadata(key_value_metadata)) {
+                if (!allow_blob) {
+                    return Status::Invalid("Blob field must be a top-level field.");
+                }
                 break;
             }
             [[fallthrough]];
@@ -164,6 +176,11 @@ Status ArrowSchemaValidator::ValidateDataTypeWithFieldId(
 }
 
 Status ArrowSchemaValidator::ValidateField(const std::shared_ptr& field) {
+    return ValidateField(field, /*allow_blob=*/true);
+}
+
+Status ArrowSchemaValidator::ValidateField(const std::shared_ptr& field,
+                                           bool allow_blob) {
     const auto kind = field->type()->id();
     switch (kind) {
         case arrow::Type::type::BOOL:
@@ -185,7 +202,7 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr&
             const auto& value_field =
                 arrow::internal::checked_cast(*field->type())
                     .value_field();
-            PAIMON_RETURN_NOT_OK(ValidateField(value_field));
+            PAIMON_RETURN_NOT_OK(ValidateField(value_field, /*allow_blob=*/false));
             break;
         }
         case arrow::Type::type::STRUCT: {
@@ -202,7 +219,7 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr&
             arrow::FieldVector arrow_fields =
                 arrow::internal::checked_cast(*field->type()).fields();
             for (const auto& sub_field : arrow_fields) {
-                PAIMON_RETURN_NOT_OK(ValidateField(sub_field));
+                PAIMON_RETURN_NOT_OK(ValidateField(sub_field, /*allow_blob=*/false));
             }
             break;
         }
@@ -211,12 +228,15 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr&
                 arrow::internal::checked_cast(*field->type()).key_field();
             const auto& item_field =
                 arrow::internal::checked_cast(*field->type()).item_field();
-            PAIMON_RETURN_NOT_OK(ValidateField(key_field));
-            PAIMON_RETURN_NOT_OK(ValidateField(item_field));
+            PAIMON_RETURN_NOT_OK(ValidateField(key_field, /*allow_blob=*/false));
+            PAIMON_RETURN_NOT_OK(ValidateField(item_field, /*allow_blob=*/false));
             break;
         }
         case arrow::Type::type::LARGE_BINARY: {
             if (BlobUtils::IsBlobField(field)) {
+                if (!allow_blob) {
+                    return Status::Invalid("Blob field must be a top-level field.");
+                }
                 break;
             }
             [[fallthrough]];
diff --git a/src/paimon/core/schema/arrow_schema_validator.h b/src/paimon/core/schema/arrow_schema_validator.h
index 39c6ef77c..940f2ab80 100644
--- a/src/paimon/core/schema/arrow_schema_validator.h
+++ b/src/paimon/core/schema/arrow_schema_validator.h
@@ -55,7 +55,8 @@ class PAIMON_EXPORT ArrowSchemaValidator {
  private:
     static Status ValidateDataTypeWithFieldId(
         const std::shared_ptr& type,
-        const std::shared_ptr& key_value_metadata,
+        const std::shared_ptr& key_value_metadata, bool allow_blob,
         std::set* field_id_set);
+    static Status ValidateField(const std::shared_ptr& field, bool allow_blob);
 };
 }  // namespace paimon
diff --git a/src/paimon/core/schema/arrow_schema_validator_test.cpp b/src/paimon/core/schema/arrow_schema_validator_test.cpp
index 310955622..ef4375893 100644
--- a/src/paimon/core/schema/arrow_schema_validator_test.cpp
+++ b/src/paimon/core/schema/arrow_schema_validator_test.cpp
@@ -21,6 +21,7 @@
 
 #include "arrow/type.h"
 #include "gtest/gtest.h"
+#include "paimon/common/data/blob_utils.h"
 #include "paimon/common/data/variant/variant_access_utils.h"
 #include "paimon/common/data/variant/variant_defs.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
@@ -151,6 +152,51 @@ TEST(ArrowSchemaValidatorTest, TestInvalidDataType) {
     }
 }
 
+TEST(ArrowSchemaValidatorTest, TestBlobFieldMustBeTopLevel) {
+    {
+        auto arrow_schema =
+            arrow::schema(arrow::FieldVector({BlobUtils::ToArrowField("blob", true)}));
+        ASSERT_OK(ArrowSchemaValidator::ValidateSchema(*arrow_schema));
+    }
+    {
+        std::vector fields = {DataField(0, BlobUtils::ToArrowField("blob", true))};
+        auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(fields);
+        ASSERT_OK(ArrowSchemaValidator::ValidateSchemaWithFieldId(*arrow_schema));
+    }
+    {
+        auto nested_blob_field =
+            arrow::field("nested", arrow::struct_({BlobUtils::ToArrowField("blob", true)}));
+        auto arrow_schema = arrow::schema(arrow::FieldVector({nested_blob_field}));
+        ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchema(*arrow_schema),
+                            "Blob field must be a top-level field.");
+    }
+    {
+        auto array_blob_field =
+            arrow::field("array_blob", arrow::list(BlobUtils::ToArrowField("item", true)));
+        auto arrow_schema = arrow::schema(arrow::FieldVector({array_blob_field}));
+        ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchema(*arrow_schema),
+                            "Blob field must be a top-level field.");
+    }
+    {
+        auto map_blob_field = arrow::field(
+            "map_blob",
+            arrow::map(arrow::utf8(), arrow::struct_({BlobUtils::ToArrowField("blob", true)})));
+        auto arrow_schema = arrow::schema(arrow::FieldVector({map_blob_field}));
+        ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchema(*arrow_schema),
+                            "Blob field must be a top-level field.");
+    }
+    {
+        std::vector nested_fields = {
+            DataField(1, BlobUtils::ToArrowField("blob", true))};
+        std::vector fields = {DataField(
+            0,
+            arrow::field("nested", DataField::ConvertDataFieldsToArrowStructType(nested_fields)))};
+        auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(fields);
+        ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchemaWithFieldId(*arrow_schema),
+                            "Blob field must be a top-level field.");
+    }
+}
+
 TEST(ArrowSchemaValidatorTest, ValidateDataTypeWithFieldId) {
     {
         std::vector fields = {DataField(3, arrow::field("f3", arrow::float64())),
@@ -268,7 +314,8 @@ TEST(ArrowSchemaValidatorTest, ValidateDataTypeWithFieldId) {
         auto struct_type = DataField::ConvertDataFieldsToArrowStructType(fields);
         std::set field_id_set;
         ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateDataTypeWithFieldId(
-                                struct_type, /*key_value_metadata=*/nullptr, &field_id_set),
+                                struct_type, /*key_value_metadata=*/nullptr,
+                                /*allow_blob=*/true, &field_id_set),
                             "Unknown or unsupported arrow type: large_string");
     }
 }
diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp
index 1159a9518..a73c73df6 100644
--- a/test/inte/blob_table_inte_test.cpp
+++ b/test/inte/blob_table_inte_test.cpp
@@ -997,30 +997,92 @@ TEST_P(BlobTableInteTest, TestMultipleAppends) {
     }
 }
 
-TEST_P(BlobTableInteTest, TestOnlySomeColumns) {
-    CreateTable();
+TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyWriteWithFirstRowId) {
+    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
+                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
+    std::map options = {{Options::MANIFEST_FORMAT, "orc"},
+                                                  {Options::FILE_FORMAT, GetParam()},
+                                                  {Options::FILE_SYSTEM, "local"},
+                                                  {Options::ROW_TRACKING_ENABLED, "true"},
+                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
+    CreateTable(fields, /*partition_keys=*/{}, options);
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
-    auto schema = arrow::schema(fields_);
+    auto schema = arrow::schema(fields);
 
-    // write field: f0
-    std::vector write_cols0 = {"f0"};
+    // Initial full-row write assigns row ids 0 and 1.
     auto src_array0 = std::dynamic_pointer_cast(
-        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0]}), R"([
-        [1]
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+        [1, "a", "old_blob_0"],
+        [2, "b", "old_blob_1"]
     ])")
             .ValueOrDie());
-    ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, {}, write_cols0, {src_array0}));
-    ASSERT_OK(Commit(table_path, commit_msgs));
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
+                         WriteArray(table_path, {}, schema->field_names(), {src_array0}));
+    ASSERT_OK(Commit(table_path, commit_msgs0));
 
-    // write field: f1
-    std::vector write_cols1 = {"f1"};
+    // Update only b0 and align it with the existing row ids.
+    std::vector blob_write_cols = {"b0"};
     auto src_array1 = std::dynamic_pointer_cast(
-        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
-        ["a"]
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
+        ["new_blob_0"],
+        ["new_blob_1"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
+                         WriteArray(table_path, {}, blob_write_cols, {src_array1}));
+    SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
+    ASSERT_OK(Commit(table_path, commit_msgs1));
+
+    auto expected_array = std::dynamic_pointer_cast(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+        [1, "a", "new_blob_0"],
+        [2, "b", "new_blob_1"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+
+    if (GetParam() != "lance") {
+        auto expected_with_row_id = std::dynamic_pointer_cast(
+            arrow::ipc::internal::json::ArrayFromJSON(
+                arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().field_}),
+                R"([
+        [1, "a", "new_blob_0", 0],
+        [2, "b", "new_blob_1", 1]
+    ])")
+                .ValueOrDie());
+        ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));
+    }
+}
+
+TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyFirstCommitFailsWithoutFirstRowId) {
+    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
+                                 arrow::field("f1", arrow::utf8()), BlobUtils::ToArrowField("b0")};
+    std::map options = {{Options::MANIFEST_FORMAT, "orc"},
+                                                  {Options::FILE_FORMAT, GetParam()},
+                                                  {Options::FILE_SYSTEM, "local"},
+                                                  {Options::ROW_TRACKING_ENABLED, "true"},
+                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
+    CreateTable(fields, /*partition_keys=*/{}, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    auto blob_array = std::dynamic_pointer_cast(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields[2]}), R"([
+        ["blob_0"]
     ])")
             .ValueOrDie());
-    ASSERT_NOK_WITH_MSG(WriteArray(table_path, {}, write_cols1, {src_array1}),
-                        "SeparateBlobArray expects at least one main field, but got none.");
+    std::vector blob_write_cols = {"b0"};
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+                         WriteArray(table_path, {}, blob_write_cols, {blob_array}));
+
+    ASSERT_EQ(commit_msgs.size(), 1);
+    auto commit_msg = std::dynamic_pointer_cast(commit_msgs[0]);
+    ASSERT_TRUE(commit_msg);
+    ASSERT_EQ(commit_msg->data_increment_.new_files_.size(), 1);
+    const auto& blob_file = commit_msg->data_increment_.new_files_[0];
+    ASSERT_TRUE(BlobUtils::IsBlobFile(blob_file->file_name));
+    ASSERT_FALSE(blob_file->first_row_id.has_value());
+
+    ASSERT_NOK_WITH_MSG(Commit(table_path, commit_msgs), "blobStart 0 should be less than start 0");
 }
 
 TEST_P(BlobTableInteTest, TestMultipleAppendsDifferentFirstRowIds) {

From 8b64b52aec8b88be98885e7ddea0c6b90fc71471 Mon Sep 17 00:00:00 2001
From: lxy264173 
Date: Thu, 23 Jul 2026 13:34:08 +0800
Subject: [PATCH 2/3] fix review

---
 .../core/append/append_only_writer_test.cpp   | 19 ++++++--
 .../core/io/rolling_blob_file_writer.cpp      | 44 ++++++++-----------
 src/paimon/core/io/rolling_blob_file_writer.h |  3 +-
 .../core/io/rolling_blob_file_writer_test.cpp | 14 +++---
 4 files changed, 44 insertions(+), 36 deletions(-)

diff --git a/src/paimon/core/append/append_only_writer_test.cpp b/src/paimon/core/append/append_only_writer_test.cpp
index df7d912aa..00dcb61b3 100644
--- a/src/paimon/core/append/append_only_writer_test.cpp
+++ b/src/paimon/core/append/append_only_writer_test.cpp
@@ -763,6 +763,7 @@ TEST_F(AppendOnlyWriterTest, TestWriteWithOnlyBlobField) {
 
     ASSERT_OK(writer->Write(CreateStructBatch(schema, {blob_array})));
     ASSERT_OK_AND_ASSIGN(CommitIncrement inc, writer->PrepareCommit(/*wait_compaction=*/true));
+    ASSERT_OK(writer->Close());
 
     const auto& new_files = inc.GetNewFilesIncrement().NewFiles();
     ASSERT_EQ(new_files.size(), 1);
@@ -770,9 +771,21 @@ TEST_F(AppendOnlyWriterTest, TestWriteWithOnlyBlobField) {
     ASSERT_EQ(new_files[0]->row_count, 2);
     ASSERT_TRUE(new_files[0]->write_cols.has_value());
     ASSERT_EQ(new_files[0]->write_cols.value(), std::vector({"blob"}));
-    ASSERT_TRUE(
-        options.GetFileSystem()->Exists(path_factory->ToPath(new_files[0]->file_name)).value());
-    ASSERT_OK(writer->Close());
+    std::string blob_file_path = path_factory->ToPath(new_files[0]->file_name);
+    ASSERT_TRUE(options.GetFileSystem()->Exists(blob_file_path).value());
+
+    auto blob_reader = OpenFormatReader(blob_file_path, "blob");
+    ::ArrowSchema c_blob_schema;
+    ASSERT_TRUE(arrow::ExportSchema(*schema, &c_blob_schema).ok());
+    ASSERT_OK(blob_reader->SetReadSchema(&c_blob_schema, /*predicate=*/nullptr,
+                                         /*selection_bitmap=*/std::nullopt));
+    ASSERT_OK_AND_ASSIGN(auto actual_array, ReadResultCollector::CollectResult(blob_reader.get()));
+    auto expected_struct_array =
+        arrow::StructArray::Make({blob_array}, {blob_field->name()}).ValueOrDie();
+    auto expected_array = std::make_shared(expected_struct_array);
+    ASSERT_TRUE(expected_array->Equals(actual_array)) << "Expected:\n"
+                                                      << expected_array->ToString() << "\nActual:\n"
+                                                      << actual_array->ToString();
 }
 
 TEST_F(AppendOnlyWriterTest, TestWriteWithMultipleBlobFields) {
diff --git a/src/paimon/core/io/rolling_blob_file_writer.cpp b/src/paimon/core/io/rolling_blob_file_writer.cpp
index a8c9f4f8b..59ed5ec17 100644
--- a/src/paimon/core/io/rolling_blob_file_writer.cpp
+++ b/src/paimon/core/io/rolling_blob_file_writer.cpp
@@ -16,6 +16,7 @@
 
 #include "paimon/core/io/rolling_blob_file_writer.h"
 
+#include 
 #include 
 #include 
 #include 
@@ -27,7 +28,6 @@
 #include "arrow/c/bridge.h"
 #include "arrow/c/helpers.h"
 #include "fmt/format.h"
-#include "fmt/ranges.h"
 #include "paimon/common/data/blob_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/scope_guard.h"
@@ -97,9 +97,6 @@ Status RollingBlobFileWriter::Write(::ArrowArray* record) {
 }
 
 Status RollingBlobFileWriter::CloseCurrentWriter() {
-    if (current_writer_ == nullptr && blob_writer_ == nullptr) {
-        return Status::OK();
-    }
     if (blob_writer_ == nullptr) {
         return Status::OK();
     }
@@ -111,8 +108,7 @@ Status RollingBlobFileWriter::CloseCurrentWriter() {
                            CloseBlobWriter());
 
     if (main_data_file_meta != nullptr) {
-        PAIMON_RETURN_NOT_OK(
-            ValidateFileConsistency(main_data_file_meta, blob_metas, blob_schema_->num_fields()));
+        PAIMON_RETURN_NOT_OK(ValidateFileConsistency(main_data_file_meta, blob_metas));
         results_.push_back(main_data_file_meta);
     }
     results_.insert(results_.end(), blob_metas.begin(), blob_metas.end());
@@ -146,29 +142,25 @@ Result>> RollingBlobFileWriter::CloseB
 
 Status RollingBlobFileWriter::ValidateFileConsistency(
     const std::shared_ptr& main_data_file_meta,
-    const std::vector>& blob_tagged_metas, int32_t blob_field_count) {
-    if (blob_tagged_metas.empty()) {
-        return Status::OK();
-    }
-    // With multiple blob fields, each blob field produces its own set of files.
-    // total_blob_row_count should be exactly main_row_count * blob_field_count.
-    int64_t main_row_count = main_data_file_meta->row_count;
-    int64_t expected_blob_row_count = main_row_count * blob_field_count;
-    int64_t total_blob_row_count = 0;
+    const std::vector>& blob_tagged_metas) {
+    std::map blob_field_row_counts;
     for (const auto& blob_tagged_meta : blob_tagged_metas) {
-        total_blob_row_count += blob_tagged_meta->row_count;
+        if (!blob_tagged_meta->write_cols || blob_tagged_meta->write_cols->empty()) {
+            return Status::Invalid(
+                fmt::format("This is a bug: Blob file {} must contain a write column.",
+                            blob_tagged_meta->file_name));
+        }
+        blob_field_row_counts[blob_tagged_meta->write_cols->at(0)] += blob_tagged_meta->row_count;
     }
-    if (total_blob_row_count != expected_blob_row_count) {
-        std::vector blob_file_names;
-        for (const auto& blob_tagged_meta : blob_tagged_metas) {
-            blob_file_names.push_back(blob_tagged_meta->file_name);
+
+    int64_t main_row_count = main_data_file_meta->row_count;
+    for (const auto& [field_name, row_count] : blob_field_row_counts) {
+        if (row_count != main_row_count) {
+            return Status::Invalid(fmt::format(
+                "This is a bug: The row count of main file and blob file does not match. Main "
+                "file: {} (row count: {}), blob field name: {} (row count: {})",
+                main_data_file_meta->file_name, main_row_count, field_name, row_count));
         }
-        return Status::Invalid(fmt::format(
-            "This is a bug: The row count of main file and blob files does not match. "
-            "Main file: {} (row count: {}), blob field count: {}, "
-            "expected blob row count: {}, blob files: {} (actual total row count: {})",
-            main_data_file_meta->file_name, main_row_count, blob_field_count,
-            expected_blob_row_count, fmt::join(blob_file_names, ", "), total_blob_row_count));
     }
     return Status::OK();
 }
diff --git a/src/paimon/core/io/rolling_blob_file_writer.h b/src/paimon/core/io/rolling_blob_file_writer.h
index 0aa998f3b..346824508 100644
--- a/src/paimon/core/io/rolling_blob_file_writer.h
+++ b/src/paimon/core/io/rolling_blob_file_writer.h
@@ -77,8 +77,7 @@ class RollingBlobFileWriter
  private:
     static Status ValidateFileConsistency(
         const std::shared_ptr& main_data_file_meta,
-        const std::vector>& blob_tagged_metas,
-        int32_t blob_field_count);
+        const std::vector>& blob_tagged_metas);
 
     Status CloseCurrentWriter();
 
diff --git a/src/paimon/core/io/rolling_blob_file_writer_test.cpp b/src/paimon/core/io/rolling_blob_file_writer_test.cpp
index 1d72563e3..4ce38fb58 100644
--- a/src/paimon/core/io/rolling_blob_file_writer_test.cpp
+++ b/src/paimon/core/io/rolling_blob_file_writer_test.cpp
@@ -82,11 +82,15 @@ TEST_F(RollingBlobFileWriterTest, ValidateFileConsistency) {
         /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
         /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt, /*first_row_id=*/3,
         /*write_cols=*/std::vector({"blob"}));
-    ASSERT_OK(RollingBlobFileWriter::ValidateFileConsistency(file_meta1, {file_meta2, file_meta3},
-                                                             /*blob_field_count=*/1));
-    ASSERT_NOK_WITH_MSG(RollingBlobFileWriter::ValidateFileConsistency(file_meta1, {file_meta2},
-                                                                       /*blob_field_count=*/2),
-                        "This is a bug: The row count of main file and blob files does not match.");
+    ASSERT_OK(RollingBlobFileWriter::ValidateFileConsistency(file_meta1, {file_meta2, file_meta3}));
+    ASSERT_NOK_WITH_MSG(RollingBlobFileWriter::ValidateFileConsistency(file_meta1, {file_meta2}),
+                        "This is a bug: The row count of main file and blob file does not match.");
+
+    file_meta2->write_cols = std::vector({"blob1"});
+    file_meta3->write_cols = std::vector({"blob2"});
+    ASSERT_NOK_WITH_MSG(
+        RollingBlobFileWriter::ValidateFileConsistency(file_meta1, {file_meta2, file_meta3}),
+        "This is a bug: The row count of main file and blob file does not match.");
 }
 
 }  // namespace paimon::test

From 402c473b5f86526c09489427e37b3af65054f01c Mon Sep 17 00:00:00 2001
From: lxy264173 
Date: Fri, 24 Jul 2026 11:18:21 +0800
Subject: [PATCH 3/3] rm lance

---
 test/inte/blob_table_inte_test.cpp | 14 ++++++--------
 1 file changed, 6 insertions(+), 8 deletions(-)

diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp
index a73c73df6..79fb5ac45 100644
--- a/test/inte/blob_table_inte_test.cpp
+++ b/test/inte/blob_table_inte_test.cpp
@@ -1041,17 +1041,15 @@ TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyWriteWithFirstRowId) {
             .ValueOrDie());
     ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
 
-    if (GetParam() != "lance") {
-        auto expected_with_row_id = std::dynamic_pointer_cast(
-            arrow::ipc::internal::json::ArrayFromJSON(
-                arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().field_}),
-                R"([
+    auto expected_with_row_id = std::dynamic_pointer_cast(
+        arrow::ipc::internal::json::ArrayFromJSON(
+            arrow::struct_({fields[0], fields[1], fields[2], SpecialFields::RowId().field_}),
+            R"([
         [1, "a", "new_blob_0", 0],
         [2, "b", "new_blob_1", 1]
     ])")
-                .ValueOrDie());
-        ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));
-    }
+            .ValueOrDie());
+    ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "b0", "_ROW_ID"}, expected_with_row_id));
 }
 
 TEST_P(BlobTableInteTest, TestDataEvolutionBlobOnlyFirstCommitFailsWithoutFirstRowId) {