Task apache#27: Add IN predicate test (apache#64)

- Added InPredicate test with 15 comprehensive test cases
- Added arrow/compute/api_scalar.h include for SetLookupOptions
- Tests IN predicate optimization with min/max range intersection
- Tests single stripe matches (values in one stripe only)
- Tests multi-stripe matches (values spanning multiple stripes)
- Tests out-of-range optimization (all values outside min/max)
- Tests negative values and boundary values
- Tests single value IN (equivalent to equality)
- Tests adjacent and non-adjacent stripe combinations
- Tests IN combined with AND and OR operators
- Tests large IN lists with many values
- Tests empty IN list (no matches)
- Tests duplicate values in IN list
- Tests int64 type support
- Uses 5-stripe test file (500 rows) with distinct value ranges

Verified: All 15 test cases correctly leverage stripe statistics
to skip stripes when no IN values fall within [min, max] range.

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: cbb330

cpp/src/arrow/dataset/file_orc_test.cc

@@ -22,6 +22,7 @@
 
 #include "arrow/adapters/orc/adapter.h"
 #include "arrow/array/builder_primitive.h"
+#include "arrow/compute/api_scalar.h"
 #include "arrow/dataset/dataset_internal.h"
 #include "arrow/dataset/discovery.h"
 #include "arrow/dataset/file_base.h"
@@ -1046,5 +1047,136 @@ TEST_P(TestOrcFileFormatScan, CompoundPredicates) {
   ASSERT_EQ(count_rows(fragment), 150);  // 50 + 50 + 50
 }
 
