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

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 <yan.duan9759@gmail.com>

Author: duanyyyyyyy

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 rebound to the projected `read_schema`. Use `FieldName()` as
+    /// the schema-independent identity.
     int32_t FieldIndex() const {
         return field_index_;
     }

include/paimon/predicate/predicate_builder.h

@@ -32,15 +32,25 @@ 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.
+/// `field_name` is the stable identity used to locate the field. When the resulting
+/// predicate is later attached to an `InternalReadContext`, `InternalReadContext::Create`
+/// automatically remaps `field_index` to the field's position in the projected
+/// `read_schema` using `field_name`, so callers do not need to track projection state
+/// themselves. `field_name` must therefore name a field present in the read schema.
 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); rebound to the read schema by
+    ///                    `InternalReadContext::Create`. See class doc for details.
+    /// @param field_name The name of the field. Must exist in the read schema.
     /// @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 +109,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); rebound to the read schema by
+    ///                    `InternalReadContext::Create`. See class doc for details.
+    /// @param field_name The name of the field. Must exist in the read schema.
     /// @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).

include/paimon/read_context.h

@@ -206,6 +206,11 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// It can significantly improve performance by reducing the amount of data
     /// that needs to be read and processed.
     ///
+    /// The caller is free to construct the predicate against any schema that contains the
+    /// referenced fields (typically the latest table schema). `InternalReadContext::Create`
+    /// rebinds each leaf's `field_index` to the field's position in the projected read
+    /// schema using `field_name`, so the predicate 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);

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,

src/paimon/core/operation/internal_read_context.cpp

@@ -18,6 +18,8 @@
 
 #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"
@@ -29,6 +31,50 @@ class Schema;
 }  // namespace arrow
 
 namespace paimon {
+namespace {
+// Remap each leaf predicate's field index to the position of its field in
+// `read_schema`. The field name is the stable identity used for lookup. The
+// upstream (Java SDK / FE) constructs predicates against the latest table
+// schema, so when a query projects a subset of columns the predicate's field
+// index no longer matches the position in the projected `read_schema`. The
+// downstream predicate-test code uses the field index directly, so leaving
+// the mismatch in place would silently read the wrong column. Remap once at
+// context construction so every subsequent reader sees a predicate already
+// aligned with `read_schema`.
+Result<std::shared_ptr<Predicate>> RemapPredicateFieldIndex(
+    const arrow::Schema& read_schema, const std::shared_ptr<Predicate>& predicate) {
+    if (auto leaf = std::dynamic_pointer_cast<LeafPredicateImpl>(predicate)) {
+        int32_t new_index = read_schema.GetFieldIndex(leaf->FieldName());
+        if (new_index == -1) {
+            return Status::Invalid(
+                fmt::format("field {} does not exist in schema", leaf->FieldName()));
+        }
+        if (new_index == leaf->FieldIndex()) {
+            return predicate;
+        }
+        return std::static_pointer_cast<Predicate>(leaf->NewLeafPredicate(new_index));
+    }
+    if (auto compound = std::dynamic_pointer_cast<CompoundPredicateImpl>(predicate)) {
+        std::vector<std::shared_ptr<Predicate>> remapped_children;
+        remapped_children.reserve(compound->Children().size());
+        bool any_changed = false;
+        for (const auto& child : compound->Children()) {
+            PAIMON_ASSIGN_OR_RAISE(auto remapped, RemapPredicateFieldIndex(read_schema, child));
+            if (remapped != child) {
+                any_changed = true;
+            }
+            remapped_children.push_back(std::move(remapped));
+        }
+        if (!any_changed) {
+            return predicate;
+        }
+        return std::static_pointer_cast<Predicate>(
+            compound->NewCompoundPredicate(remapped_children));
+    }
+    return predicate;
+}
+}  // 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 +141,48 @@ 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
+    // remap and validate predicate
+    std::shared_ptr<Predicate> remapped_predicate;
     if (context->GetPredicate()) {
+        PAIMON_ASSIGN_OR_RAISE(remapped_predicate,
+                               RemapPredicateFieldIndex(*read_schema, context->GetPredicate()));
         PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema(
-            *read_schema, context->GetPredicate(), /*validate_field_idx=*/true));
-        PAIMON_RETURN_NOT_OK(
-            PredicateValidator::ValidatePredicateWithLiterals(context->GetPredicate()));
+            *read_schema, remapped_predicate, /*validate_field_idx=*/true));
+        PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithLiterals(remapped_predicate));
     }
 
