fix: release global index writer before reader to avoid mem issue when write index failed (alibaba#398)

Author: lxy-9602

src/paimon/core/global_index/global_index_write_task.cpp

@@ -20,6 +20,7 @@
 #include "paimon/common/table/special_fields.h"
 #include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/scope_guard.h"
 #include "paimon/core/core_options.h"
 #include "paimon/core/global_index/global_index_file_manager.h"
 #include "paimon/core/io/data_increment.h"
@@ -204,16 +205,21 @@ Result<std::shared_ptr<CommitMessage>> GlobalIndexWriteTask::WriteIndex(
         std::shared_ptr<GlobalIndexFileManager> index_file_manager,
         CreateGlobalIndexFileManager(table_path, table_schema, core_options, pool));
 
+    // create batch reader
+    PAIMON_ASSIGN_OR_RAISE(
+        std::unique_ptr<BatchReader> batch_reader,
+        CreateBatchReader(table_path, field_name, indexed_split, core_options, pool));
+
     // create global index writer
     PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema->GetField(field_name));
     PAIMON_ASSIGN_OR_RAISE(
         std::shared_ptr<GlobalIndexWriter> global_index_writer,
         CreateGlobalIndexWriter(index_type, field, index_file_manager, core_options, pool));
 
-    // create batch reader
-    PAIMON_ASSIGN_OR_RAISE(
-        std::unique_ptr<BatchReader> batch_reader,
-        CreateBatchReader(table_path, field_name, indexed_split, core_options, pool));
+    ScopeGuard guard([&]() {
+        global_index_writer.reset();
+        batch_reader.reset();
+    });
 
     // read from data split and write to index writer
     PAIMON_ASSIGN_OR_RAISE(

test/inte/global_index_test.cpp

@@ -292,6 +292,41 @@ TEST_P(GlobalIndexTest, TestWriteLuminaIndex) {
     ASSERT_TRUE(expected_commit_message->TEST_Equal(*index_commit_msg_impl));
 }
 
+TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithMismatchedDimension) {
+    arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+                                 arrow::field("f1", arrow::list(arrow::float32()))};
+    auto schema = arrow::schema(fields);
+    std::map<std::string, std::string> lumina_options = {{"lumina.index.dimension", "3"},
+                                                         {"lumina.index.type", "bruteforce"},
+                                                         {"lumina.distance.metric", "l2"},
+                                                         {"lumina.encoding.type", "rawf32"},
+                                                         {"lumina.search.parallel_number", "10"}};
+
+    std::map<std::string, std::string> options = {
+        {Options::MANIFEST_FORMAT, "orc"},         {Options::FILE_FORMAT, file_format_},
+        {Options::FILE_SYSTEM, "local"},           {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"}, {Options::READ_BATCH_SIZE, "1"}};
+
+    CreateTable(/*partition_keys=*/{}, schema, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    std::vector<std::string> write_cols = schema->field_names();
+    auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+        ["a", [0.0, 0.0, 0.0]],
+        ["b", [0.0, 0.0, 0.0, 0.0]]
+    ])")
+                         .ValueOrDie();
+
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, write_cols, src_array));
+    ASSERT_OK(Commit(table_path, commit_msgs));
+
+    ASSERT_NOK_WITH_MSG(
+        WriteIndex(table_path, /*partition_filters=*/{}, "f1", "lumina",
+                   /*options=*/lumina_options, Range(0, 1)),
+        "invalid input array in LuminaIndexWriter, length of field array [1] multiplied "
+        "dimension [3] must match length of field value array [4]");
+}
+
 TEST_P(GlobalIndexTest, TestWriteIndex) {
     CreateTable();
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");