mihaibudiu commented on code in PR #5048:
URL: https://github.com/apache/calcite/pull/5048#discussion_r3555006246
##########
file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java:
##########
@@ -254,6 +254,30 @@ static CSVReader openCsv(Source source, char separator)
throws IOException {
return new CSVReader(source.reader(), separator);
}
+ /**
+ * Evaluates equality between two 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 objectsEqual(@Nullable Object o1, @Nullable Object o2) {
Review Comment:
My suggestion was to implement `sameValue(Comparable left, Comparable
right)`, this would be more general.
##########
file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java:
##########
@@ -84,21 +97,19 @@ protected CsvTableScan(RelOptCluster cluster, RelOptTable
table,
@Override public void register(RelOptPlanner planner) {
planner.addRule(FileRules.PROJECT_SCAN);
+ planner.addRule(FileRules.FILTER_SCAN);
+ planner.addRule(FileRules.PROJECT_FILTER_SCAN);
}
@Override public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner,
RelMetadataQuery mq) {
- // Multiply the cost by a factor that makes a scan more attractive if it
- // has significantly fewer fields than the original scan.
- //
- // The "+ 2D" on top and bottom keeps the function fairly smooth.
- //
- // For example, if table has 3 fields, project has 1 field,
- // then factor = (1 + 2) / (3 + 2) = 0.6
final RelOptCost cost = requireNonNull(super.computeSelfCost(planner, mq));
- return cost
- .multiplyBy(((double) fields.length + 2D)
- / ((double) table.getRowType().getFieldCount() + 2D));
+ double factor = ((double) fields.length + 2D)
Review Comment:
the comment deleted seemed useful, maybe some of it can be preserved and
updated
##########
file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java:
##########
@@ -417,6 +418,173 @@ private static void checkEmpty(ResultSet resultSet) {
sql("model-with-custom-table", sql).ok();
}
+ /** Test case for
+ * <a
href="https://issues.apache.org/jira/browse/CALCITE-7618";>[CALCITE-7618]
+ * Add filter pushdown support to file adapter's CSV implementation</a>.
+ *
+ * <p>Verifies that a simple equality filter is pushed into {@link
CsvTableScan},
+ * eliminating the {@code EnumerableCalc} that would otherwise evaluate it.
*/
+ @Test void testFilterPushDown() {
+ final String sql = "explain plan for select * from EMPS where deptno = 20";
+ final String expected = "PLAN=CsvTableScan(table=[[SALES, EMPS]], "
+ + "fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]], condition=[=($2, 20)])\n";
+ sql("smart", sql).returns(expected).ok();
+ }
+
+ @Test void testFilterPushDownWithProject() {
+ final String sql = "explain plan for select name, empno from EMPS where
deptno = 20";
+ final String expected = "PLAN=EnumerableCalc(expr#0..2=[{inputs}],"
Review Comment:
So here the filter was not pushed down?
Or is this displaying the dynamically-generated plan which synthesizes a
calc?
##########
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
+ */
[email protected]
+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];
Review Comment:
Is scan.fields always present?
##########
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
+ */
[email protected]
+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 ->
Review Comment:
Is the reverse configuration supported, where the filter happens after the
project?
Maybe it happens as a composition of two rules?
##########
file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java:
##########
@@ -417,6 +418,173 @@ private static void checkEmpty(ResultSet resultSet) {
sql("model-with-custom-table", sql).ok();
}
+ /** Test case for
+ * <a
href="https://issues.apache.org/jira/browse/CALCITE-7618";>[CALCITE-7618]
+ * Add filter pushdown support to file adapter's CSV implementation</a>.
+ *
+ * <p>Verifies that a simple equality filter is pushed into {@link
CsvTableScan},
+ * eliminating the {@code EnumerableCalc} that would otherwise evaluate it.
*/
+ @Test void testFilterPushDown() {
+ final String sql = "explain plan for select * from EMPS where deptno = 20";
+ final String expected = "PLAN=CsvTableScan(table=[[SALES, EMPS]], "
+ + "fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]], condition=[=($2, 20)])\n";
+ sql("smart", sql).returns(expected).ok();
+ }
+
+ @Test void testFilterPushDownWithProject() {
+ final String sql = "explain plan for select name, empno from EMPS where
deptno = 20";
+ final String expected = "PLAN=EnumerableCalc(expr#0..2=[{inputs}],"
+ + " expr#3=[20], expr#4=[=($t2, $t3)], NAME=[$t1], EMPNO=[$t0],
$condition=[$t4])\n"
+ + " CsvTableScan(table=[[SALES, EMPS]], fields=[[0, 1, 2]])\n";
+ sql("smart", sql).returns(expected).ok();
+ }
+
+ /** Test case for
+ * <a
href="https://issues.apache.org/jira/browse/CALCITE-7618";>[CALCITE-7618]
+ * Add filter pushdown support to file adapter's CSV implementation</a>.
+ *
+ * <p>Verifies that filter pushdown returns correct query results. */
+ @Test void testFilterPushDownResult() {
+ final String sql = "select name, empno from EMPS where deptno = 20";
+ sql("smart", sql)
+ .returns("NAME=Eric; EMPNO=110",
+ "NAME=Wilma; EMPNO=120")
+ .ok();
+ }
+
+ /** Test case for
+ * <a
href="https://issues.apache.org/jira/browse/CALCITE-7618";>[CALCITE-7618]
+ * Add filter pushdown support to file adapter's CSV implementation</a>.
+ *
+ * <p>Verifies that range filters are evaluated correctly under the new
compiled-filter
+ * pushdown mechanism. */
+ @Test void testRangeFilterPushDown() {
+ // empno > 110 is a range filter; the compiler-based pushdown handles it
+ // like any other predicate, pushing it into the scan via EnumerableCalc.
+ final String sql = "select name from EMPS where empno > 110";
+ sql("smart", sql)
+ .returns("NAME=Wilma",
+ "NAME=Alice")
+ .ok();
+ }
+
+ @Test void testFilterOnNullValues() {
+ final String sql = "select name, age from long_emps where age is null";
+ sql("bug", sql)
+ .returns("NAME=John; AGE=null",
+ "NAME=Alice; AGE=null")
+ .ok();
+ }
+
+ @Test void testFilterPushDownLong() {
+ final String sql = "select name from long_emps where empno = 130";
+ sql("bug", sql)
+ .returns("NAME=Alice")
+ .ok();
+ }
+
+ @Test void testFilterPushDownBoolean() {
+ final String sql = "select name from long_emps where slacker = true";
+ sql("bug", sql)
+ .returns("NAME=Fred")
+ .ok();
+ }
+
+ @Test void testFilterPushDownString() {
+ final String sql = "select empno from long_emps where gender = 'F'";
+ sql("bug", sql)
+ .returns("EMPNO=120", "EMPNO=130")
+ .ok();
+ }
+
+ @Test void testFilterPushDownDecimal() {
+ final String sql = "select deptno from sales.\"DECIMAL\" where budget =
100.01";
+ sql("sales-csv", sql)
+ .returns("DEPTNO=20")
+ .ok();
+ }
+
+ @Test void testFilterPushDownDate() {
+ final String sql = "select name from long_emps where joinedat = DATE
'2001-01-01'";
+ sql("bug", sql)
+ .returns("NAME=Eric")
+ .ok();
+ }
+
+ @Test void testFilterPushDownTime() {
+ final String sql = "select empno from \"DATE\" where jointime = TIME
'07:15:56'";
+ sql("bug", sql)
+ .returns("EMPNO=140")
+ .ok();
+ }
+
+ @Test void testFilterPushDownTimestamp() {
Review Comment:
We have all these tests, but we are not sure whether the pushdown worked for
them, only that the result is correct. maybe you can add in the same function
for each of them the plan?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
