From 3fd66a7f36557fc5819df137608684b18fa7e2c6 Mon Sep 17 00:00:00 2001 From: Diveyam Mishra Date: Tue, 23 Jun 2026 00:33:23 +0530 Subject: [PATCH 1/3] [CALCITE-7618] Add filter pushdown support to the file adapter's CSV table implementation --- .../calcite/adapter/file/CsvEnumerator.java | 47 +++- .../adapter/file/CsvFilterTableScanRule.java | 217 ++++++++++++++++++ .../file/CsvProjectFilterTableScanRule.java | 169 ++++++++++++++ .../adapter/file/CsvProjectTableScanRule.java | 14 +- .../calcite/adapter/file/CsvTableScan.java | 69 +++++- .../adapter/file/CsvTranslatableTable.java | 32 +++ .../calcite/adapter/file/FileRules.java | 11 + .../adapter/file/CsvEnumeratorTest.java | 14 ++ .../calcite/adapter/file/FileAdapterTest.java | 69 +++++- 9 files changed, 611 insertions(+), 31 deletions(-) create mode 100644 file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java create mode 100644 file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java index f62433beab47..0261a5f59b50 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java @@ -71,6 +71,7 @@ public class CsvEnumerator implements Enumerator { private final CSVReader reader; private final @Nullable List<@Nullable String> filterValues; + private final @Nullable List fieldTypes; private final AtomicBoolean cancelFlag; private final RowConverter rowConverter; private @Nullable E current; @@ -115,18 +116,25 @@ private static void clearTimeFormats() { public CsvEnumerator(Source source, AtomicBoolean cancelFlag, List fieldTypes, List fields, char separator) { //noinspection unchecked - this(source, cancelFlag, false, null, + this(source, cancelFlag, false, null, fieldTypes, (RowConverter) converter(fieldTypes, fields), separator); } public CsvEnumerator(Source source, AtomicBoolean cancelFlag, boolean stream, @Nullable String @Nullable [] filterValues, RowConverter rowConverter, char separator) { + this(source, cancelFlag, stream, filterValues, null, rowConverter, separator); + } + + public CsvEnumerator(Source source, AtomicBoolean cancelFlag, boolean stream, + @Nullable String @Nullable [] filterValues, @Nullable List fieldTypes, + RowConverter rowConverter, char separator) { this.cancelFlag = cancelFlag; this.rowConverter = rowConverter; this.filterValues = filterValues == null ? null : ImmutableNullableList.copyOf(filterValues); + this.fieldTypes = fieldTypes; try { if (stream) { this.reader = new CsvStreamReader(source, separator); @@ -139,7 +147,7 @@ public CsvEnumerator(Source source, AtomicBoolean cancelFlag, boolean stream, } } - private static RowConverter converter(List fieldTypes, + static RowConverter converter(List fieldTypes, List fields) { if (fields.size() == 1) { final int field = fields.get(0); @@ -254,6 +262,19 @@ static CSVReader openCsv(Source source, char separator) throws IOException { return new CSVReader(source.reader(), separator); } + private static boolean objectsEqual(@Nullable Object o1, @Nullable Object o2) { + if (o1 == o2) { + return true; + } + if (o1 == null || o2 == null) { + return false; + } + if (o1 instanceof BigDecimal && o2 instanceof BigDecimal) { + return ((BigDecimal) o1).compareTo((BigDecimal) o2) == 0; + } + return o1.equals(o2); + } + @Override public E current() { return castNonNull(current); } @@ -284,10 +305,17 @@ static CSVReader openCsv(Source source, char separator) throws IOException { 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])) { + final @Nullable String fieldValue = field(strings, i); + if (fieldTypes != null && i < fieldTypes.size()) { + Object convertedFilter = rowConverter.convert(fieldTypes.get(i), filterValue); + Object convertedValue = rowConverter.convert(fieldTypes.get(i), fieldValue); + if (!objectsEqual(convertedFilter, convertedValue)) { + continue outer; + } + } else if (!filterValue.equals(fieldValue)) { continue outer; } } @@ -329,6 +357,11 @@ private static RelDataType toNullableRelDataType(JavaTypeFactory typeFactory, 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 element type */ @@ -480,7 +513,7 @@ static class ArrayRowConverter extends RowConverter<@Nullable Object[]> { 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 +523,7 @@ static class ArrayRowConverter extends RowConverter<@Nullable Object[]> { 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 +540,7 @@ private SingleColumnRowConverter(RelDataType fieldType, int fieldIndex) { } @Override public @Nullable Object convertRow(@Nullable String[] strings) { - return convert(fieldType, strings[fieldIndex]); + return convert(fieldType, field(strings, fieldIndex)); } } } diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java new file mode 100644 index 000000000000..b0ea564c0dd1 --- /dev/null +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java @@ -0,0 +1,217 @@ +/* + * 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.rex.RexCall; +import org.apache.calcite.rex.RexInputRef; +import org.apache.calcite.rex.RexLiteral; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexUtil; +import org.apache.calcite.sql.SqlKind; + +import org.checkerframework.checker.nullness.qual.Nullable; +import org.immutables.value.Value; + +import java.util.ArrayList; +import java.util.List; + +/** + * Planner rule that pushes simple equality filter predicates into a + * {@link CsvTableScan}. + * + *

Only equality conditions of the form {@code column = literal} can be + * pushed down, because {@link CsvEnumerator} only supports per-column + * equality filtering via its {@code filterValues} array. + * Any predicates that cannot be pushed down (e.g., range comparisons, + * {@code OR} expressions, or comparisons involving expressions) are left + * as a residual {@link LogicalFilter} above the scan. + * + *

This rule fires after {@link CsvProjectTableScanRule} so that the + * scan's field list may already be a subset of the full table fields. + * Filter column references are mapped back through the projected field list + * to the original column indices expected by {@link CsvEnumerator}. + * + * @see FileRules#FILTER_SCAN + */ +@Value.Enclosing +public class CsvFilterTableScanRule + extends RelRule { + + /** 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); + + // filterValues is indexed by the *full* table column index so that + // CsvEnumerator can match against the raw CSV row directly. + final int fullFieldCount = scan.getTable().getRowType().getFieldCount(); + final @Nullable String[] filterValues = new String[fullFieldCount]; + + // Partition the filter condition into predicates we can push down + // (simple column = literal equality) and predicates we cannot. + final List residualFilters = new ArrayList<>(); + decomposeFilter(filter.getCondition(), scan.fields, filterValues, residualFilters); + + // Only transform if at least one predicate could be pushed into the scan. + boolean anyPushed = false; + for (String v : filterValues) { + if (v != null) { + anyPushed = true; + break; + } + } + if (!anyPushed) { + return; + } + + // Build a new scan that carries the pushed-down filter values. + final CsvTableScan newScan = + new CsvTableScan(scan.getCluster(), scan.getTable(), scan.csvTable, + scan.fields, filterValues); + + // If there are residual predicates that could not be pushed down, + // keep a LogicalFilter node above the new scan to evaluate them. + final RelNode result; + if (residualFilters.isEmpty()) { + result = newScan; + } else { + final RexNode residual = + RexUtil.composeConjunction(scan.getCluster().getRexBuilder(), residualFilters); + result = filter.copy(filter.getTraitSet(), newScan, residual); + } + call.transformTo(result); + } + + /** + * Decomposes a filter condition into pushable equality predicates and + * non-pushable residual predicates. + * + *

AND conjunctions are recursively decomposed so that pushable + * sub-predicates can be extracted even when some siblings are not pushable. + * + *

A predicate {@code col = literal} is pushable when: + *

    + *
  • The left operand is a {@link RexInputRef} (optionally wrapped in + * a {@code CAST}), referring to one of the projected columns in + * {@code projectedFields}.
    • + *
  • The right operand is a {@link RexLiteral}.
    • + *
  • No earlier predicate has already set a value for the same column + * (first match wins).
    • + *
