[core] Remap predicate field index to projected read schema in InternalReadContext

duanyyyyyyy · duanyyyyyyy · commit cf44831f258a · 2026-06-04T20:55:39.000+08:00
Predicates are constructed against the latest table schema, so each
LeafPredicate carries a field index that points to the column's position
in the full table schema. When the query projects a subset of columns
the read_schema built inside InternalReadContext::Create lays those
columns out at different positions; the leaf's field index no longer
matches the column it names.

The strict field-id validation that runs immediately afterwards then
fails:

    Paimon TableRead::Create error: Invalid: field obs_index has field
    idx 0 in input schema, mismatch field idx 1 in predicate

The downstream LeafPredicateImpl::Test paths use field_index_ directly
to index into arrow arrays / internal rows built from read_schema, so
even if validation were relaxed, leaving the mismatch in place would
silently read the wrong column.

Walk the predicate tree once at context construction and rebuild each
LeafPredicate with the field index resolved from the read_schema by
field name. CompoundPredicate is reconstructed only when any descendant
actually changed, otherwise the original shared_ptr is reused.
GetPredicate() now returns the remapped predicate; the original (with
latest-schema indices) is still available via the wrapped ReadContext if
ever needed.

CreateWithSchema lays the context over a different read_schema (e.g.
the minimal column set for COUNT(*)), so it also remaps from the
original FE predicate against the new read_schema rather than copying
the already-remapped one whose indices are aligned with the original
read_schema.

Repro:

    -- DE table with btree global index on obs_index
    CREATE TABLE t (
      clip_id STRING, obs_index INT, time_offset_ms BIGINT, collected_date DATE
    ) PARTITIONED BY (collected_date, clip_id)
    TBLPROPERTIES (
      'data-evolution.enabled' = 'true',
      'global-index.btree.index-column' = 'obs_index',
      'bucket' = '-1'
    );
    -- after INSERT + CALL paimon.sys.create_global_index(...)
    SELECT clip_id, obs_index, time_offset_ms FROM t
    WHERE collected_date = '2026-05-26' AND clip_id = 'clip_a'
      AND obs_index BETWEEN 0 AND 10;
    -- before: TableRead::Create error (field idx mismatch)
    -- after:  returns matching rows

Coverage: InternalReadContext gains four cases —
TestPredicateFieldIdxRemappedWhenProjected,
TestPredicateUnchangedWhenAligned,
TestCompoundPredicateRemap,
TestPredicateOnFieldMissingFromReadSchema —
covering the projected/aligned/compound/missing-field paths through
the remap.

