fix review

Author: lxy-9602

src/paimon/core/append/append_only_writer_test.cpp

@@ -763,16 +763,29 @@ 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);
     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<std::string>({"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<arrow::ChunkedArray>(expected_struct_array);
+    ASSERT_TRUE(expected_array->Equals(actual_array)) << "Expected:\n"
+                                                      << expected_array->ToString() << "\nActual:\n"
+                                                      << actual_array->ToString();
 }
 
 TEST_F(AppendOnlyWriterTest, TestWriteWithMultipleBlobFields) {

src/paimon/core/io/rolling_blob_file_writer.cpp

@@ -16,6 +16,7 @@
 
 #include "paimon/core/io/rolling_blob_file_writer.h"
 
+#include <map>
 #include <memory>
 #include <string>
 #include <utility>
@@ -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<std::vector<std::shared_ptr<DataFileMeta>>> RollingBlobFileWriter::CloseB
 
 Status RollingBlobFileWriter::ValidateFileConsistency(
     const std::shared_ptr<DataFileMeta>& main_data_file_meta,
-    const std::vector<std::shared_ptr<DataFileMeta>>& 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<std::shared_ptr<DataFileMeta>>& blob_tagged_metas) {
+    std::map<std::string, int64_t> 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<std::string> 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();
 }

src/paimon/core/io/rolling_blob_file_writer.h

@@ -77,8 +77,7 @@ class RollingBlobFileWriter
  private:
     static Status ValidateFileConsistency(
         const std::shared_ptr<DataFileMeta>& main_data_file_meta,
-        const std::vector<std::shared_ptr<DataFileMeta>>& blob_tagged_metas,
-        int32_t blob_field_count);
+        const std::vector<std::shared_ptr<DataFileMeta>>& blob_tagged_metas);
 
     Status CloseCurrentWriter();
 

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<std::string>({"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<std::string>({"blob1"});
+    file_meta3->write_cols = std::vector<std::string>({"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