refractor(parquet): simplify FileReaderWrapper and PageFilteredRowGroupReader with unified TargetRowGroup structure

src/paimon/format/parquet/column_index_filter.cpp

@@ -96,6 +96,10 @@ Result<RowRanges> ColumnIndexFilter::VisitLeafPredicate(
     const auto& literals = leaf_predicate->Literals();
     FieldType field_type = leaf_predicate->GetFieldType();
 
+    if (function_type != Function::Type::IS_NULL && function_type != Function::Type::IS_NOT_NULL &&
+        literals.empty()) {
+        return RowRanges::CreateSingle(row_group_row_count);
+    }
     std::vector<int32_t> matching_pages;
 
     switch (function_type) {
@@ -106,37 +110,22 @@ Result<RowRanges> ColumnIndexFilter::VisitLeafPredicate(
             matching_pages = FilterPagesByIsNotNull(column_index_ptr);
             break;
         case Function::Type::EQUAL:
-            if (!literals.empty()) {
-                matching_pages = FilterPagesByEqual(column_index_ptr, literals[0], field_type);
-            }
+            matching_pages = FilterPagesByEqual(column_index_ptr, literals[0], field_type);
             break;
         case Function::Type::NOT_EQUAL:
-            if (!literals.empty()) {
-                matching_pages = FilterPagesByNotEqual(column_index_ptr, literals[0], field_type);
-            }
+            matching_pages = FilterPagesByNotEqual(column_index_ptr, literals[0], field_type);
             break;
         case Function::Type::LESS_THAN:
-            if (!literals.empty()) {
-                matching_pages = FilterPagesByLessThan(column_index_ptr, literals[0], field_type);
-            }
+            matching_pages = FilterPagesByLessThan(column_index_ptr, literals[0], field_type);
             break;
         case Function::Type::LESS_OR_EQUAL:
-            if (!literals.empty()) {
-                matching_pages =
-                    FilterPagesByLessOrEqual(column_index_ptr, literals[0], field_type);
-            }
+            matching_pages = FilterPagesByLessOrEqual(column_index_ptr, literals[0], field_type);
             break;
         case Function::Type::GREATER_THAN:
-            if (!literals.empty()) {
-                matching_pages =
-                    FilterPagesByGreaterThan(column_index_ptr, literals[0], field_type);
-            }
+            matching_pages = FilterPagesByGreaterThan(column_index_ptr, literals[0], field_type);
             break;
         case Function::Type::GREATER_OR_EQUAL:
-            if (!literals.empty()) {
-                matching_pages =
-                    FilterPagesByGreaterOrEqual(column_index_ptr, literals[0], field_type);
-            }
+            matching_pages = FilterPagesByGreaterOrEqual(column_index_ptr, literals[0], field_type);
             break;
         case Function::Type::IN:
             matching_pages = FilterPagesByIn(column_index_ptr, literals, field_type);

src/paimon/format/parquet/column_index_filter_test.cpp

@@ -26,6 +26,9 @@
 #include "arrow/c/abi.h"
 #include "arrow/c/bridge.h"
 #include "gtest/gtest.h"
+#include "paimon/common/predicate/equal.h"
+#include "paimon/common/predicate/in.h"
+#include "paimon/common/predicate/leaf_predicate_impl.h"
 #include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
 #include "paimon/common/utils/arrow/mem_utils.h"
 #include "paimon/defs.h"
@@ -480,4 +483,29 @@ TEST_F(ColumnIndexFilterTest, NullPredicateReturnsAllRows) {
     EXPECT_EQ(row_group_row_count_, ranges.RowCount());
 }
 
+/// When literals is empty for comparison predicates (EQUAL, NOT_EQUAL, LESS_THAN,
+/// LESS_OR_EQUAL, GREATER_THAN, GREATER_OR_EQUAL), the filter should return all
+/// rows (conservative fallback) rather than returning empty ranges.
+TEST_F(ColumnIndexFilterTest, EmptyLiteralsReturnsAllRows) {
+    // Construct a LeafPredicate with EQUAL function but empty literals vector.
+    // This simulates the edge case where literals are unexpectedly empty.
+    auto pred = std::make_shared<paimon::LeafPredicateImpl>(paimon::Equal::Instance(), 0, "val",
+                                                            FieldType::INT, std::vector<Literal>());
+    ASSERT_OK_AND_ASSIGN(auto ranges, Filter(pred));
+    // With empty literals, the filter cannot evaluate the comparison,
+    // so it should conservatively return all rows.
+    EXPECT_EQ(row_group_row_count_, ranges.RowCount());
+}
+
+/// Empty literals for IN predicate — the early guard in VisitLeafPredicate treats
+/// all non-IS_NULL/IS_NOT_NULL predicates with empty literals conservatively,
+/// returning all rows rather than risking incorrect filtering.
+TEST_F(ColumnIndexFilterTest, EmptyLiteralsInReturnsAllRows) {
+    auto pred = std::make_shared<paimon::LeafPredicateImpl>(paimon::In::Instance(), 0, "val",
+                                                            FieldType::INT, std::vector<Literal>());
+    ASSERT_OK_AND_ASSIGN(auto ranges, Filter(pred));
+    // Empty literals → conservative fallback → all rows.
+    EXPECT_EQ(row_group_row_count_, ranges.RowCount());
+}
+
 }  // namespace paimon::parquet::test