diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxMetricsSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxMetricsSuite.scala index 8c858362404..1e8ce70d897 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxMetricsSuite.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxMetricsSuite.scala @@ -227,6 +227,50 @@ class VeloxMetricsSuite extends VeloxWholeStageTransformerSuite with AdaptiveSpa } } + test("Literal-on-left comparison prunes Parquet row groups") { + withTempPath { + dir => + withTempView("sorted_ids") { + spark + .range(0, 10000, 1, 1) + .write + .format("parquet") + .option("parquet.block.rows", "1000") + .save(dir.getCanonicalPath) + spark.read.parquet(dir.getCanonicalPath).createOrReplaceTempView("sorted_ids") + // The file has 10 row groups. Returns (skipped, processed) row groups of the scan. + def strides(query: String): (Long, Long) = { + var result = (0L, 0L) + runQueryAndCompare(query) { + df => + val scan = find(df.queryExecution.executedPlan) { + case _: FileSourceScanExecTransformer => true + case _ => false + } + assert(scan.isDefined) + val metrics = scan.get.metrics + result = (metrics("skippedStrides").value, metrics("processedStrides").value) + } + result + } + // Each literal-on-left predicate must skip the same row groups as its mirrored form. + Seq( + "100 > id" -> "id < 100", + "100 >= id" -> "id <= 100", + "9900 < id" -> "id > 9900", + "9900 <= id" -> "id >= 9900", + "5000 = id" -> "id = 5000" + ).foreach { + case (left, right) => + val expected = strides(s"SELECT id FROM sorted_ids WHERE $right") + val actual = strides(s"SELECT id FROM sorted_ids WHERE $left") + assert(expected._1 > 0, s"$right should prune row groups: $expected") + assert(actual == expected, s"$left vs $right") + } + } + } + } + test("Metrics of noop filter's children") { runQueryAndCompare("SELECT c1, c2 FROM metrics_t1 where c1 < 50") { df => diff --git a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc index 31d44211cd5..49e70c249bd 100644 --- a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc +++ b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc @@ -16,6 +16,8 @@ */ #include "operators/functions/SparkExprToSubfieldFilterParser.h" +#include + #include "utils/Exception.h" #include "velox/common/base/BloomFilter.h" #include "velox/expression/Expr.h" @@ -117,6 +119,21 @@ SparkExprToSubfieldFilterParser::leafCallToSubfieldFilter( const auto* leftSide = call.inputs()[0].get(); common::Subfield subfield; + // Spark keeps a literal-on-left comparison such as `5 < col` as written. Swap the operands + // and mirror the operator so it is handled like `col > 5`. + static const std::unordered_map kMirroredComparisons = { + {"equalto", "equalto"}, + {"lessthan", "greaterthan"}, + {"lessthanorequal", "greaterthanorequal"}, + {"greaterthan", "lessthan"}, + {"greaterthanorequal", "lessthanorequal"}}; + if (call.inputs().size() == 2 && !toSubfield(leftSide, subfield) && toSubfield(call.inputs()[1].get(), subfield)) { + auto it = kMirroredComparisons.find(call.name()); + if (it != kMirroredComparisons.end()) { + const core::CallTypedExpr swapped(call.type(), {call.inputs()[1], call.inputs()[0]}, it->second); + return leafCallToSubfieldFilter(swapped, evaluator, negated); + } + } if (call.name() == "equalto") { if (toSubfield(leftSide, subfield)) { auto filter =