-    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(remapped_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> remapped_predicate)
     : read_context_(read_context),
       table_schema_(table_schema),
       read_schema_(read_schema),
-      options_(options) {}
+      options_(options),
+      remapped_predicate_(std::move(remapped_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_));
+    // Create a new InternalReadContext sharing all properties except read_schema. The
+    // new read_schema is a different column layout (e.g. the minimal column set for
+    // COUNT(*)), so the predicate's field index needs to be re-aligned against it.
+    // Remap from the original (latest-schema-indexed) predicate so the mapping is
+    // computed against new_read_schema directly.
+    std::shared_ptr<Predicate> remapped_predicate;
+    if (original->read_context_->GetPredicate()) {
+        PAIMON_ASSIGN_OR_RAISE(
+            remapped_predicate,
+            RemapPredicateFieldIndex(*new_read_schema, original->read_context_->GetPredicate()));
+    }
+    return std::shared_ptr<InternalReadContext>(
+        new InternalReadContext(original->read_context_, original->table_schema_, new_read_schema,
+                                original->options_, std::move(remapped_predicate)));
 }
 
 }  // namespace paimon

src/paimon/core/operation/internal_read_context.h

@@ -62,8 +62,13 @@ class InternalReadContext {
     const std::vector<std::string>& GetPrimaryKeys() const {
         return table_schema_->PrimaryKeys();
     }
+    // Returns the predicate with each leaf's field index already remapped to its position
+    // in `read_schema_`. Upstream constructs predicates against the latest table schema,
+    // so when the query projects a subset of columns the original indices no longer
+    // match the projected schema. The remapping is done once at context construction; all
+    // subsequent reader code can use this index directly.
     const std::shared_ptr<Predicate>& GetPredicate() const {
-        return read_context_->GetPredicate();
+        return remapped_predicate_;
     }
     bool EnablePredicateFilter() const {
         return read_context_->EnablePredicateFilter();
@@ -110,12 +115,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> remapped_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> remapped_predicate_;
 };
 
 }  // namespace paimon

src/paimon/core/operation/internal_read_context_test.cpp

@@ -25,6 +25,9 @@
 #include "paimon/core/schema/schema_manager.h"
 #include "paimon/defs.h"
 #include "paimon/fs/local/local_file_system.h"
+#include "paimon/predicate/compound_predicate.h"
+#include "paimon/predicate/leaf_predicate.h"
+#include "paimon/predicate/predicate_builder.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/testharness.h"
 
@@ -191,4 +194,99 @@ TEST(InternalReadContext, TestReadWithFieldIdsAndSpecialFields) {
     }
 }
 