+TEST_P(TestOrcFileFormatScan, InPredicate) {
+  // Test IN predicate optimization with stripe statistics.
+  // IN predicate (x IN (v1, v2, v3)) can skip stripes by checking if any
+  // values in the IN list fall within the stripe's [min, max] range.
+  //
+  // Range intersection optimization:
+  // - If no IN values are within [min, max], the stripe can be skipped
+  // - If some IN values are within [min, max], the stripe must be read
+
+  constexpr int kNumStripes = 5;
+  constexpr int kRowsPerStripe = 100;
+
+  auto schema = arrow::schema({field("x", int32())});
+  ASSERT_OK_AND_ASSIGN(auto buffer,
+                       OrcTestFileGenerator::MakeMultiStripeFile(schema, kNumStripes, kRowsPerStripe));
+
+  auto source = std::make_shared<io::BufferReader>(buffer);
+  SetSchema(schema->fields());
+  ASSERT_OK_AND_ASSIGN(auto fragment,
+                       format_->MakeFragment(FileSource(source), literal(true)));
+
+  // Helper to count rows
+  auto count_rows = [&](std::shared_ptr<Fragment> frag) {
+    int64_t total = 0;
+    for (auto maybe_batch : Batches(frag)) {
+      ASSERT_OK_AND_ASSIGN(auto batch, maybe_batch);
+      total += batch->num_rows();
+    }
+    return total;
+  };
+
+  // Test 1: IN with values in a single stripe
+  // x IN (150, 151, 152) - all values in stripe 1 [100-199]
+  auto set1 = ArrayFromJSON(int32(), "[150, 151, 152]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set1}));
+  ASSERT_EQ(count_rows(fragment), 3);  // Only stripe 1 read
+
+  // Test 2: IN with values spanning multiple stripes
+  // x IN (50, 150, 250, 350, 450) - one value per stripe
+  auto set2 = ArrayFromJSON(int32(), "[50, 150, 250, 350, 450]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set2}));
+  ASSERT_EQ(count_rows(fragment), 5);  // All stripes read (one match each)
+
+  // Test 3: IN with values completely outside all stripe ranges
+  // x IN (1000, 2000, 3000) - all values > 499
+  auto set3 = ArrayFromJSON(int32(), "[1000, 2000, 3000]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set3}));
+  ASSERT_EQ(count_rows(fragment), 0);  // All stripes skipped
+
+  // Test 4: IN with negative values (below all stripe ranges)
+  // x IN (-100, -50, -1) - all values < 0
+  auto set4 = ArrayFromJSON(int32(), "[-100, -50, -1]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set4}));
+  ASSERT_EQ(count_rows(fragment), 0);  // All stripes skipped
+
+  // Test 5: IN with mixed in-range and out-of-range values
+  // x IN (5, 15, 1000, 2000) - only 5 and 15 are in range (stripe 0)
+  auto set5 = ArrayFromJSON(int32(), "[5, 15, 1000, 2000]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set5}));
+  ASSERT_EQ(count_rows(fragment), 2);  // Only stripe 0 read
+
+  // Test 6: IN with boundary values (stripe boundaries)
+  // x IN (0, 99, 100, 199) - values at stripe 0/1 boundaries
+  auto set6 = ArrayFromJSON(int32(), "[0, 99, 100, 199]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set6}));
+  ASSERT_EQ(count_rows(fragment), 4);  // Stripes 0 and 1 read
+
+  // Test 7: IN with single value (equivalent to equality)
+  // x IN (250) should behave like x = 250
+  auto set7 = ArrayFromJSON(int32(), "[250]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set7}));
+  ASSERT_EQ(count_rows(fragment), 1);  // Only stripe 2 read
+
+  // Test 8: IN with values in two adjacent stripes
+  // x IN (180, 190, 210, 220) - values in stripes 1 and 2
+  auto set8 = ArrayFromJSON(int32(), "[180, 190, 210, 220]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set8}));
+  ASSERT_EQ(count_rows(fragment), 4);  // Stripes 1 and 2 read
+
+  // Test 9: IN with values in non-adjacent stripes
+  // x IN (50, 250, 450) - stripes 0, 2, 4
+  auto set9 = ArrayFromJSON(int32(), "[50, 250, 450]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set9}));
+  ASSERT_EQ(count_rows(fragment), 3);  // Stripes 0, 2, 4 read
+
+  // Test 10: IN combined with AND
+  // (x IN (50, 150, 250)) AND (x < 200) - only 50 and 150 satisfy both
+  auto set10 = ArrayFromJSON(int32(), "[50, 150, 250]");
+  SetFilter(and_(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set10}),
+                 less(field_ref("x"), literal<int32_t>(200))));
+  ASSERT_EQ(count_rows(fragment), 2);  // Stripes 0 and 1 read
+
+  // Test 11: IN combined with OR
+  // (x IN (5, 15)) OR (x >= 400) - stripe 0 and stripe 4
+  auto set11 = ArrayFromJSON(int32(), "[5, 15]");
+  SetFilter(or_(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set11}),
+                greater_equal(field_ref("x"), literal<int32_t>(400))));
+  ASSERT_EQ(count_rows(fragment), 102);  // 2 from stripe 0, 100 from stripe 4
+
+  // Test 12: Large IN list with values across all stripes
+  // x IN (10, 20, 30, ..., 490) - values from all stripes
+  auto set12 = ArrayFromJSON(int32(), "[10, 20, 30, 110, 120, 130, 210, 220, 230, 310, 320, 330, 410, 420, 430]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set12}));
+  ASSERT_EQ(count_rows(fragment), 15);  // All stripes read (3 matches each)
+
+  // Test 13: Empty IN list - no values to match
+  auto set13 = ArrayFromJSON(int32(), "[]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set13}));
+  ASSERT_EQ(count_rows(fragment), 0);  // All stripes skipped
+
+  // Test 14: IN with duplicate values (should behave like unique set)
+  // x IN (150, 150, 150) should find x = 150 once
+  auto set14 = ArrayFromJSON(int32(), "[150, 150, 150]");
+  SetFilter(call("is_in", {field_ref("x")}, compute::SetLookupOptions{set14}));
+  ASSERT_EQ(count_rows(fragment), 1);  // Only stripe 1 read
+
+  // Test 15: Int64 type - verify IN works with int64
+  auto schema_i64 = arrow::schema({field("y", int64())});
+  ASSERT_OK_AND_ASSIGN(auto buffer_i64,
+                       OrcTestFileGenerator::MakeMultiStripeFile(schema_i64, kNumStripes, kRowsPerStripe));
+
+  auto source_i64 = std::make_shared<io::BufferReader>(buffer_i64);
+  SetSchema(schema_i64->fields());
+  ASSERT_OK_AND_ASSIGN(auto fragment_i64,
+                       format_->MakeFragment(FileSource(source_i64), literal(true)));
+
+  auto set15 = ArrayFromJSON(int64(), "[50, 150, 250]");
+  SetFilter(call("is_in", {field_ref("y")}, compute::SetLookupOptions{set15}));
+  ASSERT_EQ(count_rows(fragment_i64), 3);  // Stripes 0, 1, 2 read
+}
+
 }  // namespace dataset
 }  // namespace arrow