perf(read): add bounded late materialization

docs/source/user_guide/read.rst

@@ -40,6 +40,24 @@ the two layers can evolve their functionality and optimize code relatively indep
 cross-language task scheduling and interaction (e.g., Java and C++), substantially reducing engineering maintenance costs
 across the two language ecosystems.
 
+Late Materialization
+--------------------
+
+In the raw-file read path, late materialization can reduce payload-column decoding when a
+predicate selects only a small number of rows. The reader first loads the columns required by the
+predicate (the *probe* columns), evaluates the predicate, and then reads the remaining projected
+columns (the *payload* columns) only for matching row ranges.
+
+The optimization is disabled by default. Enable it with
+``read.late-materialization.enabled=true``. It is applied only when predicate filtering is enabled,
+and the probe and payload projections are both non-empty. The selected file-local row IDs are
+pushed directly to each file reader, so row tracking and global row IDs are not required.
+Otherwise, the reader uses the normal single-pass path.
+
+``read.late-materialization.max-match-rows`` limits the number of matching rows accumulated for a
+split and defaults to ``1024``. If the limit is exceeded, the reader falls back to the normal path.
+The probe work already completed before the fallback is retained in the reader metrics.
+
 
 Schema Evolution
 -----------------------

include/paimon/defs.h

@@ -202,6 +202,15 @@ struct PAIMON_EXPORT Options {
     /// The default value is 1024.
     static const char READ_BATCH_SIZE[];
 
+    /// "read.late-materialization.enabled" - Whether to read predicate columns first and then
+    /// fetch payload columns only for matching rows. Default value is false.
+    static const char READ_LATE_MATERIALIZATION_ENABLED[];
+
+    /// "read.late-materialization.max-match-rows" - Maximum number of matching rows per split
+    /// allowed for late materialization. If the predicate matches more rows, the reader falls back
+    /// to the normal read path. Default value is 1024.
+    static const char READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS[];
+
     /// "write.batch-size" - Write batch size for any file format if it supports.
     /// The default value is 1024.
     static const char WRITE_BATCH_SIZE[];

src/paimon/CMakeLists.txt

@@ -133,6 +133,7 @@ set(PAIMON_COMMON_SRCS
     common/predicate/starts_with.cpp
     common/reader/batch_reader.cpp
     common/reader/concat_batch_reader.cpp
+    common/reader/late_materialization_batch_reader.cpp
     common/reader/predicate_batch_reader.cpp
     common/reader/prefetch_file_batch_reader_impl.cpp
     common/reader/reader_utils.cpp
@@ -584,6 +585,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_materialization_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

src/paimon/common/defs.cpp

@@ -60,6 +60,9 @@ const char Options::SCAN_MODE[] = "scan.mode";
 const char Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS[] =
     "scan.manifest-entry-cache.max-snapshots";
 const char Options::READ_BATCH_SIZE[] = "read.batch-size";
+const char Options::READ_LATE_MATERIALIZATION_ENABLED[] = "read.late-materialization.enabled";
+const char Options::READ_LATE_MATERIALIZATION_MAX_MATCH_ROWS[] =
+    "read.late-materialization.max-match-rows";
 const char Options::WRITE_BATCH_SIZE[] = "write.batch-size";
 const char Options::WRITE_BUFFER_SIZE[] = "write-buffer-size";
 const char Options::WRITE_BUFFER_SPILLABLE[] = "write-buffer-spillable";

src/paimon/common/reader/concat_batch_reader.cpp

@@ -30,8 +30,12 @@ namespace paimon {
 class MemoryPool;
 
 ConcatBatchReader::ConcatBatchReader(std::vector<std::unique_ptr<BatchReader>>&& readers,
-                                     const std::shared_ptr<MemoryPool>& pool)
-    : arrow_pool_(GetArrowPool(pool)), readers_(std::move(readers)), current_(0) {}
+                                     const std::shared_ptr<MemoryPool>& pool,
+                                     const std::shared_ptr<Metrics>& completed_metrics)
+    : arrow_pool_(GetArrowPool(pool)),
+      readers_(std::move(readers)),
+      completed_metrics_(completed_metrics),
+      current_(0) {}
 
 Result<BatchReader::ReadBatch> ConcatBatchReader::NextBatch() {
     PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap,
@@ -46,7 +50,9 @@ void ConcatBatchReader::Close() {
 }
 
 std::shared_ptr<Metrics> ConcatBatchReader::GetReaderMetrics() const {
-    return MetricsImpl::CollectReadMetrics(readers_);
+    std::shared_ptr<Metrics> metrics = MetricsImpl::CollectReadMetrics(readers_);
+    metrics->Merge(completed_metrics_);
+    return metrics;
 }
 
 Result<BatchReader::ReadBatchWithBitmap> ConcatBatchReader::NextBatchWithBitmap() {

src/paimon/common/reader/concat_batch_reader.h

@@ -36,7 +36,8 @@ class MemoryPool;
 class ConcatBatchReader : public BatchReader {
  public:
     ConcatBatchReader(std::vector<std::unique_ptr<BatchReader>>&& readers,
-                      const std::shared_ptr<MemoryPool>& pool);
+                      const std::shared_ptr<MemoryPool>& pool,
+                      const std::shared_ptr<Metrics>& completed_metrics = nullptr);
 
     Result<ReadBatch> NextBatch() override;
     Result<ReadBatchWithBitmap> NextBatchWithBitmap() override;
@@ -46,6 +47,7 @@ class ConcatBatchReader : public BatchReader {
  private:
     std::unique_ptr<arrow::MemoryPool> arrow_pool_;
     std::vector<std::unique_ptr<BatchReader>> readers_;
+    std::shared_ptr<Metrics> completed_metrics_;
     size_t current_;
 };
 }  // namespace paimon

src/paimon/common/reader/concat_batch_reader_test.cpp

@@ -29,6 +29,7 @@
 #include "arrow/array/array_nested.h"
 #include "arrow/ipc/json_simple.h"
 #include "gtest/gtest.h"
+#include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/status.h"
 #include "paimon/testing/mock/mock_file_batch_reader.h"
@@ -169,4 +170,24 @@ TEST_F(ConcatBatchReaderTest, TestSimpleWithBitmap) {
     }
 }
 
+TEST_F(ConcatBatchReaderTest, TestMergeCompletedReaderMetrics) {
+    std::shared_ptr<arrow::StructArray> data =
+        arrow::StructArray::Make(
+            {arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[1, 2, 3]").ValueOrDie()},
+            {arrow::field("f1", arrow::int32())})
+            .ValueOrDie();
+    std::vector<std::unique_ptr<BatchReader>> readers;
+    readers.push_back(
+        std::make_unique<MockFileBatchReader>(data, data->type(), /*read_batch_size=*/2));
+
+    std::shared_ptr<Metrics> completed_metrics = std::make_shared<MetricsImpl>();
+    completed_metrics->SetCounter("mock.number.of.rows", 4);
+    std::unique_ptr<ConcatBatchReader> concat_reader = std::make_unique<ConcatBatchReader>(
+        std::move(readers), GetDefaultPool(), completed_metrics);
+
+    ASSERT_OK_AND_ASSIGN(uint64_t row_count,
+                         concat_reader->GetReaderMetrics()->GetCounter("mock.number.of.rows"));
+    ASSERT_EQ(7, row_count);
+}
+
 }  // namespace paimon::test

src/paimon/common/reader/late_materialization_batch_reader.cpp

@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/common/reader/late_materialization_batch_reader.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/array/array_base.h"
+#include "arrow/array/array_nested.h"
+#include "arrow/c/bridge.h"
+#include "arrow/util/checked_cast.h"
+#include "fmt/format.h"
+#include "paimon/common/metrics/metrics_impl.h"
+#include "paimon/common/reader/reader_utils.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+Result<std::unique_ptr<LateMaterializationBatchReader>> LateMaterializationBatchReader::Create(
+    const std::shared_ptr<arrow::Schema>& read_schema,
+    const std::shared_ptr<arrow::Schema>& probe_schema,
+    std::shared_ptr<arrow::StructArray> probe_data,
+    const std::shared_ptr<arrow::Schema>& payload_schema,
+    std::unique_ptr<BatchReader>&& payload_reader, int32_t read_batch_size,
+    const std::shared_ptr<MemoryPool>& pool, std::unique_ptr<arrow::MemoryPool> arrow_pool) {
+    if (!read_schema || !probe_schema || !payload_schema || !probe_data) {
+        return Status::Invalid("late materialization reader requires non-null schemas and data");
+    }
+    if (read_batch_size <= 0) {
+        return Status::Invalid("late materialization read batch size should be positive");
+    }
+
+    std::vector<FieldSource> field_sources;
+    field_sources.reserve(read_schema->num_fields());
+    bool needs_payload = false;
+    for (const std::shared_ptr<arrow::Field>& read_field : read_schema->fields()) {
+        int32_t probe_idx = probe_schema->GetFieldIndex(read_field->name());
+        if (probe_idx >= 0) {
+            field_sources.push_back({Source::PROBE, probe_idx});
+            continue;
+        }
+        int32_t payload_idx = payload_schema->GetFieldIndex(read_field->name());
+        if (payload_idx >= 0) {
+            field_sources.push_back({Source::PAYLOAD, payload_idx});
+            needs_payload = true;
+            continue;
+        }
+        return Status::Invalid(fmt::format(
+            "field {} is missing from both probe and payload schemas", read_field->name()));
+    }
+    if (needs_payload && !payload_reader) {
+        return Status::Invalid(
+            "late materialization reader requires a payload reader for payload fields");
+    }
+
+    return std::unique_ptr<LateMaterializationBatchReader>(new LateMaterializationBatchReader(
+        read_schema, std::move(probe_data), std::move(payload_reader), std::move(field_sources),
+        read_batch_size, pool, std::move(arrow_pool)));
+}
+
+LateMaterializationBatchReader::LateMaterializationBatchReader(
+    const std::shared_ptr<arrow::Schema>& read_schema,
+    std::shared_ptr<arrow::StructArray> probe_data, std::unique_ptr<BatchReader>&& payload_reader,
+    std::vector<FieldSource>&& field_sources, int32_t read_batch_size,
+    const std::shared_ptr<MemoryPool>& pool, std::unique_ptr<arrow::MemoryPool> arrow_pool)
+    : arrow_pool_(arrow_pool ? std::move(arrow_pool) : GetArrowPool(pool)),
+      read_schema_(read_schema),
+      probe_data_(std::move(probe_data)),
+      payload_reader_(std::move(payload_reader)),
+      field_sources_(std::move(field_sources)),
+      read_batch_size_(read_batch_size) {}
+
+Result<BatchReader::ReadBatch> LateMaterializationBatchReader::NextBatch() {
+    PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap batch_with_bitmap, NextBatchWithBitmap());
+    return ReaderUtils::ApplyBitmapToReadBatch(std::move(batch_with_bitmap), arrow_pool_.get());
+}
+
+Result<BatchReader::ReadBatchWithBitmap> LateMaterializationBatchReader::NextBatchWithBitmap() {
+    if (payload_reader_) {
+        PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap payload_batch_with_bitmap,
+                               payload_reader_->NextBatchWithBitmap());
+        if (BatchReader::IsEofBatch(payload_batch_with_bitmap)) {
+            if (probe_offset_ != probe_data_->length()) {
+                return Status::Invalid(fmt::format(
+                    "late materialization payload ended at {}, but probe row count is {}",
+                    probe_offset_, probe_data_->length()));
+            }
+            return BatchReader::MakeEofBatchWithBitmap();
+        }
+
+        ReadBatchWithBitmap moved_payload_batch = std::move(payload_batch_with_bitmap);
+        ReadBatch& payload_batch = moved_payload_batch.first;
+        RoaringBitmap32& payload_bitmap = moved_payload_batch.second;
+        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+            std::shared_ptr<arrow::Array> payload_array,
+            arrow::ImportArray(payload_batch.first.get(), payload_batch.second.get()));
+        if (payload_bitmap.Cardinality() != payload_array->length()) {
+            PAIMON_ASSIGN_OR_RAISE(
+                arrow::ArrayVector filtered_payload,
+                ReaderUtils::GenerateFilteredArrayVector(payload_array, payload_bitmap));
+            PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+                payload_array, arrow::Concatenate(filtered_payload, arrow_pool_.get()));
+        }
+        std::shared_ptr<arrow::StructArray> payload_struct =
+            arrow::internal::checked_pointer_cast<arrow::StructArray>(payload_array);
+        return MakeBatch(payload_struct);
+    }
+
+    if (probe_offset_ >= probe_data_->length()) {
+        return BatchReader::MakeEofBatchWithBitmap();
+    }
+    return MakeBatch(/*payload_data=*/nullptr);
+}
+
+Result<BatchReader::ReadBatchWithBitmap> LateMaterializationBatchReader::MakeBatch(
+    const std::shared_ptr<arrow::StructArray>& payload_data) {
+    int64_t length =
+        payload_data ? payload_data->length()
+                     : std::min<int64_t>(read_batch_size_, probe_data_->length() - probe_offset_);
+    if (probe_offset_ + length > probe_data_->length()) {
+        return Status::Invalid(fmt::format(
+            "late materialization payload row count exceeds probe row count: offset {}, length {}, "
+            "probe {}",
+            probe_offset_, length, probe_data_->length()));
+    }
+
+    std::shared_ptr<arrow::StructArray> probe_slice =
+        arrow::internal::checked_pointer_cast<arrow::StructArray>(
+            probe_data_->Slice(probe_offset_, length));
+    arrow::ArrayVector arrays;
+    arrays.reserve(field_sources_.size());
+    for (const FieldSource& field_source : field_sources_) {
+        if (field_source.source == Source::PROBE) {
+            PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+                std::shared_ptr<arrow::Array> normalized_probe_field,
+                arrow::Concatenate({probe_slice->field(field_source.index)}, arrow_pool_.get()));
+            arrays.push_back(std::move(normalized_probe_field));
+        } else {
+            if (!payload_data) {
+                return Status::Invalid(
+                    "late materialization payload field requested without "
+                    "payload reader");
+            }
+            arrays.push_back(payload_data->field(field_source.index));
+        }
+    }
+
+    std::shared_ptr<arrow::StructArray> struct_array = std::make_shared<arrow::StructArray>(
+        arrow::struct_(read_schema_->fields()), length, arrays);
+    std::unique_ptr<ArrowArray> c_array = std::make_unique<ArrowArray>();
+    std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(
+        arrow::ExportArray(*struct_array, c_array.get(), c_schema.get()));
+
+    RoaringBitmap32 bitmap;
+    bitmap.AddRange(0, static_cast<int32_t>(length));
+    probe_offset_ += length;
+    return std::make_pair(std::make_pair(std::move(c_array), std::move(c_schema)),
+                          std::move(bitmap));
+}
+
+std::shared_ptr<Metrics> LateMaterializationBatchReader::GetReaderMetrics() const {
+    if (payload_reader_) {
+        return payload_reader_->GetReaderMetrics();
+    }
+    return std::make_shared<MetricsImpl>();
+}
+
+void LateMaterializationBatchReader::Close() {
+    if (payload_reader_) {
+        payload_reader_->Close();
+    }
+}
+
+}  // namespace paimon