[CALCITE-7618] Add filter pushdown support to the file adapter's CSV table implementation

example/csv/src/test/java/org/apache/calcite/test/CsvTest.java

@@ -383,6 +383,22 @@ void testPushDownProjectAggregateNested(String format) {
         .ok();
   }
 
+  @Test void testFilterableWhereAge() {
+    // age column has nulls in the data — make sure they're excluded under objectsEqual
+    final String sql = "select name from EMPS where age = 25";
+    sql("filterable-model", sql)
+        .returns("NAME=Fred")
+        .ok();
+  }
+
+  @Test void testFilterableWhereSlacker() {
+    // slacker column has nulls in the data — make sure they're excluded under objectsEqual
+    final String sql = "select name from EMPS where slacker = false";
+    sql("filterable-model", sql)
+        .returns("NAME=John", "NAME=Alice")
+        .ok();
+  }
+
   /** Test case for
    * <a href="https://issues.apache.org/jira/browse/CALCITE-2272">[CALCITE-2272]
    * Incorrect result for {@code name like '%E%' and city not like '%W%'}</a>.

file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java

@@ -139,7 +139,7 @@
     }
   }
 
-  private static RowConverter<?> converter(List<RelDataType> fieldTypes,
+  static RowConverter<?> converter(List<RelDataType> fieldTypes,
       List<Integer> fields) {
     if (fields.size() == 1) {
       final int field = fields.get(0);
@@ -254,6 +254,30 @@
     return new CSVReader(source.reader(), separator);
   }
 
+  /**
+   * Evaluates equality between 2 Comparable objects, conforming to SQL WHERE filter '=' semantics.
+   *
+   * <p>Returns {@code false} if either operand is null. Because of this, it cannot be
+   * directly used for {@code IS NOT DISTINCT FROM} comparisons without additional null handling.
+   *
+   * <p>For Comparable objects of the same class (like BigDecimal), it utilizes
+   * {@code compareTo()} to ignore differences in representation (e.g. scale)
+   * that would cause standard {@code equals()} to fail.
+   */
+  @SuppressWarnings("unchecked")
+  static boolean sameValue(@Nullable Comparable o1, @Nullable Comparable o2) {
+    if (o1 == null || o2 == null) {
+      return false;
+    }
+    if (o1 == o2) {
+      return true;
+    }
+    if (o1.getClass().isInstance(o2)) {
+      return o1.compareTo(o2) == 0;
+    }
+    return o1.equals(o2);
+  }
+
   @Override public E current() {
     return castNonNull(current);
   }
@@ -284,7 +308,7 @@
           return false;
         }
         if (filterValues != null) {
-          for (int i = 0; i < strings.length; i++) {
+          for (int i = 0; i < filterValues.size(); i++) {
             String filterValue = filterValues.get(i);
             if (filterValue != null) {
               if (!filterValue.equals(strings[i])) {
@@ -329,6 +353,11 @@
     return typeFactory.createTypeWithNullability(typeFactory.createSqlType(sqlTypeName), true);
   }
 
+  /** Returns a field from a CSV row, or null if the row is too short. */
+  private static @Nullable String field(String[] strings, int index) {
+    return index < strings.length ? strings[index] : null;
+  }
+
   /** Row converter.
    *
    * @param <E> element type */
@@ -480,7 +509,7 @@
       final @Nullable Object[] objects = new Object[fields.size()];
       for (int i = 0; i < fields.size(); i++) {
         int field = fields.get(i);
-        objects[i] = convert(fieldTypes.get(field), strings[field]);
+        objects[i] = convert(fieldTypes.get(field), field(strings, field));
       }
       return objects;
     }
@@ -490,7 +519,7 @@
       objects[0] = System.currentTimeMillis();
       for (int i = 0; i < fields.size(); i++) {
         int field = fields.get(i);
-        objects[i + 1] = convert(fieldTypes.get(field), strings[field]);
+        objects[i + 1] = convert(fieldTypes.get(field), field(strings, field));
       }
       return objects;
     }
@@ -507,7 +536,7 @@
     }
 
     @Override public @Nullable Object convertRow(@Nullable String[] strings) {
-      return convert(fieldType, strings[fieldIndex]);
+      return convert(fieldType, field(strings, fieldIndex));
     }
   }
 }

