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

duanyyyyyyy · duanyyyyyyy · commit e5c7e75213be · 2026-06-04T12:01:06.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/src/paimon/core/operation/internal_read_context.cpp b/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
diff --git a/src/paimon/core/operation/internal_read_context.h b/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
diff --git a/src/paimon/core/operation/internal_read_context_test.cpp b/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