+ * + * @param condition The filter condition to decompose + * @param projectedFields The field projection array from the current + * {@link CsvTableScan} ({@code scan.fields}), used + * to map projected column indices back to full-table + * column indices for {@code filterValues} + * @param filterValues Output: per-full-table-column equality values; + * populated in place for pushable predicates + * @param residualFilters Output: predicates that could not be pushed down + */ + private static void decomposeFilter(RexNode condition, int[] projectedFields, + @Nullable String[] filterValues, List residualFilters) { + if (condition.isA(SqlKind.AND)) { + // Decompose AND: process each conjunct independently so that pushable + // sub-predicates can be separated from non-pushable ones. + for (RexNode operand : ((RexCall) condition).getOperands()) { + decomposeFilter(operand, projectedFields, filterValues, residualFilters); + } + } else if (condition.isA(SqlKind.EQUALS)) { + if (!tryPushEquality((RexCall) condition, projectedFields, filterValues)) { + residualFilters.add(condition); + } + } else { + // Any other predicate kind (OR, comparison, function call, etc.) + // cannot be pushed into the CsvEnumerator's string-equality filter. + residualFilters.add(condition); + } + } + + /** + * Attempts to push a single equality predicate into {@code filterValues}. + * + * @return {@code true} if the predicate was pushed; {@code false} if it + * must remain as a residual filter + */ + private static boolean tryPushEquality(RexCall call, int[] projectedFields, + @Nullable String[] filterValues) { + RexNode left = call.getOperands().get(0); + final RexNode right = call.getOperands().get(1); + + // Unwrap a CAST on the left side (e.g., CAST(col AS VARCHAR) = 'val'). + if (left.isA(SqlKind.CAST)) { + left = ((RexCall) left).getOperands().get(0); + } + + if (!(left instanceof RexInputRef) || !(right instanceof RexLiteral)) { + return false; + } + + // The RexInputRef index is relative to the *projected* row type of the + // current scan, not to the full table row type. Map it back. + final int projectedIndex = ((RexInputRef) left).getIndex(); + if (projectedIndex >= projectedFields.length) { + return false; + } + final int fullTableIndex = projectedFields[projectedIndex]; + + // First equality for this column wins; ignore duplicates. + if (filterValues[fullTableIndex] != null) { + return false; + } + + final RexLiteral literal = (RexLiteral) right; + final Object value; + switch (literal.getTypeName()) { + case CHAR: + case VARCHAR: + value = literal.getValueAs(String.class); + break; + default: + value = literal.getValueAs(Comparable.class); + } + if (value == null) { + // NULL literals cannot be represented by CsvEnumerator's filter values. + return false; + } + + filterValues[fullTableIndex] = value.toString(); + return true; + } + + /** 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); + } + } +} diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java new file mode 100644 index 000000000000..7693e9c176d5 --- /dev/null +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java @@ -0,0 +1,169 @@ +/* + * 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.RexCall; +import org.apache.calcite.rex.RexInputRef; +import org.apache.calcite.rex.RexLiteral; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexUtil; +import org.apache.calcite.sql.SqlKind; + +import org.checkerframework.checker.nullness.qual.Nullable; +import org.immutables.value.Value; + +import java.util.ArrayList; +import java.util.List; + +/** + * Planner rule that matches a {@link LogicalProject} on a {@link LogicalFilter} + * on a {@link CsvTableScan}, and pushes simple equality filter predicates + * into the scan. + * + * @see FileRules#PROJECT_FILTER_SCAN + */ +@Value.Enclosing +public class CsvProjectFilterTableScanRule + extends RelRule { + + /** 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); + + final int fullFieldCount = scan.getTable().getRowType().getFieldCount(); + final @Nullable String[] filterValues = new String[fullFieldCount]; + + // Partition the filter condition into predicates we can push down + // (simple column = literal equality) and predicates we cannot. + final List residualFilters = new ArrayList<>(); + decomposeFilter(filter.getCondition(), scan.fields, filterValues, residualFilters); + + // Only transform if at least one predicate could be pushed into the scan. + boolean anyPushed = false; + for (String v : filterValues) { + if (v != null) { + anyPushed = true; + break; + } + } + if (!anyPushed) { + return; + } + + // Build a new scan that carries the pushed-down filter values. + final CsvTableScan newScan = + new CsvTableScan(scan.getCluster(), scan.getTable(), scan.csvTable, + scan.fields, filterValues); + + // If there are residual predicates that could not be pushed down, + // keep a LogicalFilter node above the new scan to evaluate them. + RelNode rel = newScan; + if (!residualFilters.isEmpty()) { + final RexNode residual = + RexUtil.composeConjunction(scan.getCluster().getRexBuilder(), residualFilters); + rel = filter.copy(filter.getTraitSet(), newScan, residual); + } + + // Keep the LogicalProject on top of the scan/filter. + final RelNode result = + project.copy(project.getTraitSet(), rel, project.getProjects(), project.getRowType()); + + call.transformTo(result); + } + + private static void decomposeFilter(RexNode condition, int[] projectedFields, + @Nullable String[] filterValues, List residualFilters) { + if (condition.isA(SqlKind.AND)) { + for (RexNode operand : ((RexCall) condition).getOperands()) { + decomposeFilter(operand, projectedFields, filterValues, residualFilters); + } + } else if (condition.isA(SqlKind.EQUALS)) { + if (!tryPushEquality((RexCall) condition, projectedFields, filterValues)) { + residualFilters.add(condition); + } + } else { + residualFilters.add(condition); + } + } + + private static boolean tryPushEquality(RexCall call, int[] projectedFields, + @Nullable String[] filterValues) { + RexNode left = call.getOperands().get(0); + final RexNode right = call.getOperands().get(1); + + if (left.isA(SqlKind.CAST)) { + left = ((RexCall) left).getOperands().get(0); + } + + if (!(left instanceof RexInputRef) || !(right instanceof RexLiteral)) { + return false; + } + + final int projectedIndex = ((RexInputRef) left).getIndex(); + if (projectedIndex >= projectedFields.length) { + return false; + } + final int fullTableIndex = projectedFields[projectedIndex]; + + if (filterValues[fullTableIndex] != null) { + return false; + } + + final RexLiteral literal = (RexLiteral) right; + final Object value; + switch (literal.getTypeName()) { + case CHAR: + case VARCHAR: + value = literal.getValueAs(String.class); + break; + default: + value = literal.getValueAs(Comparable.class); + } + if (value == null) { + return false; + } + + filterValues[fullTableIndex] = value.toString(); + return true; + } + + /** 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); + } + } +} diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java index a0e006ae4ca8..67d40f18b167 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java @@ -45,17 +45,25 @@ 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; } + // 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.filterValues)); } private static int[] getProjectFields(List exps) { diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java index 8d7e80c6ee71..6313d2920f20 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java @@ -42,7 +42,10 @@ import org.checkerframework.checker.nullness.qual.Nullable; +import java.util.Arrays; import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import static java.util.Objects.requireNonNull; @@ -53,23 +56,45 @@ */ public class CsvTableScan extends TableScan implements EnumerableRel { final CsvTranslatableTable csvTable; - private final int[] fields; + final int[] fields; + final @Nullable String[] filterValues; protected CsvTableScan(RelOptCluster cluster, RelOptTable table, CsvTranslatableTable csvTable, int[] fields) { + this(cluster, table, csvTable, fields, null); + } + + protected CsvTableScan(RelOptCluster cluster, RelOptTable table, + CsvTranslatableTable csvTable, int[] fields, + @Nullable String @Nullable [] filterValues) { super(cluster, cluster.traitSetOf(EnumerableConvention.INSTANCE), ImmutableList.of(), table); this.csvTable = requireNonNull(csvTable, "csvTable"); this.fields = fields; + this.filterValues = filterValues; } @Override public RelNode copy(RelTraitSet traitSet, List inputs) { assert inputs.isEmpty(); - return new CsvTableScan(getCluster(), table, csvTable, fields); + return new CsvTableScan(getCluster(), table, csvTable, fields, filterValues); } @Override public RelWriter explainTerms(RelWriter pw) { return super.explainTerms(pw) - .item("fields", Primitive.asList(fields)); + .item("fields", Primitive.asList(fields)) + .itemIf("filters", filtersToString(filterValues), filterValues != null); + } + + /** Returns a human-readable representation of the filter values for EXPLAIN output. + * + *

For example, if column 3 equals "F", returns "[3=F]". */ + private static @Nullable String filtersToString(@Nullable String @Nullable [] filterValues) { + if (filterValues == null) { + return null; + } + return IntStream.range(0, filterValues.length) + .filter(i -> filterValues[i] != null) + .mapToObj(i -> i + "=" + filterValues[i]) + .collect(Collectors.joining(", ", "[", "]")); } @Override public RelDataType deriveRowType() { @@ -84,6 +109,8 @@ 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, @@ -96,9 +123,14 @@ protected CsvTableScan(RelOptCluster cluster, RelOptTable table, // 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) + / ((double) table.getRowType().getFieldCount() + 2D); + if (filterValues != null) { + // A scan with filters pushed down eliminates rows early; reduce cost further. + long filterCount = Arrays.stream(filterValues).filter(v -> v != null).count(); + factor *= Math.pow(0.5, filterCount); + } + return cost.multiplyBy(factor); } @Override public Result implement(EnumerableRelImplementor implementor, Prefer pref) { @@ -110,11 +142,24 @@ protected CsvTableScan(RelOptCluster cluster, RelOptTable table, final Expression expression = requireNonNull(table.getExpression(CsvTranslatableTable.class)); - return implementor.result( - physType, - Blocks.toBlock( - Expressions.call(expression, - "project", implementor.getRootExpression(), - Expressions.constant(fields)))); + + if (filterValues != null) { + // Call CsvTranslatableTable.scan(root, fields, filterValues) + return implementor.result( + physType, + Blocks.toBlock( + Expressions.call(expression, + "scan", implementor.getRootExpression(), + Expressions.constant(fields), + Expressions.constant(filterValues)))); + } else { + // Call CsvTranslatableTable.project(root, fields) — existing path + return implementor.result( + physType, + Blocks.toBlock( + Expressions.call(expression, + "project", implementor.getRootExpression(), + Expressions.constant(fields)))); + } } } diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java index 7f81defebe01..704994f69544 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java @@ -26,6 +26,7 @@ import org.apache.calcite.linq4j.tree.Expression; import org.apache.calcite.plan.RelOptTable; import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelProtoDataType; import org.apache.calcite.schema.QueryableTable; import org.apache.calcite.schema.SchemaPlus; @@ -37,6 +38,7 @@ import org.checkerframework.checker.nullness.qual.Nullable; import java.lang.reflect.Type; +import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; /** @@ -72,6 +74,36 @@ public Enumerable project(final DataContext root, }; } + /** Returns an enumerable over a given projection of the fields, with + * filter values applied during the scan to skip non-matching rows. + * + *

This method is called from generated code (via + * {@link CsvTableScan#implement}) when filter predicates have been pushed + * down into the scan by {@link CsvFilterTableScanRule}. + * + * @param root Data context (provides type factory and cancel flag) + * @param fields Indices of the fields to project (into full table schema) + * @param filterValues Per-column equality filter values; null means no filter + * for that column. Indexed by full-table column index. + */ + @SuppressWarnings({"unchecked", "unused"}) // called from generated code + public Enumerable scan(final DataContext root, + final int[] fields, final @Nullable String @Nullable [] filterValues) { + final AtomicBoolean cancelFlag = DataContext.Variable.CANCEL_FLAG.get(root); + return new AbstractEnumerable() { + @Override public Enumerator enumerator() { + JavaTypeFactory typeFactory = root.getTypeFactory(); + List fieldTypes = getFieldTypes(typeFactory); + return (Enumerator) (Enumerator) + new CsvEnumerator<>(source, cancelFlag, false, filterValues, + fieldTypes, + CsvEnumerator.converter(fieldTypes, + ImmutableIntList.of(fields)), + separator); + } + }; + } + @Override public Expression getExpression(SchemaPlus schema, String tableName, Class clazz) { return Schemas.tableExpression(schema, getElementType(), tableName, clazz); diff --git a/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java b/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java index 9c7e228c746d..8c29b44e292b 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java @@ -24,4 +24,15 @@ private FileRules() {} * a {@link CsvTableScan} and pushes down projects if possible. */ public static final CsvProjectTableScanRule PROJECT_SCAN = CsvProjectTableScanRule.Config.DEFAULT.toRule(); + + /** Rule that matches a {@link org.apache.calcite.rel.core.Filter} on + * a {@link CsvTableScan} and pushes down simple equality predicates. */ + public static final CsvFilterTableScanRule FILTER_SCAN = + CsvFilterTableScanRule.Config.DEFAULT.toRule(); + + /** Rule that matches a {@link org.apache.calcite.rel.core.Project} on + * a {@link org.apache.calcite.rel.core.Filter} on a {@link CsvTableScan} + * and pushes down simple equality predicates. */ + public static final CsvProjectFilterTableScanRule PROJECT_FILTER_SCAN = + CsvProjectFilterTableScanRule.Config.DEFAULT.toRule(); } diff --git a/file/src/test/java/org/apache/calcite/adapter/file/CsvEnumeratorTest.java b/file/src/test/java/org/apache/calcite/adapter/file/CsvEnumeratorTest.java index 43f4a6b24f9f..92566afbb546 100644 --- a/file/src/test/java/org/apache/calcite/adapter/file/CsvEnumeratorTest.java +++ b/file/src/test/java/org/apache/calcite/adapter/file/CsvEnumeratorTest.java @@ -16,12 +16,16 @@ */ package org.apache.calcite.adapter.file; +import org.apache.calcite.rel.type.RelDataType; + import org.junit.jupiter.api.Test; import java.math.BigDecimal; +import java.util.Arrays; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertThrows; /** @@ -57,4 +61,14 @@ private static void checkThrows(int precision, int scale, String s) { assertThrows(IllegalArgumentException.class, () -> CsvEnumerator.parseDecimal(precision, scale, s)); } + + @Test void testConvertRowWithMissingFields() { + final CsvEnumerator.RowConverter converter = + CsvEnumerator.arrayConverter( + Arrays.asList(null, null, null, null), + Arrays.asList(0, 1, 2, 3), false); + + assertArrayEquals(new Object[] {"a", "b", "c", null}, + converter.convertRow(new String[] {"a", "b", "c"})); + } } diff --git a/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java b/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java index ad774f3964e6..ea7f03767d4f 100644 --- a/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java +++ b/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java @@ -417,6 +417,63 @@ private static void checkEmpty(ResultSet resultSet) { sql("model-with-custom-table", sql).ok(); } + /** Test case for + * [CALCITE-7618] + * Add filter pushdown support to file adapter's CSV implementation. + * + *

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]], filters=[[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=CsvTableScan(table=[[SALES, EMPS]]," + + " fields=[[1, 0]], filters=[[2=20]])\n"; + sql("smart", sql).returns(expected).ok(); + } + + /** Test case for + * [CALCITE-7618] + * Add filter pushdown support to file adapter's CSV implementation. + * + *

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 + * [CALCITE-7618] + * Add filter pushdown support to file adapter's CSV implementation. + * + *

Verifies that non-equality (non-pushable) filters remain as a residual + * {@code EnumerableCalc} above the scan rather than being silently dropped. */ + @Test void testNonPushableFilterRemains() { + // empno > 110 is a range filter; CsvEnumerator only supports equality, + // so it cannot be pushed down and must stay in the plan as 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 testPushDownProject() { final String sql = "explain plan for select * from EMPS"; final String expected = "PLAN=CsvTableScan(table=[[SALES, EMPS]], " @@ -471,21 +528,15 @@ void testPushDownProjectAggregateWithFilter(String format) { switch (format) { case "dot": expected = "PLAN=digraph {\n" - + "\"EnumerableCalc\\nexpr#0..1 = {inputs}\\nexpr#2 = 'F':VARCHAR\\nexpr#3 = =($t1, $t2)" - + "\\nproj#0..1 = {exprs}\\n$condition = $t3\" -> \"EnumerableAggregate\\ngroup = " - + "{}\\nEXPR$0 = MAX($0)\\n\" [label=\"0\"]\n" - + "\"CsvTableScan\\ntable = [SALES, EMPS\\n]\\nfields = [0, 3]\\n\" -> " - + "\"EnumerableCalc\\nexpr#0..1 = {inputs}\\nexpr#2 = 'F':VARCHAR\\nexpr#3 = =($t1, $t2)" - + "\\nproj#0..1 = {exprs}\\n$condition = $t3\" [label=\"0\"]\n" + + "\"CsvTableScan\\ntable = [SALES, EMPS\\n]\\nfields = [0]\\nfilters = [3=F]\\n\" " + + "-> \"EnumerableAggregate\\ngroup = {}\\nEXPR$0 = MAX($0)\\n\" [label=\"0\"]\n" + "}\n"; extra = " as dot "; break; case "text": expected = "PLAN=" + "EnumerableAggregate(group=[{}], EXPR$0=[MAX($0)])\n" - + " EnumerableCalc(expr#0..1=[{inputs}], expr#2=['F':VARCHAR], " - + "expr#3=[=($t1, $t2)], proj#0..1=[{exprs}], $condition=[$t3])\n" - + " CsvTableScan(table=[[SALES, EMPS]], fields=[[0, 3]])\n"; + + " CsvTableScan(table=[[SALES, EMPS]], fields=[[0]], filters=[[3=F]])\n"; extra = ""; break; } From 44f7f040ae8f6e13888c2395c95d485e59cd6b85 Mon Sep 17 00:00:00 2001 From: Diveyam Mishra Date: Fri, 3 Jul 2026 03:34:03 +0530 Subject: [PATCH 2/3] Extending Test Coverage and Support arbitrary filter predicates --- .../java/org/apache/calcite/test/CsvTest.java | 16 ++ .../calcite/adapter/file/CsvEnumerator.java | 42 ++- .../adapter/file/CsvFilterTableScanRule.java | 170 ++---------- .../file/CsvProjectFilterTableScanRule.java | 148 +++++------ .../adapter/file/CsvProjectTableScanRule.java | 7 +- .../calcite/adapter/file/CsvTableScan.java | 85 +++--- .../adapter/file/CsvTranslatableTable.java | 32 --- .../calcite/adapter/file/FileRules.java | 10 +- .../calcite/adapter/file/FileAdapterTest.java | 245 +++++++++++++++++- 9 files changed, 398 insertions(+), 357 deletions(-) diff --git a/example/csv/src/test/java/org/apache/calcite/test/CsvTest.java b/example/csv/src/test/java/org/apache/calcite/test/CsvTest.java index 7616fcef26a6..2236345f06ab 100644 --- a/example/csv/src/test/java/org/apache/calcite/test/CsvTest.java +++ b/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 * [CALCITE-2272] * Incorrect result for {@code name like '%E%' and city not like '%W%'}. diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java index 0261a5f59b50..a70246f196aa 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java @@ -71,7 +71,6 @@ public class CsvEnumerator implements Enumerator { private final CSVReader reader; private final @Nullable List<@Nullable String> filterValues; - private final @Nullable List fieldTypes; private final AtomicBoolean cancelFlag; private final RowConverter rowConverter; private @Nullable E current; @@ -116,25 +115,18 @@ private static void clearTimeFormats() { public CsvEnumerator(Source source, AtomicBoolean cancelFlag, List fieldTypes, List fields, char separator) { //noinspection unchecked - this(source, cancelFlag, false, null, fieldTypes, + this(source, cancelFlag, false, null, (RowConverter) converter(fieldTypes, fields), separator); } public CsvEnumerator(Source source, AtomicBoolean cancelFlag, boolean stream, @Nullable String @Nullable [] filterValues, RowConverter rowConverter, char separator) { - this(source, cancelFlag, stream, filterValues, null, rowConverter, separator); - } - - public CsvEnumerator(Source source, AtomicBoolean cancelFlag, boolean stream, - @Nullable String @Nullable [] filterValues, @Nullable List fieldTypes, - RowConverter rowConverter, char separator) { this.cancelFlag = cancelFlag; this.rowConverter = rowConverter; this.filterValues = filterValues == null ? null : ImmutableNullableList.copyOf(filterValues); - this.fieldTypes = fieldTypes; try { if (stream) { this.reader = new CsvStreamReader(source, separator); @@ -262,15 +254,26 @@ static CSVReader openCsv(Source source, char separator) throws IOException { return new CSVReader(source.reader(), separator); } - private static boolean objectsEqual(@Nullable Object o1, @Nullable Object o2) { - if (o1 == o2) { - return true; - } + /** + * Evaluates equality between two objects, conforming to SQL WHERE filter '=' semantics. + * + *

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. + * + *

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) { if (o1 == null || o2 == null) { return false; } - if (o1 instanceof BigDecimal && o2 instanceof BigDecimal) { - return ((BigDecimal) o1).compareTo((BigDecimal) o2) == 0; + if (o1 == o2) { + return true; + } + if (o1 instanceof Comparable && o2 instanceof Comparable && o1.getClass().isInstance(o2)) { + return ((Comparable) o1).compareTo(o2) == 0; } return o1.equals(o2); } @@ -308,14 +311,7 @@ private static boolean objectsEqual(@Nullable Object o1, @Nullable Object o2) { for (int i = 0; i < filterValues.size(); i++) { String filterValue = filterValues.get(i); if (filterValue != null) { - final @Nullable String fieldValue = field(strings, i); - if (fieldTypes != null && i < fieldTypes.size()) { - Object convertedFilter = rowConverter.convert(fieldTypes.get(i), filterValue); - Object convertedValue = rowConverter.convert(fieldTypes.get(i), fieldValue); - if (!objectsEqual(convertedFilter, convertedValue)) { - continue outer; - } - } else if (!filterValue.equals(fieldValue)) { + if (!filterValue.equals(strings[i])) { continue outer; } } diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java index b0ea564c0dd1..4ed61b2bcb58 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvFilterTableScanRule.java @@ -18,36 +18,23 @@ 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.rex.RexCall; -import org.apache.calcite.rex.RexInputRef; -import org.apache.calcite.rex.RexLiteral; import org.apache.calcite.rex.RexNode; import org.apache.calcite.rex.RexUtil; -import org.apache.calcite.sql.SqlKind; -import org.checkerframework.checker.nullness.qual.Nullable; import org.immutables.value.Value; -import java.util.ArrayList; -import java.util.List; - /** - * Planner rule that pushes simple equality filter predicates into a + * Planner rule that pushes filter predicates into a * {@link CsvTableScan}. * - *

Only equality conditions of the form {@code column = literal} can be - * pushed down, because {@link CsvEnumerator} only supports per-column - * equality filtering via its {@code filterValues} array. - * Any predicates that cannot be pushed down (e.g., range comparisons, - * {@code OR} expressions, or comparisons involving expressions) are left - * as a residual {@link LogicalFilter} above the scan. - * - *

This rule fires after {@link CsvProjectTableScanRule} so that the - * scan's field list may already be a subset of the full table fields. - * Filter column references are mapped back through the projected field list - * to the original column indices expected by {@link CsvEnumerator}. + *

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 */ @@ -64,141 +51,22 @@ protected CsvFilterTableScanRule(Config config) { final LogicalFilter filter = call.rel(0); final CsvTableScan scan = call.rel(1); - // filterValues is indexed by the *full* table column index so that - // CsvEnumerator can match against the raw CSV row directly. - final int fullFieldCount = scan.getTable().getRowType().getFieldCount(); - final @Nullable String[] filterValues = new String[fullFieldCount]; - - // Partition the filter condition into predicates we can push down - // (simple column = literal equality) and predicates we cannot. - final List residualFilters = new ArrayList<>(); - decomposeFilter(filter.getCondition(), scan.fields, filterValues, residualFilters); - - // Only transform if at least one predicate could be pushed into the scan. - boolean anyPushed = false; - for (String v : filterValues) { - if (v != null) { - anyPushed = true; - break; - } - } - if (!anyPushed) { - return; + // 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 values. + // Build a new scan that carries the pushed-down filter condition. final CsvTableScan newScan = new CsvTableScan(scan.getCluster(), scan.getTable(), scan.csvTable, - scan.fields, filterValues); - - // If there are residual predicates that could not be pushed down, - // keep a LogicalFilter node above the new scan to evaluate them. - final RelNode result; - if (residualFilters.isEmpty()) { - result = newScan; - } else { - final RexNode residual = - RexUtil.composeConjunction(scan.getCluster().getRexBuilder(), residualFilters); - result = filter.copy(filter.getTraitSet(), newScan, residual); - } - call.transformTo(result); - } - - /** - * Decomposes a filter condition into pushable equality predicates and - * non-pushable residual predicates. - * - *

AND conjunctions are recursively decomposed so that pushable - * sub-predicates can be extracted even when some siblings are not pushable. - * - *

A predicate {@code col = literal} is pushable when: - *

    - *
  • The left operand is a {@link RexInputRef} (optionally wrapped in - * a {@code CAST}), referring to one of the projected columns in - * {@code projectedFields}.
    • - *
  • The right operand is a {@link RexLiteral}.
    • - *
  • No earlier predicate has already set a value for the same column - * (first match wins).
    • - *
- * - * @param condition The filter condition to decompose - * @param projectedFields The field projection array from the current - * {@link CsvTableScan} ({@code scan.fields}), used - * to map projected column indices back to full-table - * column indices for {@code filterValues} - * @param filterValues Output: per-full-table-column equality values; - * populated in place for pushable predicates - * @param residualFilters Output: predicates that could not be pushed down - */ - private static void decomposeFilter(RexNode condition, int[] projectedFields, - @Nullable String[] filterValues, List residualFilters) { - if (condition.isA(SqlKind.AND)) { - // Decompose AND: process each conjunct independently so that pushable - // sub-predicates can be separated from non-pushable ones. - for (RexNode operand : ((RexCall) condition).getOperands()) { - decomposeFilter(operand, projectedFields, filterValues, residualFilters); - } - } else if (condition.isA(SqlKind.EQUALS)) { - if (!tryPushEquality((RexCall) condition, projectedFields, filterValues)) { - residualFilters.add(condition); - } - } else { - // Any other predicate kind (OR, comparison, function call, etc.) - // cannot be pushed into the CsvEnumerator's string-equality filter. - residualFilters.add(condition); - } - } - - /** - * Attempts to push a single equality predicate into {@code filterValues}. - * - * @return {@code true} if the predicate was pushed; {@code false} if it - * must remain as a residual filter - */ - private static boolean tryPushEquality(RexCall call, int[] projectedFields, - @Nullable String[] filterValues) { - RexNode left = call.getOperands().get(0); - final RexNode right = call.getOperands().get(1); - - // Unwrap a CAST on the left side (e.g., CAST(col AS VARCHAR) = 'val'). - if (left.isA(SqlKind.CAST)) { - left = ((RexCall) left).getOperands().get(0); - } - - if (!(left instanceof RexInputRef) || !(right instanceof RexLiteral)) { - return false; - } - - // The RexInputRef index is relative to the *projected* row type of the - // current scan, not to the full table row type. Map it back. - final int projectedIndex = ((RexInputRef) left).getIndex(); - if (projectedIndex >= projectedFields.length) { - return false; - } - final int fullTableIndex = projectedFields[projectedIndex]; - - // First equality for this column wins; ignore duplicates. - if (filterValues[fullTableIndex] != null) { - return false; - } - - final RexLiteral literal = (RexLiteral) right; - final Object value; - switch (literal.getTypeName()) { - case CHAR: - case VARCHAR: - value = literal.getValueAs(String.class); - break; - default: - value = literal.getValueAs(Comparable.class); - } - if (value == null) { - // NULL literals cannot be represented by CsvEnumerator's filter values. - return false; - } + scan.fields, newCondition); - filterValues[fullTableIndex] = value.toString(); - return true; + call.transformTo(newScan); } /** Rule configuration. */ diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java index 7693e9c176d5..3bbcb5037ffb 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectFilterTableScanRule.java @@ -21,23 +21,17 @@ 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.RexCall; import org.apache.calcite.rex.RexInputRef; -import org.apache.calcite.rex.RexLiteral; import org.apache.calcite.rex.RexNode; import org.apache.calcite.rex.RexUtil; -import org.apache.calcite.sql.SqlKind; -import org.checkerframework.checker.nullness.qual.Nullable; import org.immutables.value.Value; -import java.util.ArrayList; import java.util.List; /** * Planner rule that matches a {@link LogicalProject} on a {@link LogicalFilter} - * on a {@link CsvTableScan}, and pushes simple equality filter predicates - * into the scan. + * on a {@link CsvTableScan}, and pushes filter predicates into the scan. * * @see FileRules#PROJECT_FILTER_SCAN */ @@ -55,101 +49,79 @@ protected CsvProjectFilterTableScanRule(Config config) { final LogicalFilter filter = call.rel(1); final CsvTableScan scan = call.rel(2); - final int fullFieldCount = scan.getTable().getRowType().getFieldCount(); - final @Nullable String[] filterValues = new String[fullFieldCount]; - - // Partition the filter condition into predicates we can push down - // (simple column = literal equality) and predicates we cannot. - final List residualFilters = new ArrayList<>(); - decomposeFilter(filter.getCondition(), scan.fields, filterValues, residualFilters); + // Find all input fields referenced by the project expressions + final java.util.Set projectInputFields = new java.util.HashSet<>(); + for (RexNode proj : project.getProjects()) { + proj.accept(new org.apache.calcite.rex.RexVisitorImpl(true) { + @Override public Void visitInputRef(RexInputRef inputRef) { + projectInputFields.add(inputRef.getIndex()); + return null; + } + }); + } - // Only transform if at least one predicate could be pushed into the scan. - boolean anyPushed = false; - for (String v : filterValues) { - if (v != null) { - anyPushed = true; - break; + // Find all input fields referenced by the filter condition + final java.util.Set filterInputFields = new java.util.HashSet<>(); + filter.getCondition().accept(new org.apache.calcite.rex.RexVisitorImpl(true) { + @Override public Void visitInputRef(RexInputRef inputRef) { + filterInputFields.add(inputRef.getIndex()); + return null; } - } - if (!anyPushed) { - return; + }); + + // Union the projected/referenced indices + final java.util.Set 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 a new scan that carries the pushed-down filter values. - final CsvTableScan newScan = - new CsvTableScan(scan.getCluster(), scan.getTable(), scan.csvTable, - scan.fields, filterValues); - - // If there are residual predicates that could not be pushed down, - // keep a LogicalFilter node above the new scan to evaluate them. - RelNode rel = newScan; - if (!residualFilters.isEmpty()) { - final RexNode residual = - RexUtil.composeConjunction(scan.getCluster().getRexBuilder(), residualFilters); - rel = filter.copy(filter.getTraitSet(), newScan, residual); + // Build index map from old projected index to new index in newFields + final java.util.Map indexMap = new java.util.HashMap<>(); + int newIdx = 0; + for (int idx : neededProjectedIndices) { + indexMap.put(idx, newIdx++); } - // Keep the LogicalProject on top of the scan/filter. - final RelNode result = - project.copy(project.getTraitSet(), rel, project.getProjects(), project.getRowType()); - - call.transformTo(result); - } - - private static void decomposeFilter(RexNode condition, int[] projectedFields, - @Nullable String[] filterValues, List residualFilters) { - if (condition.isA(SqlKind.AND)) { - for (RexNode operand : ((RexCall) condition).getOperands()) { - decomposeFilter(operand, projectedFields, filterValues, residualFilters); + // 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); } - } else if (condition.isA(SqlKind.EQUALS)) { - if (!tryPushEquality((RexCall) condition, projectedFields, filterValues)) { - residualFilters.add(condition); - } - } else { - residualFilters.add(condition); - } - } - - private static boolean tryPushEquality(RexCall call, int[] projectedFields, - @Nullable String[] filterValues) { - RexNode left = call.getOperands().get(0); - final RexNode right = call.getOperands().get(1); - - if (left.isA(SqlKind.CAST)) { - left = ((RexCall) left).getOperands().get(0); - } + }; - if (!(left instanceof RexInputRef) || !(right instanceof RexLiteral)) { - return false; + final RexNode mappedCondition = filter.getCondition().accept(shuttle); + final List mappedProjects = new java.util.ArrayList<>(); + for (RexNode proj : project.getProjects()) { + mappedProjects.add(proj.accept(shuttle)); } - final int projectedIndex = ((RexInputRef) left).getIndex(); - if (projectedIndex >= projectedFields.length) { - return false; + 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 int fullTableIndex = projectedFields[projectedIndex]; - if (filterValues[fullTableIndex] != null) { - return false; - } + final CsvTableScan newScan = + new CsvTableScan(scan.getCluster(), scan.getTable(), scan.csvTable, + newFields, finalCondition); - final RexLiteral literal = (RexLiteral) right; - final Object value; - switch (literal.getTypeName()) { - case CHAR: - case VARCHAR: - value = literal.getValueAs(String.class); - break; - default: - value = literal.getValueAs(Comparable.class); - } - if (value == null) { - return false; - } + final RelNode result = + project.copy(project.getTraitSet(), newScan, mappedProjects, project.getRowType()); - filterValues[fullTableIndex] = value.toString(); - return true; + call.transformTo(result); } /** Rule configuration. */ diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java index 67d40f18b167..79ba80722e40 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvProjectTableScanRule.java @@ -50,6 +50,11 @@ protected CsvProjectTableScanRule(Config config) { // 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. @@ -63,7 +68,7 @@ protected CsvProjectTableScanRule(Config config) { scan.getTable(), scan.csvTable, newFields, - scan.filterValues)); + scan.condition)); } private static int[] getProjectFields(List exps) { diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java index 6313d2920f20..76325e5f7334 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java @@ -16,6 +16,7 @@ */ package org.apache.calcite.adapter.file; +import org.apache.calcite.adapter.enumerable.EnumerableCalc; import org.apache.calcite.adapter.enumerable.EnumerableConvention; import org.apache.calcite.adapter.enumerable.EnumerableRel; import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor; @@ -37,15 +38,15 @@ import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelDataTypeFactory; import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexProgram; import com.google.common.collect.ImmutableList; import org.checkerframework.checker.nullness.qual.Nullable; -import java.util.Arrays; +import java.util.ArrayList; import java.util.List; -import java.util.stream.Collectors; -import java.util.stream.IntStream; import static java.util.Objects.requireNonNull; @@ -57,7 +58,7 @@ public class CsvTableScan extends TableScan implements EnumerableRel { final CsvTranslatableTable csvTable; final int[] fields; - final @Nullable String[] filterValues; + final @Nullable RexNode condition; protected CsvTableScan(RelOptCluster cluster, RelOptTable table, CsvTranslatableTable csvTable, int[] fields) { @@ -66,35 +67,22 @@ protected CsvTableScan(RelOptCluster cluster, RelOptTable table, protected CsvTableScan(RelOptCluster cluster, RelOptTable table, CsvTranslatableTable csvTable, int[] fields, - @Nullable String @Nullable [] filterValues) { + @Nullable RexNode condition) { super(cluster, cluster.traitSetOf(EnumerableConvention.INSTANCE), ImmutableList.of(), table); this.csvTable = requireNonNull(csvTable, "csvTable"); this.fields = fields; - this.filterValues = filterValues; + this.condition = condition; } @Override public RelNode copy(RelTraitSet traitSet, List inputs) { assert inputs.isEmpty(); - return new CsvTableScan(getCluster(), table, csvTable, fields, filterValues); + return new CsvTableScan(getCluster(), table, csvTable, fields, condition); } @Override public RelWriter explainTerms(RelWriter pw) { return super.explainTerms(pw) .item("fields", Primitive.asList(fields)) - .itemIf("filters", filtersToString(filterValues), filterValues != null); - } - - /** Returns a human-readable representation of the filter values for EXPLAIN output. - * - *

For example, if column 3 equals "F", returns "[3=F]". */ - private static @Nullable String filtersToString(@Nullable String @Nullable [] filterValues) { - if (filterValues == null) { - return null; - } - return IntStream.range(0, filterValues.length) - .filter(i -> filterValues[i] != null) - .mapToObj(i -> i + "=" + filterValues[i]) - .collect(Collectors.joining(", ", "[", "]")); + .itemIf("condition", condition, condition != null); } @Override public RelDataType deriveRowType() { @@ -115,20 +103,11 @@ protected CsvTableScan(RelOptCluster cluster, RelOptTable table, @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)); double factor = ((double) fields.length + 2D) / ((double) table.getRowType().getFieldCount() + 2D); - if (filterValues != null) { - // A scan with filters pushed down eliminates rows early; reduce cost further. - long filterCount = Arrays.stream(filterValues).filter(v -> v != null).count(); - factor *= Math.pow(0.5, filterCount); + if (condition != null) { + factor *= 0.5; } return cost.multiplyBy(factor); } @@ -143,23 +122,31 @@ protected CsvTableScan(RelOptCluster cluster, RelOptTable table, final Expression expression = requireNonNull(table.getExpression(CsvTranslatableTable.class)); - if (filterValues != null) { - // Call CsvTranslatableTable.scan(root, fields, filterValues) - return implementor.result( - physType, - Blocks.toBlock( - Expressions.call(expression, - "scan", implementor.getRootExpression(), - Expressions.constant(fields), - Expressions.constant(filterValues)))); - } else { - // Call CsvTranslatableTable.project(root, fields) — existing path - return implementor.result( - physType, - Blocks.toBlock( - Expressions.call(expression, - "project", implementor.getRootExpression(), - Expressions.constant(fields)))); + // Call CsvTranslatableTable.project(root, fields) to get the base enumerable. + Expression enumerable = + Expressions.call(expression, + "project", implementor.getRootExpression(), + Expressions.constant(fields)); + + if (condition != null) { + final List projects = new ArrayList<>(); + for (int i = 0; i < getRowType().getFieldCount(); i++) { + projects.add( + getCluster().getRexBuilder().makeInputRef( + getRowType().getFieldList().get(i).getType(), i)); + } + final RexProgram program = + RexProgram.create(getRowType(), projects, condition, + getRowType(), getCluster().getRexBuilder()); + + // Create a scan node without the condition so EnumerableCalc sees a plain + // enumerable input, then wrap it with EnumerableCalc to apply the filter. + final CsvTableScan plainScan = + new CsvTableScan(getCluster(), table, csvTable, fields); + final EnumerableCalc calc = EnumerableCalc.create(plainScan, program); + return calc.implement(implementor, pref); } + + return implementor.result(physType, Blocks.toBlock(enumerable)); } } diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java index 704994f69544..7f81defebe01 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvTranslatableTable.java @@ -26,7 +26,6 @@ import org.apache.calcite.linq4j.tree.Expression; import org.apache.calcite.plan.RelOptTable; import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rel.type.RelProtoDataType; import org.apache.calcite.schema.QueryableTable; import org.apache.calcite.schema.SchemaPlus; @@ -38,7 +37,6 @@ import org.checkerframework.checker.nullness.qual.Nullable; import java.lang.reflect.Type; -import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; /** @@ -74,36 +72,6 @@ public Enumerable project(final DataContext root, }; } - /** Returns an enumerable over a given projection of the fields, with - * filter values applied during the scan to skip non-matching rows. - * - *

This method is called from generated code (via - * {@link CsvTableScan#implement}) when filter predicates have been pushed - * down into the scan by {@link CsvFilterTableScanRule}. - * - * @param root Data context (provides type factory and cancel flag) - * @param fields Indices of the fields to project (into full table schema) - * @param filterValues Per-column equality filter values; null means no filter - * for that column. Indexed by full-table column index. - */ - @SuppressWarnings({"unchecked", "unused"}) // called from generated code - public Enumerable scan(final DataContext root, - final int[] fields, final @Nullable String @Nullable [] filterValues) { - final AtomicBoolean cancelFlag = DataContext.Variable.CANCEL_FLAG.get(root); - return new AbstractEnumerable() { - @Override public Enumerator enumerator() { - JavaTypeFactory typeFactory = root.getTypeFactory(); - List fieldTypes = getFieldTypes(typeFactory); - return (Enumerator) (Enumerator) - new CsvEnumerator<>(source, cancelFlag, false, filterValues, - fieldTypes, - CsvEnumerator.converter(fieldTypes, - ImmutableIntList.of(fields)), - separator); - } - }; - } - @Override public Expression getExpression(SchemaPlus schema, String tableName, Class clazz) { return Schemas.tableExpression(schema, getElementType(), tableName, clazz); diff --git a/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java b/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java index 8c29b44e292b..15468c2328d8 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/FileRules.java @@ -26,13 +26,17 @@ private FileRules() {} CsvProjectTableScanRule.Config.DEFAULT.toRule(); /** Rule that matches a {@link org.apache.calcite.rel.core.Filter} on - * a {@link CsvTableScan} and pushes down simple equality predicates. */ + * a {@link CsvTableScan} and pushes arbitrary predicates into the scan. + * Any {@link org.apache.calcite.rex.RexNode} condition is compiled at plan + * time via {@link org.apache.calcite.adapter.enumerable.RexToLixTranslator} + * into a {@link org.apache.calcite.linq4j.function.Predicate1}. */ public static final CsvFilterTableScanRule FILTER_SCAN = CsvFilterTableScanRule.Config.DEFAULT.toRule(); /** Rule that matches a {@link org.apache.calcite.rel.core.Project} on - * a {@link org.apache.calcite.rel.core.Filter} on a {@link CsvTableScan} - * and pushes down simple equality predicates. */ + * a {@link org.apache.calcite.rel.core.Filter} on a {@link CsvTableScan}, + * pushes the filter condition into the scan, and remaps project and filter + * input references to match the scan's new projection. */ public static final CsvProjectFilterTableScanRule PROJECT_FILTER_SCAN = CsvProjectFilterTableScanRule.Config.DEFAULT.toRule(); } diff --git a/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java b/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java index ea7f03767d4f..8d32e042bf48 100644 --- a/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java +++ b/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java @@ -52,6 +52,7 @@ import static org.hamcrest.CoreMatchers.isA; import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * System test of the Calcite file adapter, which can read and parse @@ -426,14 +427,15 @@ private static void checkEmpty(ResultSet resultSet) { @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]], filters=[[2=20]])\n"; + + "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=CsvTableScan(table=[[SALES, EMPS]]," - + " fields=[[1, 0]], filters=[[2=20]])\n"; + 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(); } @@ -454,11 +456,11 @@ private static void checkEmpty(ResultSet resultSet) { * [CALCITE-7618] * Add filter pushdown support to file adapter's CSV implementation. * - *

Verifies that non-equality (non-pushable) filters remain as a residual - * {@code EnumerableCalc} above the scan rather than being silently dropped. */ - @Test void testNonPushableFilterRemains() { - // empno > 110 is a range filter; CsvEnumerator only supports equality, - // so it cannot be pushed down and must stay in the plan as EnumerableCalc. + *

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", @@ -474,6 +476,115 @@ private static void checkEmpty(ResultSet resultSet) { .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() { + final String sql = "select empno from \"DATE\" where" + + " jointimes = TIMESTAMP '2015-12-31 07:15:56'"; + sql("bug", sql) + .returns("EMPNO=140") + .ok(); + } + + @Test void testFilterPushDownLongPlan() { + sql("bug", "explain plan for select name from long_emps where empno = 130") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}]," + + " expr#2=[130:BIGINT], expr#3=[=($t0, $t2)], NAME=[$t1], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[0, 1]])\n") + .ok(); + } + + @Test void testFilterPushDownBooleanPlan() { + sql("bug", "explain plan for select name from long_emps where slacker = true") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], NAME=[$t0], $condition=[$t1])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[1, 7]])\n") + .ok(); + } + + @Test void testFilterPushDownStringPlan() { + sql("bug", "explain plan for select empno from long_emps where gender = 'F'") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=['F':VARCHAR]," + + " expr#3=[=($t1, $t2)], EMPNO=[$t0], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[0, 3]])\n") + .ok(); + } + + @Test void testFilterPushDownDecimalPlan() { + sql("sales-csv", "explain plan for select deptno from sales.\"DECIMAL\" where budget = 100.01") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], DEPTNO=[$t0])\n" + + " CsvTableScan(table=[[SALES, DECIMAL]], fields=[[0, 1]], condition=[=($1, 100.01)])\n") + .ok(); + } + + @Test void testFilterPushDownDatePlan() { + sql("bug", "explain plan for select name from long_emps where joinedat = DATE '2001-01-01'") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=[2001-01-01]," + + " expr#3=[=($t1, $t2)], NAME=[$t0], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[1, 9]])\n") + .ok(); + } + + @Test void testFilterPushDownTimePlan() { + sql("bug", "explain plan for select empno from \"DATE\" where jointime = TIME '07:15:56'") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=[07:15:56]," + + " expr#3=[=($t1, $t2)], EMPNO=[$t0], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, DATE]], fields=[[0, 2]])\n") + .ok(); + } + + @Test void testFilterPushDownTimestampPlan() { + sql("bug", "explain plan for select empno from \"DATE\"" + + " where jointimes = TIMESTAMP '2015-12-31 07:15:56'") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}]," + + " expr#2=[2015-12-31 07:15:56], expr#3=[=($t1, $t2)]," + + " EMPNO=[$t0], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, DATE]], fields=[[0, 3]])\n") + .ok(); + } + + + + @Test void testPushDownProject() { final String sql = "explain plan for select * from EMPS"; final String expected = "PLAN=CsvTableScan(table=[[SALES, EMPS]], " @@ -495,6 +606,47 @@ private static void checkEmpty(ResultSet resultSet) { .ok(); } + @Test void testFilterPushDownOr() { + sql("smart", "select name from EMPS where deptno = 20 or empno = 100") + .returns("NAME=Fred", "NAME=Eric", "NAME=Wilma") + .ok(); + } + + @Test void testFilterPushDownOrPlan() { + sql("smart", "explain plan for select name from EMPS where deptno = 20 or empno = 100") + .returns("PLAN=EnumerableCalc(expr#0..2=[{inputs}], expr#3=[20]," + + " expr#4=[=($t2, $t3)], expr#5=[100], expr#6=[=($t0, $t5)]," + + " expr#7=[OR($t4, $t6)], NAME=[$t1], $condition=[$t7])\n" + + " CsvTableScan(table=[[SALES, EMPS]], fields=[[0, 1, 2]])\n") + .ok(); + } + + @Test void testFilterPushDownNotEquals() { + sql("smart", "select name from EMPS where deptno <> 20") + .returns("NAME=Fred", "NAME=John", "NAME=Alice") + .ok(); + } + + @Test void testFilterPushDownNotEqualsPlan() { + sql("smart", "explain plan for select name from EMPS where deptno <> 20") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], NAME=[$t0])\n" + + " CsvTableScan(table=[[SALES, EMPS]], fields=[[1, 2]], condition=[<>($1, 20)])\n") + .ok(); + } + + @Test void testFilterPushDownRange() { + sql("smart", "select name from EMPS where empno >= 120") + .returns("NAME=Wilma", "NAME=Alice") + .ok(); + } + + @Test void testFilterPushDownRangePlan() { + sql("smart", "explain plan for select name from EMPS where empno >= 120") + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], NAME=[$t1])\n" + + " CsvTableScan(table=[[SALES, EMPS]], fields=[[0, 1]], condition=[>=($0, 120)])\n") + .ok(); + } + @ParameterizedTest @MethodSource("explainFormats") void testPushDownProjectAggregate(String format) { @@ -528,7 +680,7 @@ void testPushDownProjectAggregateWithFilter(String format) { switch (format) { case "dot": expected = "PLAN=digraph {\n" - + "\"CsvTableScan\\ntable = [SALES, EMPS\\n]\\nfields = [0]\\nfilters = [3=F]\\n\" " + + "\"CsvTableScan\\ntable = [SALES, EMPS\\n]\\nfields = [0, 3]\\ncondition = =($1, 'F\\n')\\n\" " + "-> \"EnumerableAggregate\\ngroup = {}\\nEXPR$0 = MAX($0)\\n\" [label=\"0\"]\n" + "}\n"; extra = " as dot "; @@ -536,7 +688,7 @@ void testPushDownProjectAggregateWithFilter(String format) { case "text": expected = "PLAN=" + "EnumerableAggregate(group=[{}], EXPR$0=[MAX($0)])\n" - + " CsvTableScan(table=[[SALES, EMPS]], fields=[[0]], filters=[[3=F]])\n"; + + " CsvTableScan(table=[[SALES, EMPS]], fields=[[0, 3]], condition=[=($1, 'F')])\n"; extra = ""; break; } @@ -1156,4 +1308,77 @@ private String range(int first, int count) { is(Timestamp.valueOf("1996-08-03 00:01:02"))); } } + + @Test void testFilterPushDownDoesNotReturnNullRows() { + // age column has nulls in the data — make sure they're excluded, not included + final String sql = "select name from long_emps where age = 25"; + sql("bug", sql) + .returns("NAME=Fred") // only Fred has age=25, null-age rows must not appear + .ok(); + } + + @Test void testFilterPushDownNullColumnExcluded() { + // slacker has null values — null rows must not match true or false + final String sql = "select name from long_emps where slacker = false"; + sql("bug", sql) + .returns("NAME=John", "NAME=Alice") // Eric and Wilma have null slacker — excluded + .ok(); + } + + @Test void testObjectsEqualBehavior() { + // Basic null behavior + assertFalse(CsvEnumerator.objectsEqual(null, null)); + assertFalse(CsvEnumerator.objectsEqual(null, new BigDecimal("1.0"))); + assertFalse(CsvEnumerator.objectsEqual(new BigDecimal("1.0"), null)); + assertFalse(CsvEnumerator.objectsEqual(null, "hello")); + assertFalse(CsvEnumerator.objectsEqual("hello", null)); + + // Mixed null and zero + assertFalse(CsvEnumerator.objectsEqual(null, 0)); + assertFalse(CsvEnumerator.objectsEqual(null, BigDecimal.ZERO)); + assertFalse(CsvEnumerator.objectsEqual(null, "")); + + // NULL IS NOT DISTINCT FROM NULL → should be TRUE under IS NOT DISTINCT FROM semantics, + // but objectsEqual implements SQL WHERE filter '=' semantics (where null = null evaluates + // to UNKNOWN, which behaves as false). + assertFalse(CsvEnumerator.objectsEqual(null, 1)); + assertFalse(CsvEnumerator.objectsEqual(1, null)); + + // Large scale differences + assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("1.000000"), new BigDecimal("1"))); + + // Negative zero edge case + assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("0.0"), new BigDecimal("-0.0"))); + + // Very large numbers with scale + assertTrue( + CsvEnumerator.objectsEqual( + new BigDecimal("999999999.9"), new BigDecimal("999999999.90"))); + + // Strings + assertTrue(CsvEnumerator.objectsEqual("hello", "hello")); + assertFalse(CsvEnumerator.objectsEqual("hello", "world")); + + // Integers / Longs + assertTrue(CsvEnumerator.objectsEqual(42, 42)); + assertFalse(CsvEnumerator.objectsEqual(42, 43)); + assertTrue(CsvEnumerator.objectsEqual(1L, 1L)); + + // Cross-type comparison (implicit type promotion is not handled by objectsEqual, returns false) + assertFalse(CsvEnumerator.objectsEqual(42, 42L)); + + // BigDecimal scale differences & symmetry + BigDecimal val = new BigDecimal("2.0"); + assertTrue(CsvEnumerator.objectsEqual(val, val)); + assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("2.0"), new BigDecimal("2.00"))); + assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("2.00"), new BigDecimal("2.0"))); + assertFalse(CsvEnumerator.objectsEqual(new BigDecimal("1.0"), new BigDecimal("2.0"))); + assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("0.0"), new BigDecimal("0.00"))); + assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("-1.0"), new BigDecimal("-1.00"))); + + // Objects.equals() performs exact class/structure comparison (including scale + // for BigDecimal), which incorrectly returns false for semantically equal numbers. + // Confirm Objects.equals fails here. + assertFalse(java.util.Objects.equals(new BigDecimal("2.0"), new BigDecimal("2.00"))); + } } From 3d0a4727cba4a8da218f7b20dc79d36ed532b5ef Mon Sep 17 00:00:00 2001 From: Diveyam Mishra Date: Fri, 10 Jul 2026 20:26:27 +0530 Subject: [PATCH 3/3] [CALCITE-7618] Refactor sameValue and consolidate tests --- .../calcite/adapter/file/CsvEnumerator.java | 8 +- .../calcite/adapter/file/CsvTableScan.java | 18 +- .../calcite/adapter/file/FileAdapterTest.java | 248 ++++++++++++------ 3 files changed, 185 insertions(+), 89 deletions(-) diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java index a70246f196aa..40d762600d7f 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvEnumerator.java @@ -255,7 +255,7 @@ static CSVReader openCsv(Source source, char separator) throws IOException { } /** - * Evaluates equality between two objects, conforming to SQL WHERE filter '=' semantics. + * Evaluates equality between two Comparable objects, conforming to SQL WHERE filter '=' semantics. * *

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. @@ -265,15 +265,15 @@ static CSVReader openCsv(Source source, char separator) throws IOException { * that would cause standard {@code equals()} to fail. */ @SuppressWarnings("unchecked") - static boolean objectsEqual(@Nullable Object o1, @Nullable Object o2) { + static boolean sameValue(@Nullable Comparable o1, @Nullable Comparable o2) { if (o1 == null || o2 == null) { return false; } if (o1 == o2) { return true; } - if (o1 instanceof Comparable && o2 instanceof Comparable && o1.getClass().isInstance(o2)) { - return ((Comparable) o1).compareTo(o2) == 0; + if (o1.getClass().isInstance(o2)) { + return o1.compareTo(o2) == 0; } return o1.equals(o2); } diff --git a/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java b/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java index 76325e5f7334..36b2ab332f34 100644 --- a/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java +++ b/file/src/main/java/org/apache/calcite/adapter/file/CsvTableScan.java @@ -103,12 +103,18 @@ protected CsvTableScan(RelOptCluster cluster, RelOptTable table, @Override public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) { - final RelOptCost cost = requireNonNull(super.computeSelfCost(planner, mq)); - double factor = ((double) fields.length + 2D) - / ((double) table.getRowType().getFieldCount() + 2D); - if (condition != null) { - factor *= 0.5; - } + // 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 the table has 3 fields and the scan has 1 field, + // then factor = (1 + 2) / (3 + 2) = 0.6. + final RelOptCost cost = + requireNonNull(super.computeSelfCost(planner, mq)); + final double factor = + (fields.length + 2D) + / (table.getRowType().getFieldCount() + 2D); return cost.multiplyBy(factor); } diff --git a/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java b/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java index 8d32e042bf48..4460ee926879 100644 --- a/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java +++ b/file/src/test/java/org/apache/calcite/adapter/file/FileAdapterTest.java @@ -17,8 +17,18 @@ package org.apache.calcite.adapter.file; import org.apache.calcite.jdbc.CalciteConnection; +import org.apache.calcite.plan.RelOptRule; +import org.apache.calcite.plan.RelOptUtil; +import org.apache.calcite.plan.hep.HepPlanner; +import org.apache.calcite.plan.hep.HepProgramBuilder; +import org.apache.calcite.rel.RelNode; import org.apache.calcite.schema.Schema; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.sql.SqlNode; import org.apache.calcite.sql2rel.SqlToRelConverter; +import org.apache.calcite.tools.FrameworkConfig; +import org.apache.calcite.tools.Frameworks; +import org.apache.calcite.tools.Planner; import org.apache.calcite.util.TestUtil; import com.google.common.collect.ImmutableMap; @@ -47,6 +57,7 @@ import static org.apache.calcite.adapter.file.FileAdapterTests.sql; +import static org.hamcrest.CoreMatchers.containsString; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.CoreMatchers.isA; @@ -481,6 +492,12 @@ private static void checkEmpty(ResultSet resultSet) { sql("bug", sql) .returns("NAME=Alice") .ok(); + final String plan = "explain plan for " + sql; + sql("bug", plan) + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}]," + + " expr#2=[130:BIGINT], expr#3=[=($t0, $t2)], NAME=[$t1], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[0, 1]])\n") + .ok(); } @Test void testFilterPushDownBoolean() { @@ -488,6 +505,11 @@ private static void checkEmpty(ResultSet resultSet) { sql("bug", sql) .returns("NAME=Fred") .ok(); + final String plan = "explain plan for " + sql; + sql("bug", plan) + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], NAME=[$t0], $condition=[$t1])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[1, 7]])\n") + .ok(); } @Test void testFilterPushDownString() { @@ -495,6 +517,12 @@ private static void checkEmpty(ResultSet resultSet) { sql("bug", sql) .returns("EMPNO=120", "EMPNO=130") .ok(); + final String plan = "explain plan for " + sql; + sql("bug", plan) + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=['F':VARCHAR]," + + " expr#3=[=($t1, $t2)], EMPNO=[$t0], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[0, 3]])\n") + .ok(); } @Test void testFilterPushDownDecimal() { @@ -502,6 +530,11 @@ private static void checkEmpty(ResultSet resultSet) { sql("sales-csv", sql) .returns("DEPTNO=20") .ok(); + final String plan = "explain plan for " + sql; + sql("sales-csv", plan) + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], DEPTNO=[$t0])\n" + + " CsvTableScan(table=[[SALES, DECIMAL]], fields=[[0, 1]], condition=[=($1, 100.01)])\n") + .ok(); } @Test void testFilterPushDownDate() { @@ -509,6 +542,12 @@ private static void checkEmpty(ResultSet resultSet) { sql("bug", sql) .returns("NAME=Eric") .ok(); + final String plan = "explain plan for " + sql; + sql("bug", plan) + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=[2001-01-01]," + + " expr#3=[=($t1, $t2)], NAME=[$t0], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[1, 9]])\n") + .ok(); } @Test void testFilterPushDownTime() { @@ -516,6 +555,12 @@ private static void checkEmpty(ResultSet resultSet) { sql("bug", sql) .returns("EMPNO=140") .ok(); + final String plan = "explain plan for " + sql; + sql("bug", plan) + .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=[07:15:56]," + + " expr#3=[=($t1, $t2)], EMPNO=[$t0], $condition=[$t3])\n" + + " CsvTableScan(table=[[BUG, DATE]], fields=[[0, 2]])\n") + .ok(); } @Test void testFilterPushDownTimestamp() { @@ -524,57 +569,8 @@ private static void checkEmpty(ResultSet resultSet) { sql("bug", sql) .returns("EMPNO=140") .ok(); - } - - @Test void testFilterPushDownLongPlan() { - sql("bug", "explain plan for select name from long_emps where empno = 130") - .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}]," - + " expr#2=[130:BIGINT], expr#3=[=($t0, $t2)], NAME=[$t1], $condition=[$t3])\n" - + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[0, 1]])\n") - .ok(); - } - - @Test void testFilterPushDownBooleanPlan() { - sql("bug", "explain plan for select name from long_emps where slacker = true") - .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], NAME=[$t0], $condition=[$t1])\n" - + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[1, 7]])\n") - .ok(); - } - - @Test void testFilterPushDownStringPlan() { - sql("bug", "explain plan for select empno from long_emps where gender = 'F'") - .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=['F':VARCHAR]," - + " expr#3=[=($t1, $t2)], EMPNO=[$t0], $condition=[$t3])\n" - + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[0, 3]])\n") - .ok(); - } - - @Test void testFilterPushDownDecimalPlan() { - sql("sales-csv", "explain plan for select deptno from sales.\"DECIMAL\" where budget = 100.01") - .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], DEPTNO=[$t0])\n" - + " CsvTableScan(table=[[SALES, DECIMAL]], fields=[[0, 1]], condition=[=($1, 100.01)])\n") - .ok(); - } - - @Test void testFilterPushDownDatePlan() { - sql("bug", "explain plan for select name from long_emps where joinedat = DATE '2001-01-01'") - .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=[2001-01-01]," - + " expr#3=[=($t1, $t2)], NAME=[$t0], $condition=[$t3])\n" - + " CsvTableScan(table=[[BUG, LONG_EMPS]], fields=[[1, 9]])\n") - .ok(); - } - - @Test void testFilterPushDownTimePlan() { - sql("bug", "explain plan for select empno from \"DATE\" where jointime = TIME '07:15:56'") - .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}], expr#2=[07:15:56]," - + " expr#3=[=($t1, $t2)], EMPNO=[$t0], $condition=[$t3])\n" - + " CsvTableScan(table=[[BUG, DATE]], fields=[[0, 2]])\n") - .ok(); - } - - @Test void testFilterPushDownTimestampPlan() { - sql("bug", "explain plan for select empno from \"DATE\"" - + " where jointimes = TIMESTAMP '2015-12-31 07:15:56'") + final String plan = "explain plan for " + sql; + sql("bug", plan) .returns("PLAN=EnumerableCalc(expr#0..1=[{inputs}]," + " expr#2=[2015-12-31 07:15:56], expr#3=[=($t1, $t2)]," + " EMPNO=[$t0], $condition=[$t3])\n" @@ -1325,60 +1321,154 @@ private String range(int first, int count) { .ok(); } - @Test void testObjectsEqualBehavior() { + @Test void testSameValueBehavior() { // Basic null behavior - assertFalse(CsvEnumerator.objectsEqual(null, null)); - assertFalse(CsvEnumerator.objectsEqual(null, new BigDecimal("1.0"))); - assertFalse(CsvEnumerator.objectsEqual(new BigDecimal("1.0"), null)); - assertFalse(CsvEnumerator.objectsEqual(null, "hello")); - assertFalse(CsvEnumerator.objectsEqual("hello", null)); + assertFalse(CsvEnumerator.sameValue(null, null)); + assertFalse(CsvEnumerator.sameValue(null, new BigDecimal("1.0"))); + assertFalse(CsvEnumerator.sameValue(new BigDecimal("1.0"), null)); + assertFalse(CsvEnumerator.sameValue(null, "hello")); + assertFalse(CsvEnumerator.sameValue("hello", null)); // Mixed null and zero - assertFalse(CsvEnumerator.objectsEqual(null, 0)); - assertFalse(CsvEnumerator.objectsEqual(null, BigDecimal.ZERO)); - assertFalse(CsvEnumerator.objectsEqual(null, "")); + assertFalse(CsvEnumerator.sameValue(null, 0)); + assertFalse(CsvEnumerator.sameValue(null, BigDecimal.ZERO)); + assertFalse(CsvEnumerator.sameValue(null, "")); // NULL IS NOT DISTINCT FROM NULL → should be TRUE under IS NOT DISTINCT FROM semantics, - // but objectsEqual implements SQL WHERE filter '=' semantics (where null = null evaluates + // but sameValue implements SQL WHERE filter '=' semantics (where null = null evaluates // to UNKNOWN, which behaves as false). - assertFalse(CsvEnumerator.objectsEqual(null, 1)); - assertFalse(CsvEnumerator.objectsEqual(1, null)); + assertFalse(CsvEnumerator.sameValue(null, 1)); + assertFalse(CsvEnumerator.sameValue(1, null)); // Large scale differences - assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("1.000000"), new BigDecimal("1"))); + assertTrue(CsvEnumerator.sameValue(new BigDecimal("1.000000"), new BigDecimal("1"))); // Negative zero edge case - assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("0.0"), new BigDecimal("-0.0"))); + assertTrue(CsvEnumerator.sameValue(new BigDecimal("0.0"), new BigDecimal("-0.0"))); // Very large numbers with scale assertTrue( - CsvEnumerator.objectsEqual( + CsvEnumerator.sameValue( new BigDecimal("999999999.9"), new BigDecimal("999999999.90"))); // Strings - assertTrue(CsvEnumerator.objectsEqual("hello", "hello")); - assertFalse(CsvEnumerator.objectsEqual("hello", "world")); + assertTrue(CsvEnumerator.sameValue("hello", "hello")); + assertFalse(CsvEnumerator.sameValue("hello", "world")); // Integers / Longs - assertTrue(CsvEnumerator.objectsEqual(42, 42)); - assertFalse(CsvEnumerator.objectsEqual(42, 43)); - assertTrue(CsvEnumerator.objectsEqual(1L, 1L)); + assertTrue(CsvEnumerator.sameValue(42, 42)); + assertFalse(CsvEnumerator.sameValue(42, 43)); + assertTrue(CsvEnumerator.sameValue(1L, 1L)); - // Cross-type comparison (implicit type promotion is not handled by objectsEqual, returns false) - assertFalse(CsvEnumerator.objectsEqual(42, 42L)); + // Cross-type comparison (implicit type promotion is not handled by sameValue, returns false) + assertFalse(CsvEnumerator.sameValue(42, 42L)); // BigDecimal scale differences & symmetry BigDecimal val = new BigDecimal("2.0"); - assertTrue(CsvEnumerator.objectsEqual(val, val)); - assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("2.0"), new BigDecimal("2.00"))); - assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("2.00"), new BigDecimal("2.0"))); - assertFalse(CsvEnumerator.objectsEqual(new BigDecimal("1.0"), new BigDecimal("2.0"))); - assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("0.0"), new BigDecimal("0.00"))); - assertTrue(CsvEnumerator.objectsEqual(new BigDecimal("-1.0"), new BigDecimal("-1.00"))); + assertTrue(CsvEnumerator.sameValue(val, val)); + assertTrue(CsvEnumerator.sameValue(new BigDecimal("2.0"), new BigDecimal("2.00"))); + assertTrue(CsvEnumerator.sameValue(new BigDecimal("2.00"), new BigDecimal("2.0"))); + assertFalse(CsvEnumerator.sameValue(new BigDecimal("1.0"), new BigDecimal("2.0"))); + assertTrue(CsvEnumerator.sameValue(new BigDecimal("0.0"), new BigDecimal("0.00"))); + assertTrue(CsvEnumerator.sameValue(new BigDecimal("-1.0"), new BigDecimal("-1.00"))); // Objects.equals() performs exact class/structure comparison (including scale // for BigDecimal), which incorrectly returns false for semantically equal numbers. // Confirm Objects.equals fails here. assertFalse(java.util.Objects.equals(new BigDecimal("2.0"), new BigDecimal("2.00"))); } + + @SuppressWarnings("deprecation") + private static String applyRule(String sql, RelOptRule rule) + throws Exception { + final Properties info = new Properties(); + info.put("model", FileAdapterTests.jsonPath("smart")); + + try (Connection connection = + DriverManager.getConnection("jdbc:calcite:", info)) { + final CalciteConnection calciteConnection = + connection.unwrap(CalciteConnection.class); + final SchemaPlus salesSchema = + calciteConnection.getRootSchema().getSubSchema("SALES"); + + final FrameworkConfig config = Frameworks.newConfigBuilder() + .defaultSchema(salesSchema) + .build(); + + final Planner planner = Frameworks.getPlanner(config); + final SqlNode parsed = planner.parse(sql); + final SqlNode validated = planner.validate(parsed); + final RelNode rel = planner.rel(validated).project(); + + final HepProgramBuilder programBuilder = new HepProgramBuilder(); + programBuilder.addRuleInstance(rule); + + final HepPlanner hepPlanner = + new HepPlanner(programBuilder.build()); + hepPlanner.setRoot(rel); + + return RelOptUtil.toString(hepPlanner.findBestExp()); + } + } + + @Test void testFilterPushDownRule() throws Exception { + final String plan = + applyRule("select * from EMPS where deptno = 20", + FileRules.FILTER_SCAN); + + assertThat(plan, + containsString("CsvTableScan(table=[[SALES, EMPS]], " + + "fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]], " + + "condition=[=($2, 20)])")); + } + + @Test void testProjectFilterPushDownRule() throws Exception { + final String plan = + applyRule("select name, empno from EMPS where deptno = 20", + FileRules.PROJECT_FILTER_SCAN); + + assertThat(plan, + containsString("CsvTableScan(table=[[SALES, EMPS]], " + + "fields=[[0, 1, 2]], condition=[=($2, 20)])")); + } + + @SuppressWarnings("deprecation") + @Test void testFilterProjectTransposeWithProjectFilterScan() throws Exception { + final String sql = "select name from (select name, deptno from EMPS) where deptno = 20"; + + final Properties info = new Properties(); + info.put("model", FileAdapterTests.jsonPath("smart")); + + try (Connection connection = + DriverManager.getConnection("jdbc:calcite:", info)) { + final CalciteConnection calciteConnection = + connection.unwrap(CalciteConnection.class); + final SchemaPlus salesSchema = + calciteConnection.getRootSchema().getSubSchema("SALES"); + + final FrameworkConfig config = Frameworks.newConfigBuilder() + .defaultSchema(salesSchema) + .build(); + + final Planner planner = Frameworks.getPlanner(config); + final SqlNode parsed = planner.parse(sql); + final SqlNode validated = planner.validate(parsed); + final RelNode rel = planner.rel(validated).project(); + + final HepProgramBuilder programBuilder = new HepProgramBuilder(); + programBuilder.addRuleInstance( + org.apache.calcite.rel.rules.CoreRules.FILTER_PROJECT_TRANSPOSE); + programBuilder.addRuleInstance(FileRules.PROJECT_FILTER_SCAN); + + final HepPlanner hepPlanner = + new HepPlanner(programBuilder.build()); + hepPlanner.setRoot(rel); + + final String plan = RelOptUtil.toString(hepPlanner.findBestExp()); + + assertThat(plan, + containsString("CsvTableScan(table=[[SALES, EMPS]], " + + "fields=[[1, 2]], condition=[=($1, 20)])")); + } + } }