fix code review

Author: lszskye

src/paimon/core/core_options.cpp

@@ -248,6 +248,37 @@ class ConfigParser {
         return Status::OK();
     }
 
+    // parse file.format.per.level
+    Status ParseFileFormatPerLevel(
+        std::map<int32_t, std::shared_ptr<FileFormat>>* file_format_per_level_ptr) const {
+        auto& file_format_per_level = *file_format_per_level_ptr;
+        std::string file_format_per_level_str;
+        PAIMON_RETURN_NOT_OK(
+            ParseString(Options::FILE_FORMAT_PER_LEVEL, &file_format_per_level_str));
+        if (!file_format_per_level_str.empty()) {
+            auto level2format =
+                StringUtils::Split(file_format_per_level_str, std::string(","), std::string(":"));
+            for (const auto& single_level : level2format) {
+                if (single_level.size() != 2) {
+                    return Status::Invalid(fmt::format(
+                        "fail to parse key {}, value {} (usage example: 0:avro,3:parquet)",
+                        Options::FILE_FORMAT_PER_LEVEL, file_format_per_level_str));
+                }
+                auto level = StringUtils::StringToValue<int32_t>(single_level[0]);
+                if (!level || level.value() < 0) {
+                    return Status::Invalid(
+                        fmt::format("fail to parse level {} from string to int in {}",
+                                    single_level[0], Options::FILE_FORMAT_PER_LEVEL));
+                }
+                std::shared_ptr<FileFormat> file_format;
+                PAIMON_RETURN_NOT_OK(ParseObject<FileFormatFactory>(
+                    "_no_use", /*default_identifier=*/single_level[1], &file_format));
+                file_format_per_level[level.value()] = file_format;
+            }
+        }
+        return Status::OK();
+    }
+
     bool ContainsKey(const std::string& key) const {
         return config_map_.find(key) != config_map_.end();
     }
@@ -599,31 +630,7 @@ Result<CoreOptions> CoreOptions::FromMap(
     PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::CACHE_PAGE_SIZE, &impl->cache_page_size));
 
     // parse file.format.per.level
-    std::string file_format_per_level_str;
-    PAIMON_RETURN_NOT_OK(
-        parser.ParseString(Options::FILE_FORMAT_PER_LEVEL, &file_format_per_level_str));
-    if (!file_format_per_level_str.empty()) {
-        auto level2format =
-            StringUtils::Split(file_format_per_level_str, std::string(","), std::string(":"));
-        for (const auto& single_level : level2format) {
-            if (single_level.size() != 2) {
-                return Status::Invalid(
-                    fmt::format("fail to parse key {}, value {} (usage example: 0:avro,3:parquet)",
-                                Options::FILE_FORMAT_PER_LEVEL, file_format_per_level_str));
-            }
-            auto level = StringUtils::StringToValue<int32_t>(single_level[0]);
-            if (!level || level.value() < 0) {
-                return Status::Invalid(
-                    fmt::format("fail to parse level {} from string to int in {}", single_level[0],
-                                Options::FILE_FORMAT_PER_LEVEL));
-            }
-            std::shared_ptr<FileFormat> file_format;
-            PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
-                "_no_use", /*default_identifier=*/single_level[1], &file_format));
-            impl->file_format_per_level[level.value()] = file_format;
-        }
-    }
-
+    PAIMON_RETURN_NOT_OK(parser.ParseFileFormatPerLevel(&impl->file_format_per_level));
     return options;
 }
 

src/paimon/core/core_options.h

@@ -153,7 +153,6 @@ class PAIMON_EXPORT CoreOptions {
 
  private:
     struct Impl;
-
     std::unique_ptr<Impl> impl_;
 };
 

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

@@ -16,6 +16,8 @@
 
 #pragma once
 
+#include <cassert>
+#include <functional>
 #include <memory>
 #include <optional>
 #include <utility>
@@ -49,7 +51,11 @@ class FirstRowMergeFunctionWrapper : public MergeFunctionWrapper<KeyValue> {
             Reset();
             return result;
         }
-        assert(result);
+        if (!result) {
+            Reset();
+            return Status::Invalid(
+                "In FirstRowMergeFunctionWrapper when call GetResult, there must have a result");
+        }
         PAIMON_ASSIGN_OR_RAISE(bool contains, contains_(result.value().key));
         if (contains) {
             // empty

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

@@ -16,8 +16,11 @@
 
 #pragma once
 
+#include <algorithm>
+#include <functional>
 #include <memory>
 #include <optional>
+#include <string>
 #include <utility>
 
 #include "paimon/core/deletionvectors/bucketed_dv_maintainer.h"

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

@@ -1134,7 +1134,6 @@ TEST_F(LookupMergeTreeCompactRewriterTest, TestRewriteLookupChangelogWithOutputL
     ASSERT_OK_AND_ASSIGN(auto runs, GenerateSortedRuns({file0, file1}));
 
     // When output_level is 0, RewriteLookupChangelog should return false
-    // This tests the condition at line 59 in changelog_merge_tree_rewriter.cpp
     ASSERT_OK_AND_ASSIGN(auto compact_result, rewriter->Rewrite(
                                                   /*output_level=*/0, /*drop_delete=*/true, runs));
     ASSERT_EQ(2, compact_result.Before().size());

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

@@ -92,14 +92,14 @@ class MergeTreeCompactRewriter : public CompactRewriter {
                              KeyValueRollingFileWriter* rolling_writer,
                              std::vector<std::shared_ptr<KeyValueMergeReader>>* reader_holders_ptr);
 
- private:
-    Result<std::shared_ptr<DataFilePathFactory>> CreateDataFilePathFactory(
-        const std::string& format);
-
  protected:
     CoreOptions options_;
     std::unique_ptr<MergeFileSplitRead> merge_file_split_read_;
 
+ private:
+    Result<std::shared_ptr<DataFilePathFactory>> CreateDataFilePathFactory(
+        const std::string& format);
+
  private:
     std::shared_ptr<MemoryPool> pool_;
     BinaryRow partition_;

src/paimon/core/mergetree/lookup_levels.h

@@ -57,7 +57,7 @@ class LookupLevels {
                                     const SortedRun& level);
 
     Status Close() {
-        // TODDO(xinyu.lxy): invalid cache
+        // TODO(xinyu.lxy): invalid cache
         lookup_file_cache_.clear();
         return Status::OK();
     }

src/paimon/core/utils/file_store_path_factory_cache.h

@@ -16,12 +16,15 @@
 
 #pragma once
 
+#include <map>
+#include <string>
+#include <vector>
+
 #include "paimon/core/core_options.h"
 #include "paimon/core/schema/table_schema.h"
 #include "paimon/core/utils/file_store_path_factory.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/result.h"
-
 namespace paimon {
 class FileStorePathFactoryCache {
  public: