refactor(manifest): address lazy decode review feedback

Author: zhongzheyun

src/paimon/core/manifest/manifest_entry_serializer.cpp

@@ -44,6 +44,10 @@ Status ManifestEntrySerializer::ValidateVersion(int32_t version) {
     return Status::Invalid(fmt::format("Unsupported version: {}", version));
 }
 
+int32_t ManifestEntrySerializer::GetBucket(const InternalRow& row) {
+    return row.GetInt(3);
+}
+
 Result<ManifestEntry> ManifestEntrySerializer::ConvertFrom(int32_t version,
                                                            const InternalRow& row) const {
     PAIMON_RETURN_NOT_OK(ValidateVersion(version));

src/paimon/core/manifest/manifest_entry_serializer.h

@@ -53,6 +53,9 @@ class ManifestEntrySerializer : public VersionedObjectSerializer<ManifestEntry>
     /// Validate the serialization version before reading fields that may vary by version.
     static Status ValidateVersion(int32_t version);
 
+    /// Get the bucket from a versioned manifest entry row without fully deserializing it.
+    static int32_t GetBucket(const InternalRow& row);
+
     Result<ManifestEntry> ConvertFrom(int32_t version, const InternalRow& row) const override;
 
     Result<BinaryRow> ToRow(const ManifestEntry& record) const override;

src/paimon/core/manifest/manifest_entry_serializer_test.cpp

@@ -55,6 +55,7 @@ TEST_F(ManifestEntrySerializerTest, TestToFromRow) {
     ManifestEntrySerializer serializer(pool);
     for (const auto& entry : entries) {
         ASSERT_OK_AND_ASSIGN(auto row, serializer.ToRow(entry));
+        ASSERT_EQ(entry.Bucket(), ManifestEntrySerializer::GetBucket(row));
         ASSERT_OK_AND_ASSIGN(auto result_entry, serializer.FromRow(row));
         ASSERT_EQ(entry, result_entry);
         ASSERT_EQ(entry.ToString(), result_entry.ToString());

src/paimon/core/manifest/manifest_file.cpp

@@ -95,7 +95,7 @@ Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t buc
             for (int64_t i = 0; i < batch->length(); i++) {
                 ColumnarRow row(batch->fields(), pool_, i);
                 PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(row.GetInt(0)));
-                if (row.GetInt(3) != bucket) {
+                if (ManifestEntrySerializer::GetBucket(row) != bucket) {
                     continue;
                 }
                 PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry, serializer_->FromRow(row));

src/paimon/core/utils/objects_file.h

@@ -186,7 +186,6 @@ Status ObjectsFile<T>::ReadArrowBatches(
                            reader_builder_->Build(file_input_stream));
     auto reader = std::make_unique<ManifestMetaReader>(std::move(batch_reader),
                                                        serializer_->GetDataType(), pool_);
-    ScopeGuard close_reader([&reader]() { reader->Close(); });
     while (true) {
         PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch arrow_array, reader->NextBatch());
         auto& c_array = arrow_array.first;