feat(read): support file index and predicate pushdown for data evolution (#215)

Author: lxy-9602

src/paimon/core/operation/data_evolution_split_read.cpp

@@ -22,6 +22,7 @@
 #include <cassert>
 #include <limits>
 #include <map>
+#include <set>
 #include <string>
 #include <string_view>
 #include <thread>
@@ -53,8 +54,13 @@
 #include "paimon/core/core_options.h"
 #include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
 #include "paimon/core/global_index/indexed_split_impl.h"
+#include "paimon/core/io/file_index_evaluator.h"
 #include "paimon/core/utils/blob_view_lookup.h"
 #include "paimon/core/utils/data_evolution_utils.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/file_index/file_index_result.h"
+#include "paimon/predicate/predicate_utils.h"
 
 namespace paimon {
 namespace {
@@ -389,12 +395,22 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::InnerCreateReader(
         path_factory_->CreateDataFilePathFactory(split_impl->Partition(), split_impl->Bucket()));
     auto metas = split_impl->DataFiles();
     DeletionVector::Factory split_dv_factory = CreateSplitDvFactory(*split_impl);
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Predicate> push_down_predicate,
+                           CreatePushDownPredicate(context_->GetPredicate(), raw_read_schema_));
 
     PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>> split_by_row_id,
                            MergeRangesAndSort(std::move(metas)));
 
     std::vector<std::unique_ptr<BatchReader>> sub_readers;
     for (const std::vector<std::shared_ptr<DataFileMeta>>& need_merge_files : split_by_row_id) {
+        if (need_merge_files.size() > 1) {
+            PAIMON_ASSIGN_OR_RAISE(
+                bool skip_group,
+                SkipByFileIndex(push_down_predicate, need_merge_files, data_file_path_factory));
+            if (skip_group) {
+                continue;
+            }
+        }
         PAIMON_ASSIGN_OR_RAISE(std::optional<GroupDeletionVector> group_dv,
                                ReadGroupDeletionVector(need_merge_files, split_dv_factory));
         PAIMON_ASSIGN_OR_RAISE(DeletionVector::Factory group_dv_factory,
@@ -404,10 +420,15 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::InnerCreateReader(
             PAIMON_ASSIGN_OR_RAISE(
                 std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers,
                 CreateRawFileReaders(split_impl->Partition(), need_merge_files, raw_read_schema_,
-                                     /*predicate=*/nullptr, group_dv_factory, row_ranges,
+                                     push_down_predicate, group_dv_factory, row_ranges,
                                      data_file_path_factory,
                                      /*extra_format_options=*/{}));
-            assert(raw_file_readers.size() == 1);
+            if (raw_file_readers.empty()) {
+                continue;
+            }
+            if (raw_file_readers.size() != 1) {
+                return Status::Invalid("Single-file data evolution group created multiple readers");
+            }
             sub_readers.push_back(std::move(raw_file_readers[0]));
         } else {
             PAIMON_ASSIGN_OR_RAISE(
@@ -424,17 +445,110 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::InnerCreateReader(
     return std::make_unique<CompleteRowKindBatchReader>(std::move(batch_reader), pool_);
 }
 
+Result<std::shared_ptr<Predicate>> DataEvolutionSplitRead::CreatePushDownPredicate(
+    const std::shared_ptr<Predicate>& predicate,
+    const std::shared_ptr<arrow::Schema>& read_schema) {
+    std::map<std::string, int32_t> picked_field_name_to_idx;
+    for (int32_t i = 0; i < read_schema->num_fields(); ++i) {
+        const std::string& field_name = read_schema->field(i)->name();
+        if (!SpecialFields::IsSystemField(field_name)) {
+            picked_field_name_to_idx.emplace(field_name, i);
+        }
+    }
+    return PredicateUtils::CreatePickedFieldFilter(predicate, picked_field_name_to_idx);
+}
+
+Result<bool> DataEvolutionSplitRead::SkipByFileIndex(
+    const std::shared_ptr<Predicate>& predicate,
+    const std::vector<std::shared_ptr<DataFileMeta>>& files,
+    const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+    if (!options_.FileIndexReadEnabled() || !predicate) {
+        return false;
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(
+        std::unique_ptr<FieldMappingBuilder> field_mapping_builder,
+        FieldMappingBuilder::Create(raw_read_schema_, context_->GetPartitionKeys(), predicate));
+    std::set<int32_t> claimed_field_ids;
+    for (const auto& file : files) {
+        // Blob and vector-store files may cover only part of the row range, so their indexes
+        // cannot prove that the complete merged group misses the predicate.
+        if (!DataEvolutionUtils::IsNormalFile(file->file_name)) {
+            continue;
+        }
+
+        std::shared_ptr<TableSchema> data_schema = context_->GetTableSchema();
+        if (file->schema_id != data_schema->Id()) {
+            PAIMON_ASSIGN_OR_RAISE(data_schema, schema_manager_->ReadSchema(file->schema_id));
+        }
+        std::vector<DataField> written_fields;
+        if (file->write_cols) {
+            std::vector<std::string> data_write_cols;
+            data_write_cols.reserve(file->write_cols->size());
+            for (const auto& write_col : file->write_cols.value()) {
+                if (!SpecialFields::IsSystemField(write_col)) {
+                    data_write_cols.push_back(write_col);
+                }
+            }
+            PAIMON_ASSIGN_OR_RAISE(written_fields, data_schema->GetFields(data_write_cols));
+        } else {
+            written_fields = data_schema->Fields();
+        }
+
+        std::set<std::string> overwritten_field_names;
+        for (const auto& field : written_fields) {
+            if (!claimed_field_ids.insert(field.Id()).second) {
+                overwritten_field_names.insert(field.Name());
+            }
+        }
+
+        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FieldMapping> field_mapping,
+                               field_mapping_builder->CreateFieldMapping(written_fields));
+        std::shared_ptr<Predicate> data_predicate =
+            field_mapping->non_partition_info.non_partition_filter;
+        if (!overwritten_field_names.empty()) {
+            PAIMON_ASSIGN_OR_RAISE(data_predicate, PredicateUtils::ExcludePredicateWithFields(
+                                                       data_predicate, overwritten_field_names));
+        }
+        if (!data_predicate) {
+            continue;
+        }
+
+        auto written_schema = DataField::ConvertDataFieldsToArrowSchema(written_fields);
+        PAIMON_ASSIGN_OR_RAISE(
+            std::shared_ptr<FileIndexResult> index_result,
+            FileIndexEvaluator::Evaluate(written_schema, data_predicate, data_file_path_factory,
+                                         file, options_.GetFileSystem(), pool_));
+        PAIMON_ASSIGN_OR_RAISE(bool is_remain, index_result->IsRemain());
+        if (!is_remain) {
+            return true;
+        }
+    }
+    return false;
+}
+
 Result<std::unique_ptr<FileBatchReader>> DataEvolutionSplitRead::ApplyIndexAndDvReaderIfNeeded(
     std::unique_ptr<FileBatchReader>&& file_reader, const std::shared_ptr<DataFileMeta>& file,
     const std::shared_ptr<arrow::Schema>& data_schema,
     const std::shared_ptr<arrow::Schema>& read_schema, const std::shared_ptr<Predicate>& predicate,
     DeletionVector::Factory dv_factory, const std::optional<std::vector<Range>>& row_ranges,
     const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
-    if (predicate) {
-        assert(false);
-        // as DataEvolutionSplitRead will skip predicate
-        return Status::Invalid("DataEvolutionSplitRead do not support predicate");
+    std::shared_ptr<FileIndexResult> file_index_result;
+    if (options_.FileIndexReadEnabled()) {
+        PAIMON_ASSIGN_OR_RAISE(
+            file_index_result,
+            FileIndexEvaluator::Evaluate(data_schema, predicate, data_file_path_factory, file,
+                                         options_.GetFileSystem(), pool_));
+        PAIMON_ASSIGN_OR_RAISE(bool is_remain, file_index_result->IsRemain());
+        if (!is_remain) {
+            return std::unique_ptr<FileBatchReader>();
+        }
+    }
+    const RoaringBitmap32* index_selection = nullptr;
+    if (auto* bitmap_index = dynamic_cast<BitmapIndexResult*>(file_index_result.get())) {
+        PAIMON_ASSIGN_OR_RAISE(index_selection, bitmap_index->GetBitmap());
     }
+
     // the factory is per row range group and already returns a view taking file-local positions.
     // Unlike RawFileSplitRead the vector is not folded into the format reader's selection: it is
     // no BitmapDeletionVector, and the blob fallback path's gap segments have no format reader.
@@ -444,10 +558,19 @@ Result<std::unique_ptr<FileBatchReader>> DataEvolutionSplitRead::ApplyIndexAndDv
     }
     PAIMON_ASSIGN_OR_RAISE(std::optional<RoaringBitmap32> selection_row_ids,
                            file->ToFileSelection(row_ranges));
+    if (index_selection) {
+        if (selection_row_ids) {
+            selection_row_ids.value() &= *index_selection;
+        } else {
+            selection_row_ids = *index_selection;
+        }
+    }
+    if (selection_row_ids && selection_row_ids->IsEmpty()) {
+        return std::unique_ptr<FileBatchReader>();
+    }
     ::ArrowSchema c_read_schema;
     PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema, &c_read_schema));
-    PAIMON_RETURN_NOT_OK(
-        file_reader->SetReadSchema(&c_read_schema, /*predicate=*/nullptr, selection_row_ids));
+    PAIMON_RETURN_NOT_OK(file_reader->SetReadSchema(&c_read_schema, predicate, selection_row_ids));
 
     std::unique_ptr<FileBatchReader> reader;
     if (!file_reader->SupportPreciseBitmapSelection() && selection_row_ids) {

src/paimon/core/operation/data_evolution_split_read.h

@@ -68,9 +68,10 @@ struct DeletionFile;
 /// ->(DelegatingPrefetchReader)->(PrefetchFileBatchReader)->FormatReader
 ///
 ///
-/// A union `SplitRead` to read multiple inner files to merge columns, note that this class
-/// does not support filtering push down: a predicate would have to be evaluated consistently
-/// across the files being merged, which is not implemented here.
+/// A union `SplitRead` to read multiple inner files to merge columns. A single-file row range
+/// group gets both file-index and format-level predicate pushdown. A merged group only uses file
+/// indexes to skip the whole group: filtering its child readers independently would break their
+/// positional alignment.
 ///
 /// Deletion vectors are supported: a row range group's vector is maintained against the
 /// group's anchor file (DataEvolutionUtils::RetrieveAnchorFile), so its positions are
@@ -172,6 +173,21 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
         const std::shared_ptr<DataSplit>& data_split,
         const std::optional<std::vector<Range>>& row_ranges) const;
 
+    /// Keeps top-level conjuncts whose fields all belong to `read_schema`, excluding conjuncts
+    /// over system fields. The returned predicate is for pushdown only; the original predicate is
+    /// still evaluated as a residual filter when requested by the read context.
+    static Result<std::shared_ptr<Predicate>> CreatePushDownPredicate(
+        const std::shared_ptr<Predicate>& predicate,
+        const std::shared_ptr<arrow::Schema>& read_schema);
+
+    /// Returns true when file indexes prove that no row in a merged row range group can match.
+    /// Only normal files are considered, and an older copy of a field is excluded after a newer
+    /// file has claimed the same field id.
+    Result<bool> SkipByFileIndex(
+        const std::shared_ptr<Predicate>& predicate,
+        const std::vector<std::shared_ptr<DataFileMeta>>& files,
+        const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const;
+
     /// Builds the deletion vector factory over the split's deletion files, keyed by data file
     /// name. Only anchor files carry one. Returns a null factory when the split has none.
     DeletionVector::Factory CreateSplitDvFactory(const DataSplitImpl& split_impl) const;

src/paimon/core/operation/data_evolution_split_read_test.cpp

@@ -25,6 +25,7 @@
 
 #include "gtest/gtest.h"
 #include "paimon/common/data/binary_row.h"
+#include "paimon/common/table/special_fields.h"
 #include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
 #include "paimon/core/io/data_file_meta.h"
 #include "paimon/core/manifest/file_source.h"
@@ -37,6 +38,8 @@
 #include "paimon/executor.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate_builder.h"
 #include "paimon/read_context.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/testharness.h"
@@ -102,6 +105,30 @@ class DataEvolutionSplitReadTest : public ::testing::Test {
     std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
 };
 
+TEST_F(DataEvolutionSplitReadTest, TestCreatePushDownPredicate) {
+    auto f0_predicate =
+        PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT, Literal(1));
+    auto f1_predicate =
+        PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1", FieldType::INT, Literal(2));
+    auto row_id_predicate = PredicateBuilder::Equal(
+        /*field_index=*/2, SpecialFields::RowId().Name(), FieldType::BIGINT, Literal(3l));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
+                         PredicateBuilder::And({f0_predicate, f1_predicate, row_id_predicate}));
+
+    auto read_schema = DataField::ConvertDataFieldsToArrowSchema(
+        {DataField(0, arrow::field("f0", arrow::int32())), SpecialFields::RowId()});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> push_down,
+                         DataEvolutionSplitRead::CreatePushDownPredicate(predicate, read_schema));
+    ASSERT_TRUE(push_down);
+    ASSERT_EQ(*push_down, *f0_predicate);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> or_predicate,
+                         PredicateBuilder::Or({f0_predicate, f1_predicate}));
+    ASSERT_OK_AND_ASSIGN(
+        push_down, DataEvolutionSplitRead::CreatePushDownPredicate(or_predicate, read_schema));
+    ASSERT_FALSE(push_down);
+}
+
 TEST_F(DataEvolutionSplitReadTest, TestAddSingleBlobEntry) {
     auto blob_entry =
         CreateBlobFile("blob1", /*first_row_id=*/0, /*row_count=*/100,