file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.adapter.file;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+
+import org.immutables.value.Value;
+
+/**
+ * Planner rule that pushes filter predicates into a
+ * {@link CsvTableScan}.
+ *
+ * <p>Any predicate expressible as a {@link org.apache.calcite.rex.RexNode}
+ * (including AND, OR, NOT, IS NULL, comparisons, LIKE, etc.) can be pushed
+ * down. The condition is compiled at plan time via
+ * {@link org.apache.calcite.adapter.enumerable.RexToLixTranslator} into a
+ * Java {@link org.apache.calcite.linq4j.function.Predicate1} and applied
+ * directly on the enumerable produced by the scan, so no rows that fail the
+ * predicate are ever materialised.
+ *
+ * @see FileRules#FILTER_SCAN
+ */
+@Value.Enclosing
+public class CsvFilterTableScanRule
+    extends RelRule<CsvFilterTableScanRule.Config> {
+
+  /** Creates a CsvFilterTableScanRule. */
+  protected CsvFilterTableScanRule(Config config) {
+    super(config);
+  }
+
+  @Override public void onMatch(RelOptRuleCall call) {
+    final LogicalFilter filter = call.rel(0);
+    final CsvTableScan scan = call.rel(1);
+
+    // Compose a conjunction of the existing condition and the new one.
+    final RexNode newCondition;
+    if (scan.condition == null) {
+      newCondition = filter.getCondition();
+    } else {
+      newCondition =
+          RexUtil.composeConjunction(scan.getCluster().getRexBuilder(),
+          java.util.Arrays.asList(scan.condition, filter.getCondition()));
+    }
+
+    // Build a new scan that carries the pushed-down filter condition.
+    final CsvTableScan newScan =
+        new CsvTableScan(scan.getCluster(), scan.getTable(), scan.csvTable,
+        scan.fields, newCondition);
+
+    call.transformTo(newScan);
+  }
+
+  /** Rule configuration. */
+  @Value.Immutable(singleton = false)
+  public interface Config extends RelRule.Config {
+    Config DEFAULT = ImmutableCsvFilterTableScanRule.Config.builder()
+        .withOperandSupplier(b0 ->
+            b0.operand(LogicalFilter.class).oneInput(b1 ->
+                b1.operand(CsvTableScan.class).noInputs()))
+        .build();
+
+    @Override default CsvFilterTableScanRule toRule() {
+      return new CsvFilterTableScanRule(this);
+    }
+  }
+}

file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java