+// Upstream predicates are constructed against the latest table schema. When the query
+// projects a subset of columns, the read schema built inside InternalReadContext::Create
+// lays those columns out at different positions and the original field_index no longer
+// matches. The remapping path in Create() must rewrite each leaf predicate so its
+// field_index points to the column's position in the projected read schema.
+TEST(InternalReadContext, TestPredicateFieldIdxRemappedWhenProjected) {
+    std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
+    ReadContextBuilder context_builder(path);
+    // read_schema lays out (f3, f0) — f3 ends up at position 0, f0 at position 1, while
+    // the latest table schema has them at 3 and 0 respectively.
+    context_builder.SetReadSchema({"f3", "f0"});
+    // Predicate was constructed against the latest schema, so f3 carries field_index 3.
+    auto predicate =
+        PredicateBuilder::Equal(/*field_index=*/3, "f3", FieldType::DOUBLE, Literal(1.5));
+    context_builder.SetPredicate(predicate);
+    ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
+    SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), read_context->GetPath());
+    ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+    ASSERT_OK_AND_ASSIGN(auto internal_context,
+                         InternalReadContext::Create(std::move(read_context), table_schema,
+                                                     table_schema->Options()));
+    auto leaf = std::dynamic_pointer_cast<LeafPredicate>(internal_context->GetPredicate());
+    ASSERT_NE(leaf, nullptr);
+    EXPECT_EQ(leaf->FieldName(), "f3");
+    // f3 is the first column in the projected read schema.
+    EXPECT_EQ(leaf->FieldIndex(), 0);
+}
+
+// When the predicate already aligns with the read schema, remapping should be a no-op
+// and return the original shared_ptr without reconstructing the leaf.
+TEST(InternalReadContext, TestPredicateUnchangedWhenAligned) {
+    std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
+    ReadContextBuilder context_builder(path);
+    // No projection -> read schema matches the latest schema: f0(0), f1(1), f2(2), f3(3).
+    auto predicate =
+        PredicateBuilder::Equal(/*field_index=*/3, "f3", FieldType::DOUBLE, Literal(1.5));
+    context_builder.SetPredicate(predicate);
+    ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
+    SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), read_context->GetPath());
+    ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+    ASSERT_OK_AND_ASSIGN(auto internal_context,
+                         InternalReadContext::Create(std::move(read_context), table_schema,
+                                                     table_schema->Options()));
+    // shared_ptr equality: remap returned the input unchanged.
+    EXPECT_EQ(predicate.get(), internal_context->GetPredicate().get());
+}
+
+// CompoundPredicate is recursively remapped: every nested leaf must point to the
+// column's position in the projected read schema.
+TEST(InternalReadContext, TestCompoundPredicateRemap) {
+    std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
+    ReadContextBuilder context_builder(path);
+    context_builder.SetReadSchema({"f3", "f0"});
+    // AND(f3 == 1.5, f0 == "x") with field indices from the latest schema.
+    auto left = PredicateBuilder::Equal(/*field_index=*/3, "f3", FieldType::DOUBLE, Literal(1.5));
+    auto right = PredicateBuilder::Equal(/*field_index=*/0, "f0", FieldType::STRING,
+                                         Literal(FieldType::STRING, "x", 1));
+    ASSERT_OK_AND_ASSIGN(auto compound, PredicateBuilder::And({left, right}));
+    context_builder.SetPredicate(compound);
+    ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
+    SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), read_context->GetPath());
+    ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+    ASSERT_OK_AND_ASSIGN(auto internal_context,
+                         InternalReadContext::Create(std::move(read_context), table_schema,
+                                                     table_schema->Options()));
+    auto remapped_compound =
+        std::dynamic_pointer_cast<CompoundPredicate>(internal_context->GetPredicate());
+    ASSERT_NE(remapped_compound, nullptr);
+    ASSERT_EQ(remapped_compound->Children().size(), 2);
+    auto remapped_left = std::dynamic_pointer_cast<LeafPredicate>(remapped_compound->Children()[0]);
+    auto remapped_right =
+        std::dynamic_pointer_cast<LeafPredicate>(remapped_compound->Children()[1]);
+    ASSERT_NE(remapped_left, nullptr);
+    ASSERT_NE(remapped_right, nullptr);
+    EXPECT_EQ(remapped_left->FieldName(), "f3");
+    EXPECT_EQ(remapped_left->FieldIndex(), 0);
+    EXPECT_EQ(remapped_right->FieldName(), "f0");
+    EXPECT_EQ(remapped_right->FieldIndex(), 1);
+}
+
+// Field not present in the projected read schema must surface as an Invalid status.
+TEST(InternalReadContext, TestPredicateOnFieldMissingFromReadSchema) {
+    std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
+    ReadContextBuilder context_builder(path);
+    context_builder.SetReadSchema({"f3", "f0"});
+    auto predicate = PredicateBuilder::Equal(/*field_index=*/1, "f1", FieldType::INT, Literal(7));
+    context_builder.SetPredicate(predicate);
+    ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
+    SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), read_context->GetPath());
+    ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+    ASSERT_NOK_WITH_MSG(
+        InternalReadContext::Create(std::move(read_context), table_schema, table_schema->Options()),
+        "field f1 does not exist in schema");
+}
+
 }  // namespace paimon::test

src/paimon/core/table/source/table_read_test.cpp

@@ -67,18 +67,6 @@ TEST(TableReadTest, TestReadWithInvalidContext) {
         ASSERT_NOK_WITH_MSG(TableRead::Create(std::move(read_context)),
                             "schema type double mismatches predicate field type BIGINT");
     }
-    {
-        // field idx in predicate mismatch in schema
-        auto predicate = PredicateBuilder::Equal(/*field_index=*/2, /*field_name=*/"f3",
-                                                 FieldType::DOUBLE, Literal(15.0));
-        ReadContextBuilder context_builder(path);
-        context_builder.SetReadSchema({"f3", "f0", "f1"});
-        context_builder.SetPredicate(predicate);
-        ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
-        ASSERT_NOK_WITH_MSG(
-            TableRead::Create(std::move(read_context)),
-            "field f3 has field idx 0 in input schema, mismatch field idx 2 in predicate");
-    }
     {
         // literal cannot be null
         auto predicate = PredicateBuilder::Equal(/*field_index=*/3, /*field_name=*/"f3",