Signed-off-by: duanyyyyyyy &lt;yan.duan9759@gmail.com&gt;
diff --git a/include/paimon/predicate/leaf_predicate.h b/include/paimon/predicate/leaf_predicate.h
@@ -32,6 +32,11 @@ enum class FieldType;
 /// Leaf node of a `Predicate` tree. Compares a field with literals.
 class PAIMON_EXPORT LeafPredicate : virtual public Predicate {
  public:
+    /// The field's position in the schema this predicate is currently bound to.
+    /// At construction the value reflects the schema the caller supplied to
+    /// `PredicateBuilder`; predicates obtained from `InternalReadContext::GetPredicate()`
+    /// have already been projected onto the read schema (see
+    /// `InternalReadContext::GetPredicate()` for the projection semantics).
     int32_t FieldIndex() const {
         return field_index_;
     }
diff --git a/include/paimon/predicate/predicate_builder.h b/include/paimon/predicate/predicate_builder.h
@@ -32,15 +32,28 @@ enum class FieldType;
 ///
 /// PredicateBuilder provides static factory methods to create various types of predicates
 /// that can be used for filtering data in Paimon tables.
+///
+/// The `field_index` parameter accepted by every factory method is the position of the
+/// field in the schema the caller is working with — typically the latest table schema.
+/// When the resulting predicate is later attached to an `InternalReadContext`,
+/// `InternalReadContext::Create` projects each leaf onto the read schema (mirrors
+/// paimon Java's `PredicateProjectionConverter`): leaf field indices are rewritten via
+/// the table-schema → read-schema position mapping (keyed by the stable paimon field
+/// id, so it survives column renames), and leaves / OR branches whose fields are not
+/// in the projection are dropped. Callers therefore do not need to track projection
+/// state themselves. `field_name` is informational (used for debug / display) and
+/// does not participate in projection lookup.
 class PAIMON_EXPORT PredicateBuilder {
  public:
     PredicateBuilder() = delete;
     ~PredicateBuilder() = delete;
 
     /// Create an equality predicate (field == literal).
     ///
-    /// @param field_index The index of the field in read schema (0-based).
-    /// @param field_name The name of the field.
+    /// @param field_index The position of the field in the schema the caller is working
+    ///                    with (0-based); projected onto the read schema by
+    ///                    `InternalReadContext::Create`. See class doc for details.
+    /// @param field_name The name of the field (informational; see class doc).
     /// @param field_type The data type of the field.
     /// @param literal The literal value to compare against.
     /// @return A shared pointer to the created Predicate object.
@@ -99,8 +112,10 @@ class PAIMON_EXPORT PredicateBuilder {
     ///
     /// Tests whether the field value falls within the specified range (inclusive on both ends).
     ///
-    /// @param field_index The index of the field in read schema (0-based).
-    /// @param field_name The name of the field.
+    /// @param field_index The position of the field in the schema the caller is working
+    ///                    with (0-based); projected onto the read schema by
+    ///                    `InternalReadContext::Create`. See class doc for details.
+    /// @param field_name The name of the field (informational; see class doc).
     /// @param field_type The data type of the field.
     /// @param included_lower_bound The lower bound of the range (inclusive).
     /// @param included_upper_bound The upper bound of the range (inclusive).
diff --git a/include/paimon/read_context.h b/include/paimon/read_context.h
@@ -206,6 +206,13 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// It can significantly improve performance by reducing the amount of data
     /// that needs to be read and processed.
     ///
+    /// The caller should construct the predicate against the latest table schema.
+    /// `InternalReadContext::Create` projects each leaf onto the read schema
+    /// (mirroring paimon Java's `PredicateProjectionConverter`): leaf field indices
+    /// are rewritten to positions in the read schema, and AND children / OR branches
+    /// whose fields are not in the projection are pruned. The predicate therefore
+    /// does not need to be projection-aware.
+    ///
     /// @param predicate Shared pointer to the predicate for data filtering.
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& SetPredicate(const std::shared_ptr<Predicate>& predicate);
diff --git a/src/paimon/common/predicate/predicate_builder.cpp b/src/paimon/common/predicate/predicate_builder.cpp
@@ -42,8 +42,6 @@
 namespace paimon {
 enum class FieldType;
 
-// TODO(xinyu.lxy): predicate field_index use index in read schema now, but java paimon use index
-// in file schema
 std::shared_ptr<Predicate> PredicateBuilder::Equal(int32_t field_index,
                                                    const std::string& field_name,
                                                    const FieldType& field_type,
diff --git a/src/paimon/core/operation/internal_read_context.cpp b/src/paimon/core/operation/internal_read_context.cpp
@@ -16,19 +16,124 @@
 
 #include "paimon/core/operation/internal_read_context.h"
 
+#include <optional>
 #include <utility>
 
+#include "paimon/common/predicate/compound_predicate_impl.h"
+#include "paimon/common/predicate/leaf_predicate_impl.h"
 #include "paimon/common/predicate/predicate_validator.h"
 #include "paimon/common/table/special_fields.h"
 #include "paimon/common/types/data_field.h"
 #include "paimon/core/schema/arrow_schema_validator.h"
+#include "paimon/predicate/function.h"
 #include "paimon/status.h"
 
 namespace arrow {
 class Schema;
 }  // namespace arrow
 
 namespace paimon {
+namespace {
+// Build a map from a field's position in `table_schema` (the latest table schema, the
+// index space upstream predicates are typically constructed against) to its position
+// in `read_data_fields` (the projected read schema). Field identity is the stable
+// field id, so the mapping survives column renames within the same schema. Read-only
+// special fields (RowId / SequenceNumber / etc.) have no analogue in the table
+// schema and are skipped — user-supplied predicates do not reference them.
+std::map<int32_t, int32_t> BuildLatestToReadIdxMapping(
+    const TableSchema& table_schema, const std::vector<DataField>& read_data_fields) {
+    std::map<int32_t, int32_t> id_to_latest_idx;
+    const auto& table_fields = table_schema.Fields();
+    for (size_t latest_idx = 0; latest_idx < table_fields.size(); latest_idx++) {
+        id_to_latest_idx[table_fields[latest_idx].Id()] = static_cast<int32_t>(latest_idx);
+    }
+    std::map<int32_t, int32_t> mapping;
+    for (size_t read_idx = 0; read_idx < read_data_fields.size(); read_idx++) {
+        auto iter = id_to_latest_idx.find(read_data_fields[read_idx].Id());
+        if (iter != id_to_latest_idx.end()) {
+            mapping[iter->second] = static_cast<int32_t>(read_idx);
+        }
+    }
+    return mapping;
+}
+
+// Project `predicate` onto the read schema, rewriting each leaf's field index via
+// `latest_to_read_idx` (predicate's source index space → read schema position).
+//
+// Inclusive semantics, matching paimon Java's PredicateProjectionConverter:
+//   - Leaf whose field is not in the read schema: dropped (returns nullopt).
+//   - AND: drop non-projectable children, keep the rest. Safe because if `A AND B`
+//     holds for a row, A holds too; the projected predicate is a superset
+//     (necessary, not sufficient).
+//   - OR: every child must be projectable. If any child is dropped, drop the
+//     whole OR — otherwise a row that only satisfied the dropped branch would
+//     falsely pass.
+//   - Predicates without a field index (e.g. full-text / vector search) flow
+//     through unchanged.
+Result<std::optional<std::shared_ptr<Predicate>>> ProjectPredicate(
+    const std::map<int32_t, int32_t>& latest_to_read_idx,
+    const std::shared_ptr<Predicate>& predicate) {
+    if (auto leaf = std::dynamic_pointer_cast<LeafPredicateImpl>(predicate)) {
+        auto iter = latest_to_read_idx.find(leaf->FieldIndex());
+        if (iter == latest_to_read_idx.end()) {
+            return std::optional<std::shared_ptr<Predicate>>{};
+        }
+        if (iter->second == leaf->FieldIndex()) {
+            return std::optional<std::shared_ptr<Predicate>>{predicate};
+        }
+        return std::optional<std::shared_ptr<Predicate>>{
+            std::static_pointer_cast<Predicate>(leaf->NewLeafPredicate(iter->second))};
+    }
+    if (auto compound = std::dynamic_pointer_cast<CompoundPredicateImpl>(predicate)) {
+        const bool is_and = compound->GetFunction().GetType() == Function::Type::AND;
+        std::vector<std::shared_ptr<Predicate>> projected_children;
+        projected_children.reserve(compound->Children().size());
+        bool any_changed = false;
+        for (const auto& child : compound->Children()) {
+            PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<Predicate>> projected_child,
+                                   ProjectPredicate(latest_to_read_idx, child));
+            if (!projected_child.has_value()) {
+                if (!is_and) {
+                    return std::optional<std::shared_ptr<Predicate>>{};
+                }
+                any_changed = true;
+                continue;
+            }
+            if (projected_child.value() != child) {
+                any_changed = true;
+            }
+            projected_children.push_back(std::move(projected_child.value()));
+        }
+        if (projected_children.empty()) {
+            return std::optional<std::shared_ptr<Predicate>>{};
+        }
+        if (projected_children.size() == 1) {
+            return std::optional<std::shared_ptr<Predicate>>{std::move(projected_children[0])};
+        }
+        if (!any_changed) {
+            return std::optional<std::shared_ptr<Predicate>>{predicate};
+        }
+        return std::optional<std::shared_ptr<Predicate>>{std::static_pointer_cast<Predicate>(
+            compound->NewCompoundPredicate(projected_children))};
+    }
+    return std::optional<std::shared_ptr<Predicate>>{predicate};
+}
+
+Result<std::shared_ptr<Predicate>> ProjectAndValidatePredicate(
+    const arrow::Schema& read_schema, const std::map<int32_t, int32_t>& latest_to_read_idx,
+    const std::shared_ptr<Predicate>& predicate) {
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<Predicate>> projected,
+                           ProjectPredicate(latest_to_read_idx, predicate));
+    if (!projected.has_value()) {
+        return std::shared_ptr<Predicate>{};
+    }
+    PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema(
+        read_schema, projected.value(), /*validate_field_idx=*/true));
+    PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithLiterals(projected.value()));
+    return projected.value();
+}
+}  // namespace
+
 Result<std::unique_ptr<InternalReadContext>> InternalReadContext::Create(
     const std::shared_ptr<ReadContext>& context, const std::shared_ptr<TableSchema>& table_schema,
     const std::map<std::string, std::string>& options) {
@@ -95,34 +200,51 @@ Result<std::unique_ptr<InternalReadContext>> InternalReadContext::Create(
     auto read_schema = DataField::ConvertDataFieldsToArrowSchema(read_data_fields);
     // validate read schema to avoid redundant fields
     PAIMON_RETURN_NOT_OK(ArrowSchemaValidator::ValidateSchemaWithFieldId(*read_schema));
-    // validate predicate
+    // Project the upstream predicate onto `read_schema`. Predicates carry field indices
+    // pointing into the latest table schema; rewrite them to positions in `read_schema_`
+    // so downstream readers can apply them directly.
+    std::shared_ptr<Predicate> projected_predicate;
     if (context->GetPredicate()) {
-        PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema(
-            *read_schema, context->GetPredicate(), /*validate_field_idx=*/true));
-        PAIMON_RETURN_NOT_OK(
-            PredicateValidator::ValidatePredicateWithLiterals(context->GetPredicate()));
+        auto latest_to_read_idx = BuildLatestToReadIdxMapping(*table_schema, read_data_fields);
+        PAIMON_ASSIGN_OR_RAISE(
+            projected_predicate,
+            ProjectAndValidatePredicate(*read_schema, latest_to_read_idx, context->GetPredicate()));
     }
 
-    return std::unique_ptr<InternalReadContext>(
-        new InternalReadContext(context, table_schema, read_schema, core_options));
+    return std::unique_ptr<InternalReadContext>(new InternalReadContext(
+        context, table_schema, read_schema, core_options, std::move(projected_predicate)));
 }
 
 InternalReadContext::InternalReadContext(const std::shared_ptr<ReadContext>& read_context,
                                          const std::shared_ptr<TableSchema>& table_schema,
                                          const std::shared_ptr<arrow::Schema>& read_schema,
-                                         const CoreOptions& options)
+                                         const CoreOptions& options,
+                                         std::shared_ptr<Predicate> projected_predicate)
     : read_context_(read_context),
       table_schema_(table_schema),
       read_schema_(read_schema),
-      options_(options) {}
+      options_(options),
+      projected_predicate_(std::move(projected_predicate)) {}
 
 Result<std::shared_ptr<InternalReadContext>> InternalReadContext::CreateWithSchema(
     const std::shared_ptr<InternalReadContext>& original,
     const std::shared_ptr<arrow::Schema>& new_read_schema) {
-    // Create a new InternalReadContext sharing all properties except read_schema.
-    // The new read_schema is the minimal column set for COUNT(*).
-    return std::shared_ptr<InternalReadContext>(new InternalReadContext(
-        original->read_context_, original->table_schema_, new_read_schema, original->options_));
+    // The wrapped read_context still holds the caller-supplied predicate (in the latest
+    // table schema's index space). Re-project it against `new_read_schema` directly,
+    // rather than re-projecting the already-projected predicate sitting on `original`.
+    std::shared_ptr<Predicate> projected_predicate;
+    if (original->read_context_->GetPredicate()) {
+        PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> new_read_data_fields,
+                               DataField::ConvertArrowSchemaToDataFields(new_read_schema));
+        auto latest_to_read_idx =
+            BuildLatestToReadIdxMapping(*original->table_schema_, new_read_data_fields);
+        PAIMON_ASSIGN_OR_RAISE(projected_predicate, ProjectAndValidatePredicate(
+                                                        *new_read_schema, latest_to_read_idx,
+                                                        original->read_context_->GetPredicate()));
+    }
+    return std::shared_ptr<InternalReadContext>(
+        new InternalReadContext(original->read_context_, original->table_schema_, new_read_schema,
+                                original->options_, std::move(projected_predicate)));
 }
 
 }  // namespace paimon
