feat(reader): support late materialization with probe/payload two-phase reads

include/paimon/read_context.h

@@ -51,7 +51,7 @@ class PAIMON_EXPORT ReadContext {
                 const std::vector<std::string>& read_field_names,
                 const std::vector<int32_t>& read_field_ids,
                 const std::shared_ptr<Predicate>& predicate, bool enable_predicate_filter,
-                bool enable_prefetch, uint32_t prefetch_batch_count,
+                bool enable_prefetch, bool enable_late_materializing, uint32_t prefetch_batch_count,
                 uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
                 uint32_t row_to_batch_thread_number, const std::optional<std::string>& table_schema,
                 const std::shared_ptr<MemoryPool>& memory_pool,
@@ -97,6 +97,9 @@ class PAIMON_EXPORT ReadContext {
     bool EnablePrefetch() const {
         return enable_prefetch_;
     }
+    bool EnableLateMaterializing() const {
+        return enable_late_materializing_;
+    }
     uint32_t GetPrefetchBatchCount() const {
         return prefetch_batch_count_;
     }
@@ -163,6 +166,7 @@ class PAIMON_EXPORT ReadContext {
     std::shared_ptr<Predicate> predicate_;
     bool enable_predicate_filter_;
     bool enable_prefetch_;
+    bool enable_late_materializing_;
     uint32_t prefetch_batch_count_;
     uint32_t prefetch_max_parallel_num_;
     bool enable_multi_thread_row_to_batch_;
@@ -306,6 +310,15 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& EnablePrefetch(bool enabled);
 
+    /// Enable or disable late materialization (probe/payload two-phase reads). When enabled,
+    /// each parallel reader under the prefetch layer performs a probe read of predicate
+    /// columns first and only materializes payload columns for matched rows.
+    /// @param enabled Whether to enable late materialization (default: false)
+    /// @return Reference to this builder for method chaining.
+    /// @note Without a pushed-down predicate the late-materializing reader degrades to a
+    /// plain passthrough.
+    ReadContextBuilder& EnableLateMaterializing(bool enabled);
+
     /// Enable or disable the read-ahead cache for read operations.
     ///
     /// A read-ahead cache is used to prebuffer data ranges before they are needed,

include/paimon/reader/prefetch_file_batch_reader.h

@@ -40,7 +40,7 @@ class PAIMON_EXPORT PrefetchFileBatchReader : public FileBatchReader {
     /// Retrieves the row number of the next row to be read.
     /// This method indicates the current read position within the file.
     /// @return The row number of the next row to read.
-    virtual uint64_t GetNextRowToRead() const = 0;
+    virtual Result<uint64_t> GetNextRowToRead() const = 0;
 
     /// Generates a list of row ranges to be read in batches.
     /// Each range specifies the start and end row numbers for a batch,

src/paimon/CMakeLists.txt

@@ -134,6 +134,7 @@ set(PAIMON_COMMON_SRCS
     common/predicate/starts_with.cpp
     common/reader/batch_reader.cpp
     common/reader/concat_batch_reader.cpp
+    common/reader/late_materializing_file_batch_reader.cpp
     common/reader/predicate_batch_reader.cpp
     common/reader/prefetch_file_batch_reader_impl.cpp
     common/reader/reader_utils.cpp
@@ -606,6 +607,7 @@ if(PAIMON_BUILD_TESTS)
                     common/predicate/predicate_utils_test.cpp
                     common/predicate/predicate_validator_test.cpp
                     common/reader/concat_batch_reader_test.cpp
+                    common/reader/late_materializing_file_batch_reader_test.cpp
                     common/reader/predicate_batch_reader_test.cpp
                     common/reader/prefetch_file_batch_reader_impl_test.cpp
                     common/reader/reader_utils_test.cpp