fix: close input/output streams when remote lookup file read/write failed (alibaba#380)

Author: lxy-9602

cmake_modules/CorrosionFetch.cmake

@@ -16,7 +16,7 @@
 # targets. Used to bring in third_party/tantivy_ffi for the tantivy-fulltext
 # global index (see docs/dev/tantivy_fts_migration_plan.md).
 #
-# Pinned to v0.5.0 (stable release). Requires CMake >= 3.22.
+# Pinned to v0.5.2 (stable release). Requires CMake >= 3.22.
 
 include(FetchContent)
 

src/paimon/core/mergetree/compact/lookup_merge_tree_compact_rewriter.h

@@ -15,6 +15,7 @@
  */
 
 #pragma once
+
 #include "arrow/api.h"
 #include "paimon/core/core_options.h"
 #include "paimon/core/io/data_file_meta.h"
@@ -24,7 +25,9 @@
 #include "paimon/core/mergetree/lookup/remote_lookup_file_manager.h"
 #include "paimon/core/mergetree/lookup_levels.h"
 #include "paimon/core/schema/table_schema.h"
+
 namespace paimon {
+
 /// A `MergeTreeCompactRewriter` which produces changelog files by lookup for the compaction
 /// involving level 0 files.
 template <typename T>
@@ -86,7 +89,6 @@ class LookupMergeTreeCompactRewriter : public ChangelogMergeTreeRewriter {
     Result<std::vector<std::shared_ptr<DataFileMeta>>> NotifyRewriteCompactAfter(
         const std::vector<std::shared_ptr<DataFileMeta>>& files) override;
 
- private:
     std::unique_ptr<LookupLevels<T>> lookup_levels_;
     std::shared_ptr<BucketedDvMaintainer> dv_maintainer_;
     std::shared_ptr<RemoteLookupFileManager> remote_lookup_file_manager_;

src/paimon/core/mergetree/compact/lookup_merge_tree_compact_rewriter_test.cpp

@@ -53,7 +53,9 @@
 #include "paimon/testing/utils/io_exception_helper.h"
 #include "paimon/testing/utils/read_result_collector.h"
 #include "paimon/testing/utils/testharness.h"
+
 namespace paimon::test {
+
 class LookupMergeTreeCompactRewriterTest : public ::testing::TestWithParam<std::string> {
  public:
     void SetUp() override {

src/paimon/core/mergetree/lookup/remote_lookup_file_manager.cpp

@@ -17,8 +17,10 @@
 
 #include "fmt/format.h"
 #include "paimon/common/utils/path_util.h"
+#include "paimon/common/utils/scope_guard.h"
 #include "paimon/core/mergetree/lookup/file_position.h"
 #include "paimon/core/mergetree/lookup/positioned_key_value.h"
+
 namespace paimon {
 
 RemoteLookupFileManager::RemoteLookupFileManager(
@@ -103,6 +105,15 @@ Status RemoteLookupFileManager::CopyRemoteToLocal(const std::string& remote_path
 Status RemoteLookupFileManager::CopyFromInputToOutput(
     std::unique_ptr<InputStream>&& input_stream,
     std::unique_ptr<OutputStream>&& output_stream) const {
+    ScopeGuard input_close_guard([&input_stream]() -> void {
+        Status s = input_stream->Close();
+        (void)s;
+    });
+    ScopeGuard output_close_guard([&output_stream]() -> void {
+        Status s = output_stream->Close();
+        (void)s;
+    });
+
     auto buffer = std::make_shared<Bytes>(kBufferSize, pool_.get());
     PAIMON_ASSIGN_OR_RAISE(int64_t total_length, input_stream->Length());
     PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(total_length, "input stream length"));
@@ -128,8 +139,12 @@ Status RemoteLookupFileManager::CopyFromInputToOutput(
         write_size += current_read_size;
     }
     PAIMON_RETURN_NOT_OK(output_stream->Flush());
-    PAIMON_RETURN_NOT_OK(output_stream->Close());
-    PAIMON_RETURN_NOT_OK(input_stream->Close());
+    Status output_close_status = output_stream->Close();
+    output_close_guard.Release();
+    PAIMON_RETURN_NOT_OK(output_close_status);
+    Status input_close_status = input_stream->Close();
+    input_close_guard.Release();
+    PAIMON_RETURN_NOT_OK(input_close_status);
     return Status::OK();
 }
 template Result<std::shared_ptr<DataFileMeta>>

src/paimon/core/mergetree/lookup/remote_lookup_file_manager.h

@@ -50,7 +50,6 @@ class RemoteLookupFileManager {
 
     static constexpr uint64_t kBufferSize = 1024 * 1024;
 
- private:
     int32_t level_threshold_;
     std::shared_ptr<MemoryPool> pool_;
     std::shared_ptr<DataFilePathFactory> path_factory_;

src/paimon/global_index/lucene/lucene_global_index_reader.cpp

@@ -208,7 +208,10 @@ std::shared_ptr<GlobalIndexResult> LuceneGlobalIndexReader::SearchWithNoLimit(
 
 Result<std::shared_ptr<GlobalIndexResult>> LuceneGlobalIndexReader::VisitFullTextSearch(
     const std::shared_ptr<FullTextSearch>& full_text_search) {
-    if (full_text_search && full_text_search->min_score.has_value()) {
+    if (!full_text_search) {
+        return Status::Invalid("VisitFullTextSearch: null FullTextSearch pointer");
+    }
+    if (full_text_search->min_score.has_value()) {
         // The lucene backend does not support min_score pushdown. Fail loudly
         // instead of silently ignoring the threshold and returning unfiltered
         // results, which would be a correctness bug for the caller.

src/paimon/global_index/tantivy/tantivy_streaming_test.cpp

@@ -93,7 +93,7 @@ struct WriteResult {
 class StreamingTestFixture : public ::testing::Test {
  public:
     WriteResult BuildArchive(std::size_t n_docs,
-                             const std::string& text_template = "apple banana cherry {}") {
+                             const std::string& text_template = "apple banana cherry %zu") {
         auto root_dir = paimon::test::UniqueTestDirectory::Create();
         EXPECT_TRUE(root_dir);
         std::string root = root_dir->Str();
@@ -239,7 +239,7 @@ TEST(ParseArchiveHeaderFuzz, PayloadLenNegative) {
 
 TEST_F(StreamingTestFixture, ConcurrentQueryOnSameReader) {
     // 50 docs containing "apple" in every one (all should match)
-    auto wr = BuildArchive(50, "apple banana {}");
+    auto wr = BuildArchive(50, "apple banana %zu");
     auto reader = OpenReader(wr.root_dir, wr.meta);
 
     auto fts = BuildMatchAll("apple");

src/paimon/global_index/tantivy/tantivy_writer_test.cpp

@@ -134,13 +134,6 @@ std::vector<PackedEntry> ParsePacked(const std::vector<uint8_t>& bytes) {
 
 class TantivyGlobalIndexWriterTest : public ::testing::Test {
  public:
-    std::unique_ptr<::ArrowSchema> CreateArrowSchema(
-        const std::shared_ptr<arrow::DataType>& data_type) const {
-        auto c_schema = std::make_unique<::ArrowSchema>();
-        EXPECT_TRUE(arrow::ExportType(*data_type, c_schema.get()).ok());
-        return c_schema;
-    }
-
     Result<std::vector<GlobalIndexIOMeta>> WriteIndex(
         const std::string& root, const std::shared_ptr<arrow::DataType>& data_type,
         const std::map<std::string, std::string>& options,

test/inte/append_compaction_inte_test.cpp

@@ -117,6 +117,7 @@ class AppendCompactionInteTest : public testing::Test,
             auto partition = BinaryRowGenerator::GenerateRow({10}, pool_.get());
             int32_t bucket = 1;
             auto abstract_write = dynamic_cast<AbstractFileStoreWrite*>(helper->write_.get());
+            ASSERT_NE(abstract_write, nullptr);
             ASSERT_OK_AND_ASSIGN(auto restore_files,
                                  abstract_write->ScanExistingFileMetas(partition, bucket));
             ASSERT_OK_AND_ASSIGN(

test/inte/write_and_read_inte_test.cpp

@@ -47,11 +47,6 @@
 #include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
 
-namespace paimon {
-class DataSplit;
-class RecordBatch;
-}  // namespace paimon
-
 namespace paimon::test {
 // This is a sdk end-to-end test demo that supports write, commit, scan, and read operations.
 class WriteAndReadInteTest