diff --git a/src/paimon/core/operation/internal_read_context.h b/src/paimon/core/operation/internal_read_context.h
@@ -62,8 +62,16 @@ class InternalReadContext {
     const std::vector<std::string>& GetPrimaryKeys() const {
         return table_schema_->PrimaryKeys();
     }
+    // Returns the predicate projected onto `read_schema_`. Upstream constructs predicates
+    // against the latest table schema, so when the query projects a subset of columns
+    // the leaf field indices no longer match the projected schema. The projection is
+    // done once at context construction (mirrors paimon Java's
+    // `PredicateProjectionConverter`): leaf indices are rewritten via the table-schema
+    // → read-schema mapping, AND children whose fields are absent from the read schema
+    // are dropped (inclusive), and OR is dropped wholesale if any of its children is not
+    // projectable. May be nullptr if the entire predicate is non-projectable.
     const std::shared_ptr<Predicate>& GetPredicate() const {
-        return read_context_->GetPredicate();
+        return projected_predicate_;
     }
     bool EnablePredicateFilter() const {
         return read_context_->EnablePredicateFilter();
@@ -110,12 +118,13 @@ class InternalReadContext {
     InternalReadContext(const std::shared_ptr<ReadContext>& read_context,
                         const std::shared_ptr<TableSchema>& table_schema,
                         const std::shared_ptr<arrow::Schema>& read_schema,
-                        const CoreOptions& options);
+                        const CoreOptions& options, std::shared_ptr<Predicate> projected_predicate);
 
     std::shared_ptr<ReadContext> read_context_;
     std::shared_ptr<TableSchema> table_schema_;
     std::shared_ptr<arrow::Schema> read_schema_;
     CoreOptions options_;
+    std::shared_ptr<Predicate> projected_predicate_;
 };
 
 }  // namespace paimon
diff --git a/src/paimon/core/operation/internal_read_context_test.cpp b/src/paimon/core/operation/internal_read_context_test.cpp
diff --git a/src/paimon/core/table/source/table_read_test.cpp b/src/paimon/core/table/source/table_read_test.cpp