@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.adapter.file;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+
+import org.immutables.value.Value;
+
+import java.util.List;
+
+/**
+ * Planner rule that matches a {@link LogicalProject} on a {@link LogicalFilter}
+ * on a {@link CsvTableScan}, and pushes filter predicates into the scan.
+ *
+ * @see FileRules#PROJECT_FILTER_SCAN
+ */
+@Value.Enclosing
+public class CsvProjectFilterTableScanRule
+    extends RelRule<CsvProjectFilterTableScanRule.Config> {
+
+  /** Creates a CsvProjectFilterTableScanRule. */
+  protected CsvProjectFilterTableScanRule(Config config) {
+    super(config);
+  }
+
+  @Override public void onMatch(RelOptRuleCall call) {
+    final LogicalProject project = call.rel(0);
+    final LogicalFilter filter = call.rel(1);
+    final CsvTableScan scan = call.rel(2);
+
+    // Find all input fields referenced by the project expressions
+    final java.util.Set<Integer> projectInputFields = new java.util.HashSet<>();
+    for (RexNode proj : project.getProjects()) {
+      proj.accept(new org.apache.calcite.rex.RexVisitorImpl<Void>(true) {
+        @Override public Void visitInputRef(RexInputRef inputRef) {
+          projectInputFields.add(inputRef.getIndex());
+          return null;
+        }
+      });
+    }
+
+    // Find all input fields referenced by the filter condition
+    final java.util.Set<Integer> filterInputFields = new java.util.HashSet<>();
+    filter.getCondition().accept(new org.apache.calcite.rex.RexVisitorImpl<Void>(true) {
+      @Override public Void visitInputRef(RexInputRef inputRef) {
+        filterInputFields.add(inputRef.getIndex());
+        return null;
+      }
+    });
+
+    // Union the projected/referenced indices
+    final java.util.Set<Integer> neededProjectedIndices = new java.util.TreeSet<>();
+    neededProjectedIndices.addAll(projectInputFields);
+    neededProjectedIndices.addAll(filterInputFields);
+
+    // Map needed scan projected indices to full-table indices
+    final int[] newFields = new int[neededProjectedIndices.size()];
+    int k = 0;
+    for (int idx : neededProjectedIndices) {
+      newFields[k++] = scan.fields[idx];
+    }
+
+    // Build index map from old projected index to new index in newFields
+    final java.util.Map<Integer, Integer> indexMap = new java.util.HashMap<>();
+    int newIdx = 0;
+    for (int idx : neededProjectedIndices) {
+      indexMap.put(idx, newIdx++);
+    }
+
+    // Create shuttle to map RexInputRef indices
+    final org.apache.calcite.rex.RexShuttle shuttle = new org.apache.calcite.rex.RexShuttle() {
+      @Override public RexNode visitInputRef(RexInputRef inputRef) {
+        final Integer mapped = indexMap.get(inputRef.getIndex());
+        if (mapped == null) {
+          return inputRef;
+        }
+        return scan.getCluster().getRexBuilder().makeInputRef(inputRef.getType(), mapped);
+      }
+    };
+
+    final RexNode mappedCondition = filter.getCondition().accept(shuttle);
+    final List<RexNode> mappedProjects = new java.util.ArrayList<>();
+    for (RexNode proj : project.getProjects()) {
+      mappedProjects.add(proj.accept(shuttle));
+    }
+
+    final RexNode finalCondition;
+    if (scan.condition == null) {
+      finalCondition = mappedCondition;
+    } else {
+      finalCondition =
+          RexUtil.composeConjunction(scan.getCluster().getRexBuilder(),
+          java.util.Arrays.asList(scan.condition.accept(shuttle), mappedCondition));
+    }
+
+    final CsvTableScan newScan =
+        new CsvTableScan(scan.getCluster(), scan.getTable(), scan.csvTable,
+        newFields, finalCondition);
+
+    final RelNode result =
+        project.copy(project.getTraitSet(), newScan, mappedProjects, project.getRowType());
+
+    call.transformTo(result);
+  }
+
+  /** Rule configuration. */
+  @Value.Immutable(singleton = false)
+  public interface Config extends RelRule.Config {
+    Config DEFAULT = ImmutableCsvProjectFilterTableScanRule.Config.builder()
+        .withOperandSupplier(b0 ->
+            b0.operand(LogicalProject.class).oneInput(b1 ->
+                b1.operand(LogicalFilter.class).oneInput(b2 ->
+                    b2.operand(CsvTableScan.class).noInputs())))
+        .build();
+
+    @Override default CsvProjectFilterTableScanRule toRule() {
+      return new CsvProjectFilterTableScanRule(this);
+    }
+  }
+}

file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java

@@ -45,17 +45,30 @@ protected CsvProjectTableScanRule(Config config) {
   @Override public void onMatch(RelOptRuleCall call) {
     final LogicalProject project = call.rel(0);
     final CsvTableScan scan = call.rel(1);
-    int[] fields = getProjectFields(project.getProjects());
-    if (fields == null) {
+    int[] projectFieldIndices = getProjectFields(project.getProjects());
+    if (projectFieldIndices == null) {
       // Project contains expressions more complex than just field references.
       return;
     }
+    if (scan.condition != null) {
+      // If the scan already has a condition, we cannot push the project down
+      // because the condition references the scan's current row type.
+      return;
+    }
+    // The project field indices are into the scan's *current* row type (which
+    // may already be a subset of the full table due to a prior projection).
+    // Map through scan.fields to get the original full-table column indices.
+    final int[] newFields = new int[projectFieldIndices.length];
+    for (int i = 0; i < projectFieldIndices.length; i++) {
+      newFields[i] = scan.fields[projectFieldIndices[i]];
+    }
     call.transformTo(
         new CsvTableScan(
             scan.getCluster(),
             scan.getTable(),
             scan.csvTable,
-            fields));
+            newFields,
+            scan.condition));
   }
 
   private static int[] getProjectFields(List<RexNode> exps) {