From 6042822201d8520e7848763bfe4bf4171383dd44 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 14:19:04 -0600 Subject: [PATCH 01/28] start to define function signatures --- .../scala/org/apache/comet/fuzz/Meta.scala | 222 ++++++++++++------ .../org/apache/comet/fuzz/QueryGen.scala | 6 +- .../comet/CometArrayExpressionSuite.scala | 26 ++ 3 files changed, 173 insertions(+), 81 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 246216840b7..7aab7e5b5ae 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -22,6 +22,45 @@ package org.apache.comet.fuzz import org.apache.spark.sql.types.DataType import org.apache.spark.sql.types.DataTypes +sealed trait InputType +case class SparkType(dataType: DataType) extends InputType +case class SparkTypeOneOf(dataTypes: Seq[InputType]) extends InputType +case object SparkBinaryType extends InputType +case object SparkStringType extends InputType +case object SparkIntegralType extends InputType +case object SparkByteType extends InputType +case object SparkShortType extends InputType +case object SparkIntType extends InputType +case object SparkLongType extends InputType +case object SparkFloatType extends InputType +case object SparkDoubleType extends InputType +case class SparkDecimalType(p: Int, s: Int) extends InputType +case object SparkNumericType extends InputType +case object SparkAnyType extends InputType + +case class FunctionSignature(inputTypes: Seq[InputType]) + +sealed trait Function { + + def name: String + + // query generator should generate types based on signature not just on arg count + @deprecated + def numArgs: Int +} + +@deprecated +case class FunctionWithArgCount(name: String, argCount: Int) extends Function { + // query generator should generate types based on signature not just on arg count + override def numArgs: Int = argCount +} + +case class FunctionWithSignature(name: String, signatures: Seq[FunctionSignature]) + extends Function { + // query generator should generate types based on signature not just on arg count + override def numArgs: Int = signatures.head.inputTypes.length +} + object Meta { val dataTypes: Seq[(DataType, Double)] = Seq( @@ -35,100 +74,129 @@ object Meta { (DataTypes.createDecimalType(10, 2), 0.2), (DataTypes.DateType, 0.2), (DataTypes.TimestampType, 0.2), - // TimestampNTZType only in Spark 3.4+ - // (DataTypes.TimestampNTZType, 0.2), + (DataTypes.TimestampNTZType, 0.2), (DataTypes.StringType, 0.2), (DataTypes.BinaryType, 0.1)) + @deprecated + private def createFunction(name: String, argCount: Int): FunctionWithArgCount = { + FunctionWithArgCount(name, argCount) + } + + private def createFunctionWithInputs( + name: String, + inputs: Seq[InputType]): FunctionWithSignature = { + FunctionWithSignature(name, Seq(FunctionSignature(inputs))) + } + + private def createFunctionWithSignatures( + name: String, + signatures: Seq[FunctionSignature]): FunctionWithSignature = { + FunctionWithSignature(name, signatures) + } + + private def createUnaryStringFunction(name: String): FunctionWithSignature = { + createFunctionWithInputs(name, Seq(SparkStringType)) + } + val stringScalarFunc: Seq[Function] = Seq( - Function("substring", 3), - Function("coalesce", 1), - Function("starts_with", 2), - Function("ends_with", 2), - Function("contains", 2), - Function("ascii", 1), - Function("bit_length", 1), - Function("octet_length", 1), - Function("upper", 1), - Function("lower", 1), - Function("chr", 1), - Function("init_cap", 1), - Function("trim", 1), - Function("ltrim", 1), - Function("rtrim", 1), - Function("string_space", 1), - Function("rpad", 2), - Function("rpad", 3), // rpad can have 2 or 3 arguments - Function("hex", 1), - Function("unhex", 1), - Function("xxhash64", 1), - Function("sha1", 1), - // Function("sha2", 1), -- needs a second argument for number of bits - Function("substring", 3), - Function("btrim", 1), - Function("concat_ws", 2), - Function("repeat", 2), - Function("length", 1), - Function("reverse", 1), - Function("instr", 2), - Function("replace", 2), - Function("translate", 2)) + createFunction("substring", 3), + createUnaryStringFunction("coalesce"), + createFunctionWithInputs("starts_with", Seq(SparkStringType, SparkStringType)), + createFunction("ends_with", 2), + createFunction("contains", 2), + createUnaryStringFunction("ascii"), + createUnaryStringFunction("bit_length"), + createUnaryStringFunction("octet_length"), + createUnaryStringFunction("upper"), + createUnaryStringFunction("lower"), + createUnaryStringFunction("chr"), + createUnaryStringFunction("init_cap"), + createUnaryStringFunction("trim"), + createUnaryStringFunction("ltrim"), + createUnaryStringFunction("rtrim"), + createFunction("string_space", 1), + createFunctionWithSignatures( + "rpad", + Seq( + FunctionSignature(Seq(SparkStringType, SparkIntegralType)), + FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), + createFunctionWithInputs( + "hex", + Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType, SparkIntType, SparkLongType)))), + createFunction("unhex", 1), + createFunction("xxhash64", 1), + createFunction("sha1", 1), + // createFunction("sha2", 1), -- needs a second argument for number of bits + createFunction("substring", 3), + createFunction("btrim", 1), + createFunction("concat_ws", 2), + createFunction("repeat", 2), + createFunction("length", 1), + createFunction("reverse", 1), + createFunction("instr", 2), + createFunction("replace", 2), + createFunction("translate", 2)) val dateScalarFunc: Seq[Function] = - Seq(Function("year", 1), Function("hour", 1), Function("minute", 1), Function("second", 1)) + Seq( + createFunction("year", 1), + createFunction("hour", 1), + createFunction("minute", 1), + createFunction("second", 1)) val mathScalarFunc: Seq[Function] = Seq( - Function("abs", 1), - Function("acos", 1), - Function("asin", 1), - Function("atan", 1), - Function("Atan2", 1), - Function("Cos", 1), - Function("Exp", 2), - Function("Ln", 1), - Function("Log10", 1), - Function("Log2", 1), - Function("Pow", 2), - Function("Round", 1), - Function("Signum", 1), - Function("Sin", 1), - Function("Sqrt", 1), - Function("Tan", 1), - Function("Ceil", 1), - Function("Floor", 1), - Function("bool_and", 1), - Function("bool_or", 1), - Function("bitwise_not", 1)) + createFunction("abs", 1), + createFunction("acos", 1), + createFunction("asin", 1), + createFunction("atan", 1), + createFunction("Atan2", 1), + createFunction("Cos", 1), + createFunction("Exp", 2), + createFunction("Ln", 1), + createFunction("Log10", 1), + createFunction("Log2", 1), + createFunction("Pow", 2), + createFunction("Round", 1), + createFunction("Signum", 1), + createFunction("Sin", 1), + createFunction("Sqrt", 1), + createFunction("Tan", 1), + createFunction("Ceil", 1), + createFunction("Floor", 1), + createFunction("bool_and", 1), + createFunction("bool_or", 1), + createFunction("bitwise_not", 1)) val miscScalarFunc: Seq[Function] = - Seq(Function("isnan", 1), Function("isnull", 1), Function("isnotnull", 1)) + Seq(createFunction("isnan", 1), createFunction("isnull", 1), createFunction("isnotnull", 1)) val arrayScalarFunc: Seq[Function] = Seq( - Function("array", 2), - Function("array_remove", 2), - Function("array_insert", 2), - Function("array_contains", 2), - Function("array_intersect", 2), - Function("array_append", 2)) + createFunction("array", 2), + createFunction("array_remove", 2), + createFunction("array_insert", 2), + createFunction("array_contains", 2), + createFunction("array_intersect", 2), + createFunction("array_append", 2)) val scalarFunc: Seq[Function] = stringScalarFunc ++ dateScalarFunc ++ mathScalarFunc ++ miscScalarFunc ++ arrayScalarFunc val aggFunc: Seq[Function] = Seq( - Function("min", 1), - Function("max", 1), - Function("count", 1), - Function("avg", 1), - Function("sum", 1), - Function("first", 1), - Function("last", 1), - Function("var_pop", 1), - Function("var_samp", 1), - Function("covar_pop", 1), - Function("covar_samp", 1), - Function("stddev_pop", 1), - Function("stddev_samp", 1), - Function("corr", 2)) + createFunction("min", 1), + createFunction("max", 1), + createFunction("count", 1), + createFunction("avg", 1), + createFunction("sum", 1), + createFunction("first", 1), + createFunction("last", 1), + createFunction("var_pop", 1), + createFunction("var_samp", 1), + createFunction("covar_pop", 1), + createFunction("covar_samp", 1), + createFunction("stddev_pop", 1), + createFunction("stddev_samp", 1), + createFunction("corr", 2)) val unaryArithmeticOps: Seq[String] = Seq("+", "-") diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index de1117837ea..427716ed76f 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -65,7 +65,7 @@ object QueryGen { val table = spark.table(tableName) val func = Utils.randomChoice(Meta.aggFunc, r) - val args = Range(0, func.num_args) + val args = Range(0, func.numArgs) .map(_ => Utils.randomChoice(table.columns, r)) val groupingCols = Range(0, 2).map(_ => Utils.randomChoice(table.columns, r)) @@ -87,7 +87,7 @@ object QueryGen { val table = spark.table(tableName) val func = Utils.randomChoice(Meta.scalarFunc, r) - val args = Range(0, func.num_args) + val args = Range(0, func.numArgs) .map(_ => Utils.randomChoice(table.columns, r)) // Example SELECT c0, log(c0) as x FROM test0 @@ -192,5 +192,3 @@ object QueryGen { } } - -case class Function(name: String, num_args: Int) diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 2adb7a9ed61..1647e8d985c 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -26,6 +26,7 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions._ +import org.apache.spark.sql.types.ArrayType import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus} import org.apache.comet.DataTypeSupport.isComplexType @@ -768,6 +769,31 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } test("array_reverse") { + withTempDir { dir => + val path = new Path(dir.toURI.toString, "test.parquet") + val filename = path.toString + val random = new Random(42) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val options = DataGenOptions( + allowNull = true, + generateNegativeZero = false, + generateArray = true, + generateStruct = false, + generateMap = false) + ParquetGenerator.makeParquetFile(random, spark, filename, 100, options) + } + withTempView("t1") { + val table = spark.read.parquet(filename) + table.createOrReplaceTempView("t1") + for (field <- table.schema.fields.filter(_.dataType.isInstanceOf[ArrayType])) { + val sql = s"SELECT ${field.name}, reverse(${field.name}) FROM t1 ORDER BY ${field.name}" + checkSparkAnswer(sql) + } + } + } + } + + test("array_reverse no native scan") { withTempDir { dir => val path = new Path(dir.toURI.toString, "test.parquet") val filename = path.toString From e41325b82c7cc9e2edcd40ec654a29f9d96a1215 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 14:37:07 -0600 Subject: [PATCH 02/28] update more signatures --- .../scala/org/apache/comet/fuzz/Meta.scala | 127 ++++++++++-------- 1 file changed, 72 insertions(+), 55 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 7aab7e5b5ae..fd85ab8a919 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -22,23 +22,26 @@ package org.apache.comet.fuzz import org.apache.spark.sql.types.DataType import org.apache.spark.sql.types.DataTypes -sealed trait InputType -case class SparkType(dataType: DataType) extends InputType -case class SparkTypeOneOf(dataTypes: Seq[InputType]) extends InputType -case object SparkBinaryType extends InputType -case object SparkStringType extends InputType -case object SparkIntegralType extends InputType -case object SparkByteType extends InputType -case object SparkShortType extends InputType -case object SparkIntType extends InputType -case object SparkLongType extends InputType -case object SparkFloatType extends InputType -case object SparkDoubleType extends InputType -case class SparkDecimalType(p: Int, s: Int) extends InputType -case object SparkNumericType extends InputType -case object SparkAnyType extends InputType - -case class FunctionSignature(inputTypes: Seq[InputType]) +sealed trait SparkType +case class SparkTypeOneOf(dataTypes: Seq[SparkType]) extends SparkType +case object SparkBinaryType extends SparkType +case object SparkStringType extends SparkType +case object SparkIntegralType extends SparkType +case object SparkByteType extends SparkType +case object SparkShortType extends SparkType +case object SparkIntType extends SparkType +case object SparkLongType extends SparkType +case object SparkFloatType extends SparkType +case object SparkDoubleType extends SparkType +case class SparkDecimalType(p: Int, s: Int) extends SparkType +case object SparkNumericType extends SparkType +case object SparkDateType extends SparkType +case object SparkTimestampType extends SparkType +case object SparkDateOrTimestampType extends SparkType +case class SparkArrayType(elementType: SparkType) extends SparkType +case object SparkAnyType extends SparkType + +case class FunctionSignature(inputTypes: Seq[SparkType]) sealed trait Function { @@ -85,7 +88,7 @@ object Meta { private def createFunctionWithInputs( name: String, - inputs: Seq[InputType]): FunctionWithSignature = { + inputs: Seq[SparkType]): FunctionWithSignature = { FunctionWithSignature(name, Seq(FunctionSignature(inputs))) } @@ -99,12 +102,16 @@ object Meta { createFunctionWithInputs(name, Seq(SparkStringType)) } + private def createUnaryNumericFunction(name: String): FunctionWithSignature = { + createFunctionWithInputs(name, Seq(SparkNumericType)) + } + val stringScalarFunc: Seq[Function] = Seq( - createFunction("substring", 3), + createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), createUnaryStringFunction("coalesce"), createFunctionWithInputs("starts_with", Seq(SparkStringType, SparkStringType)), - createFunction("ends_with", 2), - createFunction("contains", 2), + createFunctionWithInputs("ends_with", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs("contains", Seq(SparkStringType, SparkStringType)), createUnaryStringFunction("ascii"), createUnaryStringFunction("bit_length"), createUnaryStringFunction("octet_length"), @@ -124,46 +131,56 @@ object Meta { createFunctionWithInputs( "hex", Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType, SparkIntType, SparkLongType)))), - createFunction("unhex", 1), - createFunction("xxhash64", 1), - createFunction("sha1", 1), + createUnaryStringFunction("unhex"), + createFunctionWithInputs("xxhash64", Seq(SparkAnyType)), // TODO can take multiple columns + createFunctionWithInputs("sha1", Seq(SparkAnyType)), // createFunction("sha2", 1), -- needs a second argument for number of bits - createFunction("substring", 3), - createFunction("btrim", 1), + createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), + createUnaryStringFunction("btrim"), createFunction("concat_ws", 2), createFunction("repeat", 2), - createFunction("length", 1), - createFunction("reverse", 1), - createFunction("instr", 2), - createFunction("replace", 2), - createFunction("translate", 2)) + createFunctionWithInputs( + "length", + Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType)))), + createFunctionWithSignatures( + "reverse", + Seq( + FunctionSignature(Seq(SparkStringType)), + FunctionSignature(Seq(SparkArrayType(SparkAnyType))))), + createFunctionWithInputs("instr", Seq(SparkStringType, SparkStringType)), + createFunctionWithSignatures( + "replace", + Seq( + FunctionSignature(Seq(SparkStringType, SparkStringType)), + FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)))), + createFunctionWithInputs("translate", Seq(SparkStringType, SparkStringType))) val dateScalarFunc: Seq[Function] = Seq( - createFunction("year", 1), - createFunction("hour", 1), - createFunction("minute", 1), - createFunction("second", 1)) + createFunctionWithInputs("year", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("hour", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("minute", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("second", Seq(SparkDateOrTimestampType))) val mathScalarFunc: Seq[Function] = Seq( - createFunction("abs", 1), - createFunction("acos", 1), - createFunction("asin", 1), - createFunction("atan", 1), - createFunction("Atan2", 1), - createFunction("Cos", 1), - createFunction("Exp", 2), - createFunction("Ln", 1), - createFunction("Log10", 1), - createFunction("Log2", 1), - createFunction("Pow", 2), - createFunction("Round", 1), - createFunction("Signum", 1), - createFunction("Sin", 1), - createFunction("Sqrt", 1), - createFunction("Tan", 1), - createFunction("Ceil", 1), - createFunction("Floor", 1), + createUnaryNumericFunction("abs"), + createUnaryNumericFunction("acos"), + createUnaryNumericFunction("asin"), + createUnaryNumericFunction("atan"), + createUnaryNumericFunction("Atan2"), + createUnaryNumericFunction("Cos"), + createFunctionWithInputs("Exp", Seq(SparkNumericType, SparkNumericType)), + createUnaryNumericFunction("Ln"), + createUnaryNumericFunction("Log10"), + createUnaryNumericFunction("Log2"), + createFunctionWithInputs("Pow", Seq(SparkNumericType, SparkNumericType)), + createUnaryNumericFunction("Round"), + createUnaryNumericFunction("Signum"), + createUnaryNumericFunction("Sin"), + createUnaryNumericFunction("Sqrt"), + createUnaryNumericFunction("Tan"), + createUnaryNumericFunction("Ceil"), + createUnaryNumericFunction("Floor"), createFunction("bool_and", 1), createFunction("bool_or", 1), createFunction("bitwise_not", 1)) @@ -186,8 +203,8 @@ object Meta { createFunction("min", 1), createFunction("max", 1), createFunction("count", 1), - createFunction("avg", 1), - createFunction("sum", 1), + createUnaryNumericFunction("avg"), + createUnaryNumericFunction("sum"), createFunction("first", 1), createFunction("last", 1), createFunction("var_pop", 1), From 166f1393d3e6b039c8cc9788e81a560a69540c65 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 14:39:25 -0600 Subject: [PATCH 03/28] update more signatures --- .../src/main/scala/org/apache/comet/fuzz/Meta.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index fd85ab8a919..4e3e54672ce 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -47,20 +47,20 @@ sealed trait Function { def name: String - // query generator should generate types based on signature not just on arg count + // query generator should choose inputs based on signature not just on arg count @deprecated def numArgs: Int } @deprecated case class FunctionWithArgCount(name: String, argCount: Int) extends Function { - // query generator should generate types based on signature not just on arg count + // query generator should choose inputs based on signature not just on arg count override def numArgs: Int = argCount } case class FunctionWithSignature(name: String, signatures: Seq[FunctionSignature]) extends Function { - // query generator should generate types based on signature not just on arg count + // query generator should choose inputs based on signature not just on arg count override def numArgs: Int = signatures.head.inputTypes.length } From 0db4999dfdf81b716121d44db5b3c438acbceb60 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 14:40:28 -0600 Subject: [PATCH 04/28] update more signatures --- .../comet/CometArrayExpressionSuite.scala | 25 ------------------- 1 file changed, 25 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 1647e8d985c..ec644f383b7 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -768,31 +768,6 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } - test("array_reverse") { - withTempDir { dir => - val path = new Path(dir.toURI.toString, "test.parquet") - val filename = path.toString - val random = new Random(42) - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - val options = DataGenOptions( - allowNull = true, - generateNegativeZero = false, - generateArray = true, - generateStruct = false, - generateMap = false) - ParquetGenerator.makeParquetFile(random, spark, filename, 100, options) - } - withTempView("t1") { - val table = spark.read.parquet(filename) - table.createOrReplaceTempView("t1") - for (field <- table.schema.fields.filter(_.dataType.isInstanceOf[ArrayType])) { - val sql = s"SELECT ${field.name}, reverse(${field.name}) FROM t1 ORDER BY ${field.name}" - checkSparkAnswer(sql) - } - } - } - } - test("array_reverse no native scan") { withTempDir { dir => val path = new Path(dir.toURI.toString, "test.parquet") From 984ceef026465452748d1513b3f74318c4f3d0bf Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 14:41:22 -0600 Subject: [PATCH 05/28] update more signatures --- .../scala/org/apache/comet/CometArrayExpressionSuite.scala | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index ec644f383b7..2adb7a9ed61 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -26,7 +26,6 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions._ -import org.apache.spark.sql.types.ArrayType import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus} import org.apache.comet.DataTypeSupport.isComplexType @@ -768,7 +767,7 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } - test("array_reverse no native scan") { + test("array_reverse") { withTempDir { dir => val path = new Path(dir.toURI.toString, "test.parquet") val filename = path.toString From b0878390a4ae932d5d5534f9ca37ba3c7460459f Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 14:54:39 -0600 Subject: [PATCH 06/28] test --- .../org/apache/comet/fuzz/QueryGen.scala | 41 ++++++++++++++++--- 1 file changed, 36 insertions(+), 5 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index 427716ed76f..1c37532135e 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -24,7 +24,8 @@ import java.io.{BufferedWriter, FileWriter} import scala.collection.mutable import scala.util.Random -import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.{DataFrame, SparkSession} +import org.apache.spark.sql.types._ object QueryGen { @@ -86,16 +87,46 @@ object QueryGen { val tableName = s"test${r.nextInt(numFiles)}" val table = spark.table(tableName) - val func = Utils.randomChoice(Meta.scalarFunc, r) - val args = Range(0, func.numArgs) - .map(_ => Utils.randomChoice(table.columns, r)) + val f = Utils.randomChoice(Meta.scalarFunc, r) + val args = f match { + case func: FunctionWithArgCount => + Range(0, func.numArgs).map(_ => Utils.randomChoice(table.columns, r)) + case func: FunctionWithSignature => + val signature = Utils.randomChoice(func.signatures, r) + signature.inputTypes.map(x => pickRandomColumn(r, table, x)) + } // Example SELECT c0, log(c0) as x FROM test0 - s"SELECT ${args.mkString(", ")}, ${func.name}(${args.mkString(", ")}) AS x " + + s"SELECT ${args.mkString(", ")}, ${f.name}(${args.mkString(", ")}) AS x " + s"FROM $tableName " + s"ORDER BY ${args.mkString(", ")};" } + private def pickRandomColumn(r: Random, df: DataFrame, targetType: SparkType): String = { + targetType match { + case SparkByteType => + val candidates = df.schema.fields.filter(_.dataType == ByteType) + Utils.randomChoice(candidates, r).name + case SparkShortType => + val candidates = df.schema.fields.filter(_.dataType == ShortType) + Utils.randomChoice(candidates, r).name + case SparkIntType => + val candidates = df.schema.fields.filter(_.dataType == IntegerType) + Utils.randomChoice(candidates, r).name + case SparkLongType => + val candidates = df.schema.fields.filter(_.dataType == LongType) + Utils.randomChoice(candidates, r).name + case SparkStringType => + val candidates = df.schema.fields.filter(_.dataType == StringType) + Utils.randomChoice(candidates, r).name + case SparkTypeOneOf(choices) => + pickRandomColumn(r, df, Utils.randomChoice(choices, r)) + case _ => + throw new IllegalStateException(targetType.toString) + } + + } + private def generateUnaryArithmetic(r: Random, spark: SparkSession, numFiles: Int): String = { val tableName = s"test${r.nextInt(numFiles)}" val table = spark.table(tableName) From 4bf693924b140f1adc8665680181a538d06bc929 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 17:41:30 -0600 Subject: [PATCH 07/28] save progress --- fuzz-testing/README.md | 6 +++--- .../main/scala/org/apache/comet/fuzz/QueryGen.scala | 10 ++++++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/fuzz-testing/README.md b/fuzz-testing/README.md index 17b2c151a24..e9e28e17390 100644 --- a/fuzz-testing/README.md +++ b/fuzz-testing/README.md @@ -61,7 +61,7 @@ Set appropriate values for `SPARK_HOME`, `SPARK_MASTER`, and `COMET_JAR` environ $SPARK_HOME/bin/spark-submit \ --master $SPARK_MASTER \ --class org.apache.comet.fuzz.Main \ - target/comet-fuzz-spark3.4_2.12-0.7.0-SNAPSHOT-jar-with-dependencies.jar \ + target/comet-fuzz-spark3.4_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ data --num-files=2 --num-rows=200 --exclude-negative-zero --generate-arrays --generate-structs --generate-maps ``` @@ -77,7 +77,7 @@ Generate random queries that are based on the available test files. $SPARK_HOME/bin/spark-submit \ --master $SPARK_MASTER \ --class org.apache.comet.fuzz.Main \ - target/comet-fuzz-spark3.4_2.12-0.7.0-SNAPSHOT-jar-with-dependencies.jar \ + target/comet-fuzz-spark3.4_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ queries --num-files=2 --num-queries=500 ``` @@ -99,7 +99,7 @@ $SPARK_HOME/bin/spark-submit \ --conf spark.driver.extraClassPath=$COMET_JAR \ --conf spark.executor.extraClassPath=$COMET_JAR \ --class org.apache.comet.fuzz.Main \ - target/comet-fuzz-spark3.4_2.12-0.7.0-SNAPSHOT-jar-with-dependencies.jar \ + target/comet-fuzz-spark3.4_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ run --num-files=2 --filename=queries.sql ``` diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index 1c37532135e..a4ecf57f7a6 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -121,12 +121,22 @@ object QueryGen { Utils.randomChoice(candidates, r).name case SparkTypeOneOf(choices) => pickRandomColumn(r, df, Utils.randomChoice(choices, r)) + case SparkArrayType(elementType) => + val candidates = df.schema.fields.filter(_.dataType match { + case ArrayType(x, _) if typeMatch(elementType, x) => true + case _ => false + }) + Utils.randomChoice(candidates, r).name case _ => throw new IllegalStateException(targetType.toString) } } + private def typeMatch(s: SparkType, d: DataType): Boolean = { + false + } + private def generateUnaryArithmetic(r: Random, spark: SparkSession, numFiles: Int): String = { val tableName = s"test${r.nextInt(numFiles)}" val table = spark.table(tableName) From 5e20dd8213b54bae8c172978ce31c0f2aec06c29 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 17:55:30 -0600 Subject: [PATCH 08/28] save [skip ci] --- fuzz-testing/README.md | 6 +++--- .../main/scala/org/apache/comet/fuzz/QueryGen.scala | 13 +++++++++++++ 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/fuzz-testing/README.md b/fuzz-testing/README.md index e9e28e17390..bde1505e3e3 100644 --- a/fuzz-testing/README.md +++ b/fuzz-testing/README.md @@ -61,7 +61,7 @@ Set appropriate values for `SPARK_HOME`, `SPARK_MASTER`, and `COMET_JAR` environ $SPARK_HOME/bin/spark-submit \ --master $SPARK_MASTER \ --class org.apache.comet.fuzz.Main \ - target/comet-fuzz-spark3.4_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ + target/comet-fuzz-spark3.5_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ data --num-files=2 --num-rows=200 --exclude-negative-zero --generate-arrays --generate-structs --generate-maps ``` @@ -77,7 +77,7 @@ Generate random queries that are based on the available test files. $SPARK_HOME/bin/spark-submit \ --master $SPARK_MASTER \ --class org.apache.comet.fuzz.Main \ - target/comet-fuzz-spark3.4_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ + target/comet-fuzz-spark3.5_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ queries --num-files=2 --num-queries=500 ``` @@ -99,7 +99,7 @@ $SPARK_HOME/bin/spark-submit \ --conf spark.driver.extraClassPath=$COMET_JAR \ --conf spark.executor.extraClassPath=$COMET_JAR \ --class org.apache.comet.fuzz.Main \ - target/comet-fuzz-spark3.4_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ + target/comet-fuzz-spark3.5_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \ run --num-files=2 --filename=queries.sql ``` diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index a4ecf57f7a6..7592cca78db 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -116,6 +116,9 @@ object QueryGen { case SparkLongType => val candidates = df.schema.fields.filter(_.dataType == LongType) Utils.randomChoice(candidates, r).name + case SparkNumericType => + val candidates = df.schema.fields.filter(f => isNumeric(f.dataType)) + Utils.randomChoice(candidates, r).name case SparkStringType => val candidates = df.schema.fields.filter(_.dataType == StringType) Utils.randomChoice(candidates, r).name @@ -133,7 +136,17 @@ object QueryGen { } + private def isNumeric(d: DataType): Boolean = { + d match { + case _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: FloatType | + _: DoubleType | _: DecimalType => + true + case _ => false + } + } + private def typeMatch(s: SparkType, d: DataType): Boolean = { + // TODO false } From 73c918588757b82aea10cc50d71be480ec69ff54 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 20 Oct 2025 18:22:56 -0600 Subject: [PATCH 09/28] save [skip ci] --- .../org/apache/comet/fuzz/QueryGen.scala | 44 ++++++++++++------- 1 file changed, 27 insertions(+), 17 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index 7592cca78db..5e0a83bbc2f 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -104,36 +104,46 @@ object QueryGen { private def pickRandomColumn(r: Random, df: DataFrame, targetType: SparkType): String = { targetType match { + case SparkAnyType => + Utils.randomChoice(df.schema.fields, r).name case SparkByteType => - val candidates = df.schema.fields.filter(_.dataType == ByteType) - Utils.randomChoice(candidates, r).name + select(r, df, _.dataType == ByteType) case SparkShortType => - val candidates = df.schema.fields.filter(_.dataType == ShortType) - Utils.randomChoice(candidates, r).name + select(r, df, _.dataType == ShortType) case SparkIntType => - val candidates = df.schema.fields.filter(_.dataType == IntegerType) - Utils.randomChoice(candidates, r).name + select(r, df, _.dataType == IntegerType) case SparkLongType => - val candidates = df.schema.fields.filter(_.dataType == LongType) - Utils.randomChoice(candidates, r).name + select(r, df, _.dataType == LongType) case SparkNumericType => - val candidates = df.schema.fields.filter(f => isNumeric(f.dataType)) - Utils.randomChoice(candidates, r).name + select(r, df, f => isNumeric(f.dataType)) case SparkStringType => - val candidates = df.schema.fields.filter(_.dataType == StringType) - Utils.randomChoice(candidates, r).name + select(r, df, _.dataType == StringType) + case SparkBinaryType => + select(r, df, _.dataType == BinaryType) + case SparkDateType => + select(r, df, _.dataType == DateType) + case SparkTimestampType => + select(r, df, _.dataType == TimestampType) + case SparkDateOrTimestampType => + select(r, df, f => f.dataType == DateType || f.dataType == TimestampType) case SparkTypeOneOf(choices) => pickRandomColumn(r, df, Utils.randomChoice(choices, r)) case SparkArrayType(elementType) => - val candidates = df.schema.fields.filter(_.dataType match { - case ArrayType(x, _) if typeMatch(elementType, x) => true - case _ => false - }) - Utils.randomChoice(candidates, r).name + select( + r, + df, + _.dataType match { + case ArrayType(x, _) if typeMatch(elementType, x) => true + case _ => false + }) case _ => throw new IllegalStateException(targetType.toString) } + } + /** Select a random field that matches a predicate */ + private def select(r: Random, df: DataFrame, predicate: StructField => Boolean): String = { + Utils.randomChoice(df.schema.fields.filter(predicate), r).name } private def isNumeric(d: DataType): Boolean = { From af4ccf8979491dc0fb3be7e7b78d308fc80f13ce Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 08:44:03 -0600 Subject: [PATCH 10/28] update more signatures --- .../scala/org/apache/comet/fuzz/Meta.scala | 66 +++++++++---------- .../org/apache/comet/fuzz/QueryGen.scala | 2 - 2 files changed, 30 insertions(+), 38 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 4e3e54672ce..016acd57b0b 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -52,12 +52,6 @@ sealed trait Function { def numArgs: Int } -@deprecated -case class FunctionWithArgCount(name: String, argCount: Int) extends Function { - // query generator should choose inputs based on signature not just on arg count - override def numArgs: Int = argCount -} - case class FunctionWithSignature(name: String, signatures: Seq[FunctionSignature]) extends Function { // query generator should choose inputs based on signature not just on arg count @@ -81,11 +75,6 @@ object Meta { (DataTypes.StringType, 0.2), (DataTypes.BinaryType, 0.1)) - @deprecated - private def createFunction(name: String, argCount: Int): FunctionWithArgCount = { - FunctionWithArgCount(name, argCount) - } - private def createFunctionWithInputs( name: String, inputs: Seq[SparkType]): FunctionWithSignature = { @@ -122,7 +111,7 @@ object Meta { createUnaryStringFunction("trim"), createUnaryStringFunction("ltrim"), createUnaryStringFunction("rtrim"), - createFunction("string_space", 1), + createFunctionWithInputs("string_space", Seq(SparkIntType)), createFunctionWithSignatures( "rpad", Seq( @@ -137,8 +126,8 @@ object Meta { // createFunction("sha2", 1), -- needs a second argument for number of bits createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), createUnaryStringFunction("btrim"), - createFunction("concat_ws", 2), - createFunction("repeat", 2), + createFunctionWithInputs("concat_ws", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs("repeat", Seq(SparkStringType, SparkIntType)), createFunctionWithInputs( "length", Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType)))), @@ -181,39 +170,44 @@ object Meta { createUnaryNumericFunction("Tan"), createUnaryNumericFunction("Ceil"), createUnaryNumericFunction("Floor"), - createFunction("bool_and", 1), - createFunction("bool_or", 1), - createFunction("bitwise_not", 1)) + createFunctionWithInputs("bool_and", Seq(SparkAnyType)), + createFunctionWithInputs("bool_or", Seq(SparkAnyType)), + createFunctionWithInputs("bitwise_not", Seq(SparkIntegralType))) val miscScalarFunc: Seq[Function] = - Seq(createFunction("isnan", 1), createFunction("isnull", 1), createFunction("isnotnull", 1)) + Seq( + createFunctionWithInputs("isnan", Seq(SparkNumericType)), + createFunctionWithInputs("isnull", Seq(SparkAnyType)), + createFunctionWithInputs("isnotnull", Seq(SparkAnyType))) val arrayScalarFunc: Seq[Function] = Seq( - createFunction("array", 2), - createFunction("array_remove", 2), - createFunction("array_insert", 2), - createFunction("array_contains", 2), - createFunction("array_intersect", 2), - createFunction("array_append", 2)) + createFunctionWithInputs("array", Seq(SparkAnyType, SparkAnyType)), + createFunctionWithInputs("array_remove", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputs("array_insert", Seq(SparkArrayType(SparkAnyType), SparkIntType)), + createFunctionWithInputs("array_contains", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputs( + "array_intersect", + Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), + createFunctionWithInputs("array_append", Seq(SparkArrayType(SparkAnyType), SparkAnyType))) val scalarFunc: Seq[Function] = stringScalarFunc ++ dateScalarFunc ++ mathScalarFunc ++ miscScalarFunc ++ arrayScalarFunc val aggFunc: Seq[Function] = Seq( - createFunction("min", 1), - createFunction("max", 1), - createFunction("count", 1), + createFunctionWithInputs("min", Seq(SparkAnyType)), + createFunctionWithInputs("max", Seq(SparkAnyType)), + createFunctionWithInputs("count", Seq(SparkAnyType)), createUnaryNumericFunction("avg"), createUnaryNumericFunction("sum"), - createFunction("first", 1), - createFunction("last", 1), - createFunction("var_pop", 1), - createFunction("var_samp", 1), - createFunction("covar_pop", 1), - createFunction("covar_samp", 1), - createFunction("stddev_pop", 1), - createFunction("stddev_samp", 1), - createFunction("corr", 2)) + createFunctionWithInputs("first", Seq(SparkAnyType)), + createFunctionWithInputs("last", Seq(SparkAnyType)), + createFunctionWithInputs("var_pop", Seq(SparkNumericType)), + createFunctionWithInputs("var_samp", Seq(SparkNumericType)), + createFunctionWithInputs("covar_pop", Seq(SparkNumericType)), + createFunctionWithInputs("covar_samp", Seq(SparkNumericType)), + createFunctionWithInputs("stddev_pop", Seq(SparkNumericType)), + createFunctionWithInputs("stddev_samp", Seq(SparkNumericType)), + createFunctionWithInputs("corr", Seq(SparkNumericType, SparkNumericType))) val unaryArithmeticOps: Seq[String] = Seq("+", "-") diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index 5e0a83bbc2f..08a11ea643f 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -89,8 +89,6 @@ object QueryGen { val f = Utils.randomChoice(Meta.scalarFunc, r) val args = f match { - case func: FunctionWithArgCount => - Range(0, func.numArgs).map(_ => Utils.randomChoice(table.columns, r)) case func: FunctionWithSignature => val signature = Utils.randomChoice(func.signatures, r) signature.inputTypes.map(x => pickRandomColumn(r, table, x)) From 64d40abfaa83c6fa7330661dbd45c980e6d04c22 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 09:07:23 -0600 Subject: [PATCH 11/28] convert remaining signatures --- .../scala/org/apache/comet/fuzz/Meta.scala | 35 ++++++------------- .../org/apache/comet/fuzz/QueryGen.scala | 11 +++--- 2 files changed, 15 insertions(+), 31 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 016acd57b0b..781c62cae46 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -43,19 +43,10 @@ case object SparkAnyType extends SparkType case class FunctionSignature(inputTypes: Seq[SparkType]) -sealed trait Function { - - def name: String - +case class Function(name: String, signatures: Seq[FunctionSignature]) { // query generator should choose inputs based on signature not just on arg count @deprecated - def numArgs: Int -} - -case class FunctionWithSignature(name: String, signatures: Seq[FunctionSignature]) - extends Function { - // query generator should choose inputs based on signature not just on arg count - override def numArgs: Int = signatures.head.inputTypes.length + def numArgs: Int = signatures.head.inputTypes.length } object Meta { @@ -75,23 +66,19 @@ object Meta { (DataTypes.StringType, 0.2), (DataTypes.BinaryType, 0.1)) - private def createFunctionWithInputs( - name: String, - inputs: Seq[SparkType]): FunctionWithSignature = { - FunctionWithSignature(name, Seq(FunctionSignature(inputs))) + private def createFunctionWithInputs(name: String, inputs: Seq[SparkType]): Function = { + Function(name, Seq(FunctionSignature(inputs))) } - private def createFunctionWithSignatures( - name: String, - signatures: Seq[FunctionSignature]): FunctionWithSignature = { - FunctionWithSignature(name, signatures) + private def createFunctions(name: String, signatures: Seq[FunctionSignature]): Function = { + Function(name, signatures) } - private def createUnaryStringFunction(name: String): FunctionWithSignature = { + private def createUnaryStringFunction(name: String): Function = { createFunctionWithInputs(name, Seq(SparkStringType)) } - private def createUnaryNumericFunction(name: String): FunctionWithSignature = { + private def createUnaryNumericFunction(name: String): Function = { createFunctionWithInputs(name, Seq(SparkNumericType)) } @@ -112,7 +99,7 @@ object Meta { createUnaryStringFunction("ltrim"), createUnaryStringFunction("rtrim"), createFunctionWithInputs("string_space", Seq(SparkIntType)), - createFunctionWithSignatures( + createFunctions( "rpad", Seq( FunctionSignature(Seq(SparkStringType, SparkIntegralType)), @@ -131,13 +118,13 @@ object Meta { createFunctionWithInputs( "length", Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType)))), - createFunctionWithSignatures( + createFunctions( "reverse", Seq( FunctionSignature(Seq(SparkStringType)), FunctionSignature(Seq(SparkArrayType(SparkAnyType))))), createFunctionWithInputs("instr", Seq(SparkStringType, SparkStringType)), - createFunctionWithSignatures( + createFunctions( "replace", Seq( FunctionSignature(Seq(SparkStringType, SparkStringType)), diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index 08a11ea643f..dc3f36d46fb 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -87,15 +87,12 @@ object QueryGen { val tableName = s"test${r.nextInt(numFiles)}" val table = spark.table(tableName) - val f = Utils.randomChoice(Meta.scalarFunc, r) - val args = f match { - case func: FunctionWithSignature => - val signature = Utils.randomChoice(func.signatures, r) - signature.inputTypes.map(x => pickRandomColumn(r, table, x)) - } + val func = Utils.randomChoice(Meta.scalarFunc, r) + val signature = Utils.randomChoice(func.signatures, r) + val args = signature.inputTypes.map(x => pickRandomColumn(r, table, x)) // Example SELECT c0, log(c0) as x FROM test0 - s"SELECT ${args.mkString(", ")}, ${f.name}(${args.mkString(", ")}) AS x " + + s"SELECT ${args.mkString(", ")}, ${func.name}(${args.mkString(", ")}) AS x " + s"FROM $tableName " + s"ORDER BY ${args.mkString(", ")};" } From 3c1feaf84c4946e8f9d0ff80ba7829aa0b38b34a Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 09:23:16 -0600 Subject: [PATCH 12/28] add more functions --- .../scala/org/apache/comet/fuzz/Meta.scala | 172 +++++++++++++++--- 1 file changed, 143 insertions(+), 29 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 781c62cae46..5d45a7394f3 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -24,6 +24,7 @@ import org.apache.spark.sql.types.DataTypes sealed trait SparkType case class SparkTypeOneOf(dataTypes: Seq[SparkType]) extends SparkType +case object SparkBooleanType extends SparkType case object SparkBinaryType extends SparkType case object SparkStringType extends SparkType case object SparkIntegralType extends SparkType @@ -39,6 +40,8 @@ case object SparkDateType extends SparkType case object SparkTimestampType extends SparkType case object SparkDateOrTimestampType extends SparkType case class SparkArrayType(elementType: SparkType) extends SparkType +case class SparkMapType(keyType: SparkType, valueType: SparkType) extends SparkType +case class SparkStructType(fields: Seq[SparkType]) extends SparkType case object SparkAnyType extends SparkType case class FunctionSignature(inputTypes: Seq[SparkType]) @@ -104,13 +107,17 @@ object Meta { Seq( FunctionSignature(Seq(SparkStringType, SparkIntegralType)), FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), + createFunctions( + "lpad", + Seq( + FunctionSignature(Seq(SparkStringType, SparkIntegralType)), + FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), createFunctionWithInputs( "hex", Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType, SparkIntType, SparkLongType)))), createUnaryStringFunction("unhex"), createFunctionWithInputs("xxhash64", Seq(SparkAnyType)), // TODO can take multiple columns createFunctionWithInputs("sha1", Seq(SparkAnyType)), - // createFunction("sha2", 1), -- needs a second argument for number of bits createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), createUnaryStringFunction("btrim"), createFunctionWithInputs("concat_ws", Seq(SparkStringType, SparkStringType)), @@ -129,56 +136,160 @@ object Meta { Seq( FunctionSignature(Seq(SparkStringType, SparkStringType)), FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)))), - createFunctionWithInputs("translate", Seq(SparkStringType, SparkStringType))) + createFunctionWithInputs("translate", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs("like", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs("rlike", Seq(SparkStringType, SparkStringType)), + createFunctions( + "regexp_replace", + Seq( + FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)), + FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType, SparkIntType))))) val dateScalarFunc: Seq[Function] = Seq( createFunctionWithInputs("year", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("month", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("day", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("dayofmonth", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("dayofweek", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("dayofyear", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("weekday", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("weekofyear", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("quarter", Seq(SparkDateOrTimestampType)), createFunctionWithInputs("hour", Seq(SparkDateOrTimestampType)), createFunctionWithInputs("minute", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("second", Seq(SparkDateOrTimestampType))) + createFunctionWithInputs("second", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("date_add", Seq(SparkDateType, SparkIntType)), + createFunctionWithInputs("date_sub", Seq(SparkDateType, SparkIntType)), + createFunctionWithInputs("trunc", Seq(SparkDateOrTimestampType, SparkStringType)), + createFunctions( + "from_unixtime", + Seq( + FunctionSignature(Seq(SparkLongType)), + FunctionSignature(Seq(SparkLongType, SparkStringType))))) val mathScalarFunc: Seq[Function] = Seq( createUnaryNumericFunction("abs"), createUnaryNumericFunction("acos"), createUnaryNumericFunction("asin"), createUnaryNumericFunction("atan"), - createUnaryNumericFunction("Atan2"), - createUnaryNumericFunction("Cos"), - createFunctionWithInputs("Exp", Seq(SparkNumericType, SparkNumericType)), - createUnaryNumericFunction("Ln"), - createUnaryNumericFunction("Log10"), - createUnaryNumericFunction("Log2"), - createFunctionWithInputs("Pow", Seq(SparkNumericType, SparkNumericType)), - createUnaryNumericFunction("Round"), - createUnaryNumericFunction("Signum"), - createUnaryNumericFunction("Sin"), - createUnaryNumericFunction("Sqrt"), - createUnaryNumericFunction("Tan"), - createUnaryNumericFunction("Ceil"), - createUnaryNumericFunction("Floor"), - createFunctionWithInputs("bool_and", Seq(SparkAnyType)), - createFunctionWithInputs("bool_or", Seq(SparkAnyType)), - createFunctionWithInputs("bitwise_not", Seq(SparkIntegralType))) + createFunctionWithInputs("atan2", Seq(SparkNumericType, SparkNumericType)), + createUnaryNumericFunction("cos"), + createUnaryNumericFunction("exp"), + createUnaryNumericFunction("expm1"), + createFunctionWithInputs("log", Seq(SparkNumericType, SparkNumericType)), + createUnaryNumericFunction("log10"), + createUnaryNumericFunction("log2"), + createFunctionWithInputs("pow", Seq(SparkNumericType, SparkNumericType)), + createFunctions( + "round", + Seq( + FunctionSignature(Seq(SparkNumericType)), + FunctionSignature(Seq(SparkNumericType, SparkIntType)))), + createUnaryNumericFunction("signum"), + createUnaryNumericFunction("sin"), + createUnaryNumericFunction("sqrt"), + createUnaryNumericFunction("tan"), + createUnaryNumericFunction("ceil"), + createUnaryNumericFunction("floor"), + createFunctionWithInputs("remainder", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputs("unary_minus", Seq(SparkNumericType))) + + val hashScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs("md5", Seq(SparkAnyType)), + createFunctionWithInputs("murmur3_hash", Seq(SparkAnyType)), // TODO can take multiple columns + createFunctionWithInputs("sha2", Seq(SparkAnyType, SparkIntType))) + + val bitwiseScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs("bitwise_and", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputs("bitwise_or", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputs("bitwise_xor", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputs("bitwise_not", Seq(SparkIntegralType)), + createFunctionWithInputs("bitwise_count", Seq(SparkIntegralType)), + createFunctionWithInputs("bitwise_get", Seq(SparkIntegralType, SparkIntType)), + createFunctionWithInputs("shift_left", Seq(SparkIntegralType, SparkIntType)), + createFunctionWithInputs("shift_right", Seq(SparkIntegralType, SparkIntType))) + + val predicateScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs("and", Seq(SparkBooleanType, SparkBooleanType)), + createFunctionWithInputs("or", Seq(SparkBooleanType, SparkBooleanType)), + createFunctionWithInputs("not", Seq(SparkBooleanType)), + createFunctionWithInputs("in", Seq(SparkAnyType, SparkAnyType)) + ) // TODO: variadic + + val conditionalScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs("if", Seq(SparkBooleanType, SparkAnyType, SparkAnyType))) val miscScalarFunc: Seq[Function] = Seq( createFunctionWithInputs("isnan", Seq(SparkNumericType)), createFunctionWithInputs("isnull", Seq(SparkAnyType)), - createFunctionWithInputs("isnotnull", Seq(SparkAnyType))) + createFunctionWithInputs("isnotnull", Seq(SparkAnyType)), + createFunctionWithInputs("coalesce", Seq(SparkAnyType, SparkAnyType)) + ) // TODO: variadic val arrayScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("array", Seq(SparkAnyType, SparkAnyType)), - createFunctionWithInputs("array_remove", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), - createFunctionWithInputs("array_insert", Seq(SparkArrayType(SparkAnyType), SparkIntType)), + createFunctionWithInputs("array", Seq(SparkAnyType, SparkAnyType)), // TODO: variadic + createFunctionWithInputs("array_append", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputs("array_compact", Seq(SparkArrayType(SparkAnyType))), createFunctionWithInputs("array_contains", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputs("array_distinct", Seq(SparkArrayType(SparkAnyType))), + createFunctionWithInputs( + "array_except", + Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), + createFunctionWithInputs( + "array_insert", + Seq(SparkArrayType(SparkAnyType), SparkIntType, SparkAnyType)), createFunctionWithInputs( "array_intersect", Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), - createFunctionWithInputs("array_append", Seq(SparkArrayType(SparkAnyType), SparkAnyType))) + createFunctions( + "array_join", + Seq( + FunctionSignature(Seq(SparkArrayType(SparkAnyType), SparkStringType)), + FunctionSignature(Seq(SparkArrayType(SparkAnyType), SparkStringType, SparkStringType)))), + createFunctionWithInputs("array_max", Seq(SparkArrayType(SparkAnyType))), + createFunctionWithInputs("array_min", Seq(SparkArrayType(SparkAnyType))), + createFunctionWithInputs("array_remove", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputs("array_repeat", Seq(SparkAnyType, SparkIntType)), + createFunctionWithInputs( + "arrays_overlap", + Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), + createFunctionWithInputs( + "array_union", + Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), + createFunctionWithInputs( + "element_at", + Seq( + SparkTypeOneOf( + Seq(SparkArrayType(SparkAnyType), SparkMapType(SparkAnyType, SparkAnyType))), + SparkAnyType)), + createFunctionWithInputs("flatten", Seq(SparkArrayType(SparkArrayType(SparkAnyType)))), + createFunctionWithInputs("get_array_item", Seq(SparkArrayType(SparkAnyType), SparkIntType))) + + val mapScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs( + "map_extract", + Seq(SparkMapType(SparkAnyType, SparkAnyType), SparkAnyType)), + createFunctionWithInputs("map_keys", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputs("map_values", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputs("map_entries", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputs( + "map_from_arrays", + Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType)))) + + val structScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs( + "create_named_struct", + Seq(SparkStringType, SparkAnyType) + ), // TODO: variadic name/value pairs + createFunctionWithInputs( + "get_struct_field", + Seq(SparkStructType(Seq(SparkAnyType)), SparkStringType))) val scalarFunc: Seq[Function] = stringScalarFunc ++ dateScalarFunc ++ - mathScalarFunc ++ miscScalarFunc ++ arrayScalarFunc + mathScalarFunc ++ hashScalarFunc ++ bitwiseScalarFunc ++ predicateScalarFunc ++ + conditionalScalarFunc ++ miscScalarFunc ++ arrayScalarFunc ++ mapScalarFunc ++ structScalarFunc val aggFunc: Seq[Function] = Seq( createFunctionWithInputs("min", Seq(SparkAnyType)), @@ -190,11 +301,14 @@ object Meta { createFunctionWithInputs("last", Seq(SparkAnyType)), createFunctionWithInputs("var_pop", Seq(SparkNumericType)), createFunctionWithInputs("var_samp", Seq(SparkNumericType)), - createFunctionWithInputs("covar_pop", Seq(SparkNumericType)), - createFunctionWithInputs("covar_samp", Seq(SparkNumericType)), + createFunctionWithInputs("covar_pop", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputs("covar_samp", Seq(SparkNumericType, SparkNumericType)), createFunctionWithInputs("stddev_pop", Seq(SparkNumericType)), createFunctionWithInputs("stddev_samp", Seq(SparkNumericType)), - createFunctionWithInputs("corr", Seq(SparkNumericType, SparkNumericType))) + createFunctionWithInputs("corr", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputs("bit_and", Seq(SparkIntegralType)), + createFunctionWithInputs("bit_or", Seq(SparkIntegralType)), + createFunctionWithInputs("bit_xor", Seq(SparkIntegralType))) val unaryArithmeticOps: Seq[String] = Seq("+", "-") From 9e185eb72b50a9b75c4d50256e853dabb8d94f5d Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 09:28:43 -0600 Subject: [PATCH 13/28] refactor [skip ci] --- .../scala/org/apache/comet/fuzz/Meta.scala | 247 +++++++++--------- 1 file changed, 129 insertions(+), 118 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 5d45a7394f3..8eafbcea565 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -85,89 +85,7 @@ object Meta { createFunctionWithInputs(name, Seq(SparkNumericType)) } - val stringScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), - createUnaryStringFunction("coalesce"), - createFunctionWithInputs("starts_with", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("ends_with", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("contains", Seq(SparkStringType, SparkStringType)), - createUnaryStringFunction("ascii"), - createUnaryStringFunction("bit_length"), - createUnaryStringFunction("octet_length"), - createUnaryStringFunction("upper"), - createUnaryStringFunction("lower"), - createUnaryStringFunction("chr"), - createUnaryStringFunction("init_cap"), - createUnaryStringFunction("trim"), - createUnaryStringFunction("ltrim"), - createUnaryStringFunction("rtrim"), - createFunctionWithInputs("string_space", Seq(SparkIntType)), - createFunctions( - "rpad", - Seq( - FunctionSignature(Seq(SparkStringType, SparkIntegralType)), - FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), - createFunctions( - "lpad", - Seq( - FunctionSignature(Seq(SparkStringType, SparkIntegralType)), - FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), - createFunctionWithInputs( - "hex", - Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType, SparkIntType, SparkLongType)))), - createUnaryStringFunction("unhex"), - createFunctionWithInputs("xxhash64", Seq(SparkAnyType)), // TODO can take multiple columns - createFunctionWithInputs("sha1", Seq(SparkAnyType)), - createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), - createUnaryStringFunction("btrim"), - createFunctionWithInputs("concat_ws", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("repeat", Seq(SparkStringType, SparkIntType)), - createFunctionWithInputs( - "length", - Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType)))), - createFunctions( - "reverse", - Seq( - FunctionSignature(Seq(SparkStringType)), - FunctionSignature(Seq(SparkArrayType(SparkAnyType))))), - createFunctionWithInputs("instr", Seq(SparkStringType, SparkStringType)), - createFunctions( - "replace", - Seq( - FunctionSignature(Seq(SparkStringType, SparkStringType)), - FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)))), - createFunctionWithInputs("translate", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("like", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("rlike", Seq(SparkStringType, SparkStringType)), - createFunctions( - "regexp_replace", - Seq( - FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)), - FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType, SparkIntType))))) - - val dateScalarFunc: Seq[Function] = - Seq( - createFunctionWithInputs("year", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("month", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("day", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("dayofmonth", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("dayofweek", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("dayofyear", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("weekday", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("weekofyear", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("quarter", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("hour", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("minute", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("second", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("date_add", Seq(SparkDateType, SparkIntType)), - createFunctionWithInputs("date_sub", Seq(SparkDateType, SparkIntType)), - createFunctionWithInputs("trunc", Seq(SparkDateOrTimestampType, SparkStringType)), - createFunctions( - "from_unixtime", - Seq( - FunctionSignature(Seq(SparkLongType)), - FunctionSignature(Seq(SparkLongType, SparkStringType))))) - + // Math expressions (corresponds to mathExpressions in QueryPlanSerde) val mathScalarFunc: Seq[Function] = Seq( createUnaryNumericFunction("abs"), createUnaryNumericFunction("acos"), @@ -181,6 +99,7 @@ object Meta { createUnaryNumericFunction("log10"), createUnaryNumericFunction("log2"), createFunctionWithInputs("pow", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputs("remainder", Seq(SparkNumericType, SparkNumericType)), createFunctions( "round", Seq( @@ -192,24 +111,90 @@ object Meta { createUnaryNumericFunction("tan"), createUnaryNumericFunction("ceil"), createUnaryNumericFunction("floor"), - createFunctionWithInputs("remainder", Seq(SparkNumericType, SparkNumericType)), createFunctionWithInputs("unary_minus", Seq(SparkNumericType))) + // Hash expressions (corresponds to hashExpressions in QueryPlanSerde) val hashScalarFunc: Seq[Function] = Seq( createFunctionWithInputs("md5", Seq(SparkAnyType)), createFunctionWithInputs("murmur3_hash", Seq(SparkAnyType)), // TODO can take multiple columns createFunctionWithInputs("sha2", Seq(SparkAnyType, SparkIntType))) - val bitwiseScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("bitwise_and", Seq(SparkIntegralType, SparkIntegralType)), - createFunctionWithInputs("bitwise_or", Seq(SparkIntegralType, SparkIntegralType)), - createFunctionWithInputs("bitwise_xor", Seq(SparkIntegralType, SparkIntegralType)), - createFunctionWithInputs("bitwise_not", Seq(SparkIntegralType)), - createFunctionWithInputs("bitwise_count", Seq(SparkIntegralType)), - createFunctionWithInputs("bitwise_get", Seq(SparkIntegralType, SparkIntType)), - createFunctionWithInputs("shift_left", Seq(SparkIntegralType, SparkIntType)), - createFunctionWithInputs("shift_right", Seq(SparkIntegralType, SparkIntType))) + // String expressions (corresponds to stringExpressions in QueryPlanSerde) + val stringScalarFunc: Seq[Function] = Seq( + createUnaryStringFunction("ascii"), + createUnaryStringFunction("bit_length"), + createUnaryStringFunction("chr"), + createFunctionWithInputs("concat_ws", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs("contains", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs("ends_with", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs( + "hex", + Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType, SparkIntType, SparkLongType)))), + createUnaryStringFunction("init_cap"), + createFunctionWithInputs("instr", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs( + "length", + Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType)))), + createFunctionWithInputs("like", Seq(SparkStringType, SparkStringType)), + createUnaryStringFunction("lower"), + createFunctions( + "lpad", + Seq( + FunctionSignature(Seq(SparkStringType, SparkIntegralType)), + FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), + createUnaryStringFunction("ltrim"), + createUnaryStringFunction("octet_length"), + createFunctions( + "regexp_replace", + Seq( + FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)), + FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType, SparkIntType)))), + createFunctionWithInputs("repeat", Seq(SparkStringType, SparkIntType)), + createFunctions( + "replace", + Seq( + FunctionSignature(Seq(SparkStringType, SparkStringType)), + FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)))), + createFunctions( + "reverse", + Seq( + FunctionSignature(Seq(SparkStringType)), + FunctionSignature(Seq(SparkArrayType(SparkAnyType))))), + createFunctionWithInputs("rlike", Seq(SparkStringType, SparkStringType)), + createFunctions( + "rpad", + Seq( + FunctionSignature(Seq(SparkStringType, SparkIntegralType)), + FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), + createUnaryStringFunction("rtrim"), + createFunctionWithInputs("starts_with", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputs("string_space", Seq(SparkIntType)), + createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), + createFunctionWithInputs("translate", Seq(SparkStringType, SparkStringType)), + createUnaryStringFunction("trim"), + createUnaryStringFunction("btrim"), + createUnaryStringFunction("unhex"), + createUnaryStringFunction("upper"), + createFunctionWithInputs("xxhash64", Seq(SparkAnyType)), // TODO can take multiple columns + createFunctionWithInputs("sha1", Seq(SparkAnyType))) + // Conditional expressions (corresponds to conditionalExpressions in QueryPlanSerde) + val conditionalScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs("if", Seq(SparkBooleanType, SparkAnyType, SparkAnyType))) + + // Map expressions (corresponds to mapExpressions in QueryPlanSerde) + val mapScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs( + "map_extract", + Seq(SparkMapType(SparkAnyType, SparkAnyType), SparkAnyType)), + createFunctionWithInputs("map_keys", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputs("map_entries", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputs("map_values", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputs( + "map_from_arrays", + Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType)))) + + // Predicate expressions (corresponds to predicateExpressions in QueryPlanSerde) val predicateScalarFunc: Seq[Function] = Seq( createFunctionWithInputs("and", Seq(SparkBooleanType, SparkBooleanType)), createFunctionWithInputs("or", Seq(SparkBooleanType, SparkBooleanType)), @@ -217,9 +202,28 @@ object Meta { createFunctionWithInputs("in", Seq(SparkAnyType, SparkAnyType)) ) // TODO: variadic - val conditionalScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("if", Seq(SparkBooleanType, SparkAnyType, SparkAnyType))) + // Struct expressions (corresponds to structExpressions in QueryPlanSerde) + val structScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs( + "create_named_struct", + Seq(SparkStringType, SparkAnyType) + ), // TODO: variadic name/value pairs + createFunctionWithInputs( + "get_struct_field", + Seq(SparkStructType(Seq(SparkAnyType)), SparkStringType))) + + // Bitwise expressions (corresponds to bitwiseExpressions in QueryPlanSerde) + val bitwiseScalarFunc: Seq[Function] = Seq( + createFunctionWithInputs("bitwise_and", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputs("bitwise_count", Seq(SparkIntegralType)), + createFunctionWithInputs("bitwise_get", Seq(SparkIntegralType, SparkIntType)), + createFunctionWithInputs("bitwise_or", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputs("bitwise_not", Seq(SparkIntegralType)), + createFunctionWithInputs("bitwise_xor", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputs("shift_left", Seq(SparkIntegralType, SparkIntType)), + createFunctionWithInputs("shift_right", Seq(SparkIntegralType, SparkIntType))) + // Misc expressions (corresponds to miscExpressions in QueryPlanSerde) val miscScalarFunc: Seq[Function] = Seq( createFunctionWithInputs("isnan", Seq(SparkNumericType)), @@ -228,8 +232,8 @@ object Meta { createFunctionWithInputs("coalesce", Seq(SparkAnyType, SparkAnyType)) ) // TODO: variadic + // Array expressions (corresponds to arrayExpressions in QueryPlanSerde) val arrayScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("array", Seq(SparkAnyType, SparkAnyType)), // TODO: variadic createFunctionWithInputs("array_append", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), createFunctionWithInputs("array_compact", Seq(SparkArrayType(SparkAnyType))), createFunctionWithInputs("array_contains", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), @@ -258,6 +262,7 @@ object Meta { createFunctionWithInputs( "array_union", Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), + createFunctionWithInputs("array", Seq(SparkAnyType, SparkAnyType)), // TODO: variadic createFunctionWithInputs( "element_at", Seq( @@ -267,29 +272,35 @@ object Meta { createFunctionWithInputs("flatten", Seq(SparkArrayType(SparkArrayType(SparkAnyType)))), createFunctionWithInputs("get_array_item", Seq(SparkArrayType(SparkAnyType), SparkIntType))) - val mapScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs( - "map_extract", - Seq(SparkMapType(SparkAnyType, SparkAnyType), SparkAnyType)), - createFunctionWithInputs("map_keys", Seq(SparkMapType(SparkAnyType, SparkAnyType))), - createFunctionWithInputs("map_values", Seq(SparkMapType(SparkAnyType, SparkAnyType))), - createFunctionWithInputs("map_entries", Seq(SparkMapType(SparkAnyType, SparkAnyType))), - createFunctionWithInputs( - "map_from_arrays", - Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType)))) - - val structScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs( - "create_named_struct", - Seq(SparkStringType, SparkAnyType) - ), // TODO: variadic name/value pairs - createFunctionWithInputs( - "get_struct_field", - Seq(SparkStructType(Seq(SparkAnyType)), SparkStringType))) + // Temporal expressions (corresponds to temporalExpressions in QueryPlanSerde) + val temporalScalarFunc: Seq[Function] = + Seq( + createFunctionWithInputs("date_add", Seq(SparkDateType, SparkIntType)), + createFunctionWithInputs("date_sub", Seq(SparkDateType, SparkIntType)), + createFunctions( + "from_unixtime", + Seq( + FunctionSignature(Seq(SparkLongType)), + FunctionSignature(Seq(SparkLongType, SparkStringType)))), + createFunctionWithInputs("hour", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("minute", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("second", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("trunc", Seq(SparkDateOrTimestampType, SparkStringType)), + createFunctionWithInputs("year", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("month", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("day", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("dayofmonth", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("dayofweek", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("weekday", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("dayofyear", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("weekofyear", Seq(SparkDateOrTimestampType)), + createFunctionWithInputs("quarter", Seq(SparkDateOrTimestampType))) - val scalarFunc: Seq[Function] = stringScalarFunc ++ dateScalarFunc ++ - mathScalarFunc ++ hashScalarFunc ++ bitwiseScalarFunc ++ predicateScalarFunc ++ - conditionalScalarFunc ++ miscScalarFunc ++ arrayScalarFunc ++ mapScalarFunc ++ structScalarFunc + // Combined in same order as exprSerdeMap in QueryPlanSerde + val scalarFunc: Seq[Function] = mathScalarFunc ++ hashScalarFunc ++ stringScalarFunc ++ + conditionalScalarFunc ++ mapScalarFunc ++ predicateScalarFunc ++ + structScalarFunc ++ bitwiseScalarFunc ++ miscScalarFunc ++ arrayScalarFunc ++ + temporalScalarFunc val aggFunc: Seq[Function] = Seq( createFunctionWithInputs("min", Seq(SparkAnyType)), From d13301df49d09f0c34a0c6d8677f1e00d2fda5c6 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 09:36:42 -0600 Subject: [PATCH 14/28] query gen updates --- .../org/apache/comet/fuzz/QueryGen.scala | 63 ++++++++++++++++++- 1 file changed, 61 insertions(+), 2 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index dc3f36d46fb..be1f6d0d3a3 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -101,6 +101,8 @@ object QueryGen { targetType match { case SparkAnyType => Utils.randomChoice(df.schema.fields, r).name + case SparkBooleanType => + select(r, df, _.dataType == BooleanType) case SparkByteType => select(r, df, _.dataType == ByteType) case SparkShortType => @@ -109,6 +111,19 @@ object QueryGen { select(r, df, _.dataType == IntegerType) case SparkLongType => select(r, df, _.dataType == LongType) + case SparkFloatType => + select(r, df, _.dataType == FloatType) + case SparkDoubleType => + select(r, df, _.dataType == DoubleType) + case SparkDecimalType(_, _) => + select(r, df, _.dataType.isInstanceOf[DecimalType]) + case SparkIntegralType => + select( + r, + df, + f => + f.dataType == ByteType || f.dataType == ShortType || + f.dataType == IntegerType || f.dataType == LongType) case SparkNumericType => select(r, df, f => isNumeric(f.dataType)) case SparkStringType => @@ -131,6 +146,22 @@ object QueryGen { case ArrayType(x, _) if typeMatch(elementType, x) => true case _ => false }) + case SparkMapType(keyType, valueType) => + select( + r, + df, + _.dataType match { + case MapType(k, v, _) if typeMatch(keyType, k) && typeMatch(valueType, v) => true + case _ => false + }) + case SparkStructType(fields) => + select( + r, + df, + _.dataType match { + case StructType(structFields) if structFields.length == fields.length => true + case _ => false + }) case _ => throw new IllegalStateException(targetType.toString) } @@ -151,8 +182,36 @@ object QueryGen { } private def typeMatch(s: SparkType, d: DataType): Boolean = { - // TODO - false + (s, d) match { + case (SparkAnyType, _) => true + case (SparkBooleanType, BooleanType) => true + case (SparkByteType, ByteType) => true + case (SparkShortType, ShortType) => true + case (SparkIntType, IntegerType) => true + case (SparkLongType, LongType) => true + case (SparkFloatType, FloatType) => true + case (SparkDoubleType, DoubleType) => true + case (SparkDecimalType(_, _), _: DecimalType) => true + case (SparkIntegralType, ByteType | ShortType | IntegerType | LongType) => true + case (SparkNumericType, _) if isNumeric(d) => true + case (SparkStringType, StringType) => true + case (SparkBinaryType, BinaryType) => true + case (SparkDateType, DateType) => true + case (SparkTimestampType, TimestampType | TimestampNTZType) => true + case (SparkDateOrTimestampType, DateType | TimestampType | TimestampNTZType) => true + case (SparkArrayType(elementType), ArrayType(elementDataType, _)) => + typeMatch(elementType, elementDataType) + case (SparkMapType(keyType, valueType), MapType(keyDataType, valueDataType, _)) => + typeMatch(keyType, keyDataType) && typeMatch(valueType, valueDataType) + case (SparkStructType(fields), StructType(structFields)) => + fields.length == structFields.length && + fields.zip(structFields.map(_.dataType)).forall { case (sparkType, dataType) => + typeMatch(sparkType, dataType) + } + case (SparkTypeOneOf(choices), _) => + choices.exists(choice => typeMatch(choice, d)) + case _ => false + } } private def generateUnaryArithmetic(r: Random, spark: SparkSession, numFiles: Int): String = { From 4d4362ec3b8235ee39a228b4c1820d3acfa06107 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 09:49:13 -0600 Subject: [PATCH 15/28] error handling --- .../org/apache/comet/fuzz/QueryGen.scala | 57 ++++++++++++------- 1 file changed, 37 insertions(+), 20 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index be1f6d0d3a3..032548c1372 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -43,19 +43,25 @@ object QueryGen { val uniqueQueries = mutable.HashSet[String]() for (_ <- 0 until numQueries) { - val sql = r.nextInt().abs % 8 match { - case 0 => generateJoin(r, spark, numFiles) - case 1 => generateAggregate(r, spark, numFiles) - case 2 => generateScalar(r, spark, numFiles) - case 3 => generateCast(r, spark, numFiles) - case 4 => generateUnaryArithmetic(r, spark, numFiles) - case 5 => generateBinaryArithmetic(r, spark, numFiles) - case 6 => generateBinaryComparison(r, spark, numFiles) - case _ => generateConditional(r, spark, numFiles) - } - if (!uniqueQueries.contains(sql)) { - uniqueQueries += sql - w.write(sql + "\n") + try { + val sql = r.nextInt().abs % 8 match { + case 0 => generateJoin(r, spark, numFiles) + case 1 => generateAggregate(r, spark, numFiles) + case 2 => generateScalar(r, spark, numFiles) + case 3 => generateCast(r, spark, numFiles) + case 4 => generateUnaryArithmetic(r, spark, numFiles) + case 5 => generateBinaryArithmetic(r, spark, numFiles) + case 6 => generateBinaryComparison(r, spark, numFiles) + case _ => generateConditional(r, spark, numFiles) + } + if (!uniqueQueries.contains(sql)) { + uniqueQueries += sql + w.write(sql + "\n") + } + } catch { + case e: Exception => + // scalastyle:off + println(s"Failed to generate query: ${e.getMessage}") } } w.close() @@ -88,13 +94,20 @@ object QueryGen { val table = spark.table(tableName) val func = Utils.randomChoice(Meta.scalarFunc, r) - val signature = Utils.randomChoice(func.signatures, r) - val args = signature.inputTypes.map(x => pickRandomColumn(r, table, x)) + try { + val signature = Utils.randomChoice(func.signatures, r) + val args = signature.inputTypes.map(x => pickRandomColumn(r, table, x)) - // Example SELECT c0, log(c0) as x FROM test0 - s"SELECT ${args.mkString(", ")}, ${func.name}(${args.mkString(", ")}) AS x " + - s"FROM $tableName " + - s"ORDER BY ${args.mkString(", ")};" + // Example SELECT c0, log(c0) as x FROM test0 + s"SELECT ${args.mkString(", ")}, ${func.name}(${args.mkString(", ")}) AS x " + + s"FROM $tableName " + + s"ORDER BY ${args.mkString(", ")};" + } catch { + case e: Exception => + throw new IllegalStateException( + s"Failed to generate SQL for scalar function ${func.name}", + e) + } } private def pickRandomColumn(r: Random, df: DataFrame, targetType: SparkType): String = { @@ -169,7 +182,11 @@ object QueryGen { /** Select a random field that matches a predicate */ private def select(r: Random, df: DataFrame, predicate: StructField => Boolean): String = { - Utils.randomChoice(df.schema.fields.filter(predicate), r).name + val candidates = df.schema.fields.filter(predicate) + if (candidates.isEmpty) { + throw new IllegalStateException("Failed to find suitable column") + } + Utils.randomChoice(candidates, r).name } private def isNumeric(d: DataType): Boolean = { From a90e64f99e7eb39e1cbfc46e75102d5aba71445a Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 10:22:16 -0600 Subject: [PATCH 16/28] update agg query gen --- .../scala/org/apache/comet/fuzz/Meta.scala | 6 +--- .../org/apache/comet/fuzz/QueryGen.scala | 33 +++++++++++-------- 2 files changed, 21 insertions(+), 18 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 8eafbcea565..cf0c5e78adf 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -46,11 +46,7 @@ case object SparkAnyType extends SparkType case class FunctionSignature(inputTypes: Seq[SparkType]) -case class Function(name: String, signatures: Seq[FunctionSignature]) { - // query generator should choose inputs based on signature not just on arg count - @deprecated - def numArgs: Int = signatures.head.inputTypes.length -} +case class Function(name: String, signatures: Seq[FunctionSignature]) object Meta { diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index 032548c1372..bd1e5779d6b 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -72,20 +72,27 @@ object QueryGen { val table = spark.table(tableName) val func = Utils.randomChoice(Meta.aggFunc, r) - val args = Range(0, func.numArgs) - .map(_ => Utils.randomChoice(table.columns, r)) - - val groupingCols = Range(0, 2).map(_ => Utils.randomChoice(table.columns, r)) + try { + val signature = Utils.randomChoice(func.signatures, r) + val args = signature.inputTypes.map(x => pickRandomColumn(r, table, x)) - if (groupingCols.isEmpty) { - s"SELECT ${args.mkString(", ")}, ${func.name}(${args.mkString(", ")}) AS x " + - s"FROM $tableName " + - s"ORDER BY ${args.mkString(", ")};" - } else { - s"SELECT ${groupingCols.mkString(", ")}, ${func.name}(${args.mkString(", ")}) " + - s"FROM $tableName " + - s"GROUP BY ${groupingCols.mkString(",")} " + - s"ORDER BY ${groupingCols.mkString(", ")};" + val groupingCols = Range(0, 2).map(_ => Utils.randomChoice(table.columns, r)) + + if (groupingCols.isEmpty) { + s"SELECT ${args.mkString(", ")}, ${func.name}(${args.mkString(", ")}) AS x " + + s"FROM $tableName " + + s"ORDER BY ${args.mkString(", ")};" + } else { + s"SELECT ${groupingCols.mkString(", ")}, ${func.name}(${args.mkString(", ")}) " + + s"FROM $tableName " + + s"GROUP BY ${groupingCols.mkString(",")} " + + s"ORDER BY ${groupingCols.mkString(", ")};" + } + } catch { + case e: Exception => + throw new IllegalStateException( + s"Failed to generate SQL for scalar function ${func.name}", + e) } } From ad71962322efa9a248a24694df707036e1ef93ba Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 11:04:39 -0600 Subject: [PATCH 17/28] fix copy paste --- .../src/main/scala/org/apache/comet/fuzz/QueryGen.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index bd1e5779d6b..abd4c9b46a7 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -91,7 +91,7 @@ object QueryGen { } catch { case e: Exception => throw new IllegalStateException( - s"Failed to generate SQL for scalar function ${func.name}", + s"Failed to generate SQL for aggregate function ${func.name}", e) } } From 9841b5c9d843311b1f856e6a653b8678b69a78f5 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 11:58:35 -0600 Subject: [PATCH 18/28] fix --- fuzz-testing/README.md | 1 - .../scala/org/apache/comet/fuzz/QueryRunner.scala | 11 ++++++++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/fuzz-testing/README.md b/fuzz-testing/README.md index bde1505e3e3..966c5df4c06 100644 --- a/fuzz-testing/README.md +++ b/fuzz-testing/README.md @@ -91,7 +91,6 @@ $SPARK_HOME/bin/spark-submit \ --conf spark.plugins=org.apache.spark.CometPlugin \ --conf spark.comet.enabled=true \ --conf spark.comet.exec.enabled=true \ - --conf spark.comet.exec.all.enabled=true \ --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \ --conf spark.comet.exec.shuffle.enabled=true \ --conf spark.comet.exec.shuffle.mode=auto \ diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala index 8852f4bc171..9769de876eb 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala @@ -67,9 +67,6 @@ object QueryRunner { // execute with Comet try { spark.conf.set("spark.comet.enabled", "true") - // complex type support until we support it natively - spark.conf.set("spark.comet.sparkToColumnar.enabled", "true") - spark.conf.set("spark.comet.convert.parquet.enabled", "true") val df = spark.sql(sql) val cometRows = df.collect() val cometPlan = df.queryExecution.executedPlan.toString @@ -133,6 +130,9 @@ object QueryRunner { } private def same(l: Any, r: Any): Boolean = { + if (l == null || r == null) { + return l == null && r == null + } (l, r) match { case (a: Float, b: Float) if a.isInfinity => b.isInfinity case (a: Float, b: Float) if a.isNaN => b.isNaN @@ -144,7 +144,11 @@ object QueryRunner { a.length == b.length && a.zip(b).forall(x => same(x._1, x._2)) case (a: WrappedArray[_], b: WrappedArray[_]) => a.length == b.length && a.zip(b).forall(x => same(x._1, x._2)) + case (a: Row, b: Row) => + // struct support + format(a) == format(b) case (a, b) => a == b + } } @@ -153,6 +157,7 @@ object QueryRunner { case null => "NULL" case v: WrappedArray[_] => s"[${v.map(format).mkString(",")}]" case v: Array[Byte] => s"[${v.mkString(",")}]" + case r: Row => formatRow(r) case other => other.toString } } From 9509bfa01310b1e0dfbf39f199dbaf3a030125c3 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 12:50:29 -0600 Subject: [PATCH 19/28] partially address feedback --- .../main/scala/org/apache/comet/fuzz/QueryRunner.scala | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala index 9769de876eb..835cfe410e9 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala @@ -134,9 +134,13 @@ object QueryRunner { return l == null && r == null } (l, r) match { + case (a: Float, b: Float) if a.isPosInfinity => b.isPosInfinity + case (a: Float, b: Float) if a.isNegInfinity => b.isNegInfinity case (a: Float, b: Float) if a.isInfinity => b.isInfinity case (a: Float, b: Float) if a.isNaN => b.isNaN case (a: Float, b: Float) => (a - b).abs <= 0.000001f + case (a: Double, b: Double) if a.isPosInfinity => b.isPosInfinity + case (a: Double, b: Double) if a.isNegInfinity => b.isNegInfinity case (a: Double, b: Double) if a.isInfinity => b.isInfinity case (a: Double, b: Double) if a.isNaN => b.isNaN case (a: Double, b: Double) => (a - b).abs <= 0.000001 @@ -145,10 +149,10 @@ object QueryRunner { case (a: WrappedArray[_], b: WrappedArray[_]) => a.length == b.length && a.zip(b).forall(x => same(x._1, x._2)) case (a: Row, b: Row) => - // struct support - format(a) == format(b) + val aa = a.toSeq + val bb = b.toSeq + aa.length == bb.length && aa.zip(bb).forall(x => same(x._1, x._2)) case (a, b) => a == b - } } From 903482c3f43eb81e607c2ba341b8e38f387bc0d9 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 13:10:08 -0600 Subject: [PATCH 20/28] concat --- .../main/scala/org/apache/comet/fuzz/Meta.scala | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index cf0c5e78adf..f1235e75aeb 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -120,6 +120,23 @@ object Meta { createUnaryStringFunction("ascii"), createUnaryStringFunction("bit_length"), createUnaryStringFunction("chr"), + createFunctionWithInputs( + "concat", + Seq( + SparkTypeOneOf( + Seq( + SparkStringType, + SparkNumericType, + SparkBinaryType, + SparkArrayType( + SparkTypeOneOf(Seq(SparkStringType, SparkNumericType, SparkBinaryType))))), + SparkTypeOneOf( + Seq( + SparkStringType, + SparkNumericType, + SparkBinaryType, + SparkArrayType( + SparkTypeOneOf(Seq(SparkStringType, SparkNumericType, SparkBinaryType))))))), createFunctionWithInputs("concat_ws", Seq(SparkStringType, SparkStringType)), createFunctionWithInputs("contains", Seq(SparkStringType, SparkStringType)), createFunctionWithInputs("ends_with", Seq(SparkStringType, SparkStringType)), From aa979e6ba66ce73f2be2d28becae52776547ba54 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 16:09:50 -0600 Subject: [PATCH 21/28] reduce number of invalid queries --- .../org/apache/comet/fuzz/QueryGen.scala | 18 ++++++++------ .../org/apache/comet/fuzz/QueryRunner.scala | 24 ++++++++++++++++++- 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index abd4c9b46a7..3666647271e 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -243,7 +243,7 @@ object QueryGen { val table = spark.table(tableName) val op = Utils.randomChoice(Meta.unaryArithmeticOps, r) - val a = Utils.randomChoice(table.columns, r) + val a = pickRandomColumn(r, table, SparkNumericType) // Example SELECT a, -a FROM test0 s"SELECT $a, $op$a " + @@ -256,8 +256,8 @@ object QueryGen { val table = spark.table(tableName) val op = Utils.randomChoice(Meta.binaryArithmeticOps, r) - val a = Utils.randomChoice(table.columns, r) - val b = Utils.randomChoice(table.columns, r) + val a = pickRandomColumn(r, table, SparkNumericType) + val b = pickRandomColumn(r, table, SparkNumericType) // Example SELECT a, b, a+b FROM test0 s"SELECT $a, $b, $a $op $b " + @@ -270,8 +270,8 @@ object QueryGen { val table = spark.table(tableName) val op = Utils.randomChoice(Meta.comparisonOps, r) - val a = Utils.randomChoice(table.columns, r) - val b = Utils.randomChoice(table.columns, r) + val a = pickRandomColumn(r, table, SparkNumericType) + val b = pickRandomColumn(r, table, SparkNumericType) // Example SELECT a, b, a <=> b FROM test0 s"SELECT $a, $b, $a $op $b " + @@ -284,8 +284,12 @@ object QueryGen { val table = spark.table(tableName) val op = Utils.randomChoice(Meta.comparisonOps, r) - val a = Utils.randomChoice(table.columns, r) - val b = Utils.randomChoice(table.columns, r) + + // pick two columns with the same type + // TODO make this more comprehensive + val opType = Utils.randomChoice(Seq(SparkStringType, SparkNumericType, SparkDateType), r) + val a = pickRandomColumn(r, table, opType) + val b = pickRandomColumn(r, table, opType) // Example SELECT a, b, IF(a <=> b, 1, 2), CASE WHEN a <=> b THEN 1 ELSE 2 END FROM test0 s"SELECT $a, $b, $a $op $b, IF($a $op $b, 1, 2), CASE WHEN $a $op $b THEN 1 ELSE 2 END " + diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala index 835cfe410e9..ea3b9e5058d 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala @@ -34,6 +34,11 @@ object QueryRunner { filename: String, showFailedSparkQueries: Boolean = false): Unit = { + var queryCount = 0 + var invalidQueryCount = 0 + var cometFailureCount = 0 + var cometSuccessCount = 0 + val outputFilename = s"results-${System.currentTimeMillis()}.md" // scalastyle:off println println(s"Writing results to $outputFilename") @@ -56,7 +61,7 @@ object QueryRunner { querySource .getLines() .foreach(sql => { - + queryCount += 1 try { // execute with Spark spark.conf.set("spark.comet.enabled", "false") @@ -71,6 +76,7 @@ object QueryRunner { val cometRows = df.collect() val cometPlan = df.queryExecution.executedPlan.toString + var success = true if (sparkRows.length == cometRows.length) { var i = 0 while (i < sparkRows.length) { @@ -79,6 +85,7 @@ object QueryRunner { assert(l.length == r.length) for (j <- 0 until l.length) { if (!same(l(j), r(j))) { + success = false showSQL(w, sql) showPlans(w, sparkPlan, cometPlan) w.write(s"First difference at row $i:\n") @@ -90,15 +97,24 @@ object QueryRunner { i += 1 } } else { + success = false showSQL(w, sql) showPlans(w, sparkPlan, cometPlan) w.write( s"[ERROR] Spark produced ${sparkRows.length} rows and " + s"Comet produced ${cometRows.length} rows.\n") } + + if (success) { + cometSuccessCount += 1 + } else { + cometFailureCount += 1 + } + } catch { case e: Exception => // the query worked in Spark but failed in Comet, so this is likely a bug in Comet + cometFailureCount += 1 showSQL(w, sql) w.write(s"[ERROR] Query failed in Comet: ${e.getMessage}:\n") w.write("```\n") @@ -116,6 +132,7 @@ object QueryRunner { } catch { case e: Exception => // we expect many generated queries to be invalid + invalidQueryCount += 1 if (showFailedSparkQueries) { showSQL(w, sql) w.write(s"Query failed in Spark: ${e.getMessage}\n") @@ -123,6 +140,11 @@ object QueryRunner { } }) + w.write("# Summary\n") + w.write( + s"Total queries: $queryCount; Invalid queries: $invalidQueryCount; " + + s"Comet failed: $cometFailureCount; Comet succeeded: $cometSuccessCount\n") + } finally { w.close() querySource.close() From 1147374f92774e5222146dd3a68dc62bee0e88a3 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 16:48:51 -0600 Subject: [PATCH 22/28] save --- .../scala/org/apache/comet/fuzz/Meta.scala | 9 ++++++++ .../org/apache/comet/fuzz/QueryGen.scala | 22 ++++++++++++------- 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index f1235e75aeb..592835a3721 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -341,4 +341,13 @@ object Meta { val comparisonOps: Seq[String] = Seq("=", "<=>", ">", ">=", "<", "<=") + // TODO make this more comprehensive + val comparisonTypes: Seq[SparkType] = Seq( + SparkStringType, + SparkBinaryType, + SparkNumericType, + SparkDateType, + SparkTimestampType, + SparkArrayType(SparkTypeOneOf(Seq(SparkStringType, SparkNumericType, SparkDateType)))) + } diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala index 3666647271e..d9e3c147d25 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryGen.scala @@ -187,6 +187,13 @@ object QueryGen { } } + def pickTwoRandomColumns(r: Random, df: DataFrame, targetType: SparkType): (String, String) = { + val a = pickRandomColumn(r, df, targetType) + val df2 = df.drop(a) + val b = pickRandomColumn(r, df2, targetType) + (a, b) + } + /** Select a random field that matches a predicate */ private def select(r: Random, df: DataFrame, predicate: StructField => Boolean): String = { val candidates = df.schema.fields.filter(predicate) @@ -256,8 +263,7 @@ object QueryGen { val table = spark.table(tableName) val op = Utils.randomChoice(Meta.binaryArithmeticOps, r) - val a = pickRandomColumn(r, table, SparkNumericType) - val b = pickRandomColumn(r, table, SparkNumericType) + val (a, b) = pickTwoRandomColumns(r, table, SparkNumericType) // Example SELECT a, b, a+b FROM test0 s"SELECT $a, $b, $a $op $b " + @@ -270,8 +276,10 @@ object QueryGen { val table = spark.table(tableName) val op = Utils.randomChoice(Meta.comparisonOps, r) - val a = pickRandomColumn(r, table, SparkNumericType) - val b = pickRandomColumn(r, table, SparkNumericType) + + // pick two columns with the same type + val opType = Utils.randomChoice(Meta.comparisonTypes, r) + val (a, b) = pickTwoRandomColumns(r, table, opType) // Example SELECT a, b, a <=> b FROM test0 s"SELECT $a, $b, $a $op $b " + @@ -286,10 +294,8 @@ object QueryGen { val op = Utils.randomChoice(Meta.comparisonOps, r) // pick two columns with the same type - // TODO make this more comprehensive - val opType = Utils.randomChoice(Seq(SparkStringType, SparkNumericType, SparkDateType), r) - val a = pickRandomColumn(r, table, opType) - val b = pickRandomColumn(r, table, opType) + val opType = Utils.randomChoice(Meta.comparisonTypes, r) + val (a, b) = pickTwoRandomColumns(r, table, opType) // Example SELECT a, b, IF(a <=> b, 1, 2), CASE WHEN a <=> b THEN 1 ELSE 2 END FROM test0 s"SELECT $a, $b, $a $op $b, IF($a $op $b, 1, 2), CASE WHEN $a $op $b THEN 1 ELSE 2 END " + From f0b89a26eabc320c8a6b7c975f522e79d96f5b8e Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 17:32:23 -0600 Subject: [PATCH 23/28] save --- fuzz-testing/README.md | 3 ++ .../scala/org/apache/comet/fuzz/Main.scala | 18 ++++++++++ .../org/apache/comet/fuzz/QueryRunner.scala | 10 ++++++ .../comet/testing/FuzzDataGenerator.scala | 36 +++++++++---------- 4 files changed, 48 insertions(+), 19 deletions(-) diff --git a/fuzz-testing/README.md b/fuzz-testing/README.md index 966c5df4c06..832562b329e 100644 --- a/fuzz-testing/README.md +++ b/fuzz-testing/README.md @@ -94,6 +94,9 @@ $SPARK_HOME/bin/spark-submit \ --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \ --conf spark.comet.exec.shuffle.enabled=true \ --conf spark.comet.exec.shuffle.mode=auto \ + --conf spark.sql.parquet.inferTimestampNTZ.enabled=false \ + --conf spark.sql.session.timeZone="America/Denver" \ + --conf spark.sql.readSideCharPadding=false \ --jars $COMET_JAR \ --conf spark.driver.extraClassPath=$COMET_JAR \ --conf spark.executor.extraClassPath=$COMET_JAR \ diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Main.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Main.scala index 9b9a4b6f3e5..9ae699bdbdf 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Main.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Main.scala @@ -25,6 +25,7 @@ import org.rogach.scallop.{ScallopConf, Subcommand} import org.rogach.scallop.ScallopOption import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.types.DataTypes import org.apache.comet.testing.{DataGenOptions, ParquetGenerator} @@ -78,7 +79,24 @@ object Main { case Some(seed) => new Random(seed) case None => new Random() } + // create two columns of each primitive type + val x = Seq( + DataTypes.BooleanType, + DataTypes.ByteType, + DataTypes.ShortType, + DataTypes.IntegerType, + DataTypes.LongType, + DataTypes.FloatType, + DataTypes.DoubleType, + DataTypes.createDecimalType(10, 2), + DataTypes.createDecimalType(36, 18), + DataTypes.DateType, + DataTypes.TimestampType, + DataTypes.TimestampNTZType, + DataTypes.StringType, + DataTypes.BinaryType) val options = DataGenOptions( + primitiveTypes = x ++ x, allowNull = true, generateArray = conf.generateData.generateArrays(), generateStruct = conf.generateData.generateStructs(), diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala index ea3b9e5058d..0fc079d21f6 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala @@ -105,6 +105,13 @@ object QueryRunner { s"Comet produced ${cometRows.length} rows.\n") } + // check that the plan contains Comet operators + if (!cometPlan.contains("Comet")) { + success = false + showSQL(w, sql) + showPlans(w, sparkPlan, cometPlan) + } + if (success) { cometSuccessCount += 1 } else { @@ -116,6 +123,9 @@ object QueryRunner { // the query worked in Spark but failed in Comet, so this is likely a bug in Comet cometFailureCount += 1 showSQL(w, sql) + w.write("### Spark Plan\n") + w.write(s"```\n$sparkPlan\n```\n") + w.write(s"[ERROR] Query failed in Comet: ${e.getMessage}:\n") w.write("```\n") val sw = new StringWriter() diff --git a/spark/src/main/scala/org/apache/comet/testing/FuzzDataGenerator.scala b/spark/src/main/scala/org/apache/comet/testing/FuzzDataGenerator.scala index 7c7a6727fb0..20f961ba95e 100644 --- a/spark/src/main/scala/org/apache/comet/testing/FuzzDataGenerator.scala +++ b/spark/src/main/scala/org/apache/comet/testing/FuzzDataGenerator.scala @@ -44,24 +44,7 @@ object FuzzDataGenerator { val defaultBaseDate: Long = new SimpleDateFormat("YYYY-MM-DD hh:mm:ss").parse("3333-05-25 12:34:56").getTime - private val primitiveTypes = Seq( - DataTypes.BooleanType, - DataTypes.ByteType, - DataTypes.ShortType, - DataTypes.IntegerType, - DataTypes.LongType, - DataTypes.FloatType, - DataTypes.DoubleType, - DataTypes.createDecimalType(10, 2), - DataTypes.createDecimalType(36, 18), - DataTypes.DateType, - DataTypes.TimestampType, - DataTypes.TimestampNTZType, - DataTypes.StringType, - DataTypes.BinaryType) - - private def filteredPrimitives(excludeTypes: Seq[DataType]) = { - + private def filteredPrimitives(primitiveTypes: Seq[DataType], excludeTypes: Seq[DataType]) = { primitiveTypes.filterNot { dataType => excludeTypes.exists { case _: DecimalType => @@ -79,7 +62,7 @@ object FuzzDataGenerator { numRows: Int, options: DataGenOptions): DataFrame = { - val filteredPrimitiveTypes = filteredPrimitives(options.excludeTypes) + val filteredPrimitiveTypes = filteredPrimitives(options.primitiveTypes, options.excludeTypes) val dataTypes = ListBuffer[DataType]() dataTypes.appendAll(filteredPrimitiveTypes) @@ -248,6 +231,21 @@ object FuzzDataGenerator { } case class DataGenOptions( + primitiveTypes: Seq[DataType] = Seq( + DataTypes.BooleanType, + DataTypes.ByteType, + DataTypes.ShortType, + DataTypes.IntegerType, + DataTypes.LongType, + DataTypes.FloatType, + DataTypes.DoubleType, + DataTypes.createDecimalType(10, 2), + DataTypes.createDecimalType(36, 18), + DataTypes.DateType, + DataTypes.TimestampType, + DataTypes.TimestampNTZType, + DataTypes.StringType, + DataTypes.BinaryType), allowNull: Boolean = true, generateNegativeZero: Boolean = true, baseDate: Long = FuzzDataGenerator.defaultBaseDate, From bb0dfa10ff799112dca2d6df356dd817f4440a48 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 17:48:21 -0600 Subject: [PATCH 24/28] skip first/last --- fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index 592835a3721..a0354e8475a 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -321,8 +321,9 @@ object Meta { createFunctionWithInputs("count", Seq(SparkAnyType)), createUnaryNumericFunction("avg"), createUnaryNumericFunction("sum"), - createFunctionWithInputs("first", Seq(SparkAnyType)), - createFunctionWithInputs("last", Seq(SparkAnyType)), + // first/last are non-deterministic and known to be incompatible with Spark +// createFunctionWithInputs("first", Seq(SparkAnyType)), +// createFunctionWithInputs("last", Seq(SparkAnyType)), createFunctionWithInputs("var_pop", Seq(SparkNumericType)), createFunctionWithInputs("var_samp", Seq(SparkNumericType)), createFunctionWithInputs("covar_pop", Seq(SparkNumericType, SparkNumericType)), From 28b05098058535da4c161b43153790bec9a3e613 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 21 Oct 2025 18:31:18 -0600 Subject: [PATCH 25/28] offheap --- fuzz-testing/README.md | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/fuzz-testing/README.md b/fuzz-testing/README.md index 832562b329e..c8cea5be827 100644 --- a/fuzz-testing/README.md +++ b/fuzz-testing/README.md @@ -88,15 +88,12 @@ Note that the output filename is currently hard-coded as `queries.sql` ```shell $SPARK_HOME/bin/spark-submit \ --master $SPARK_MASTER \ + --conf spark.memory.offHeap.enabled=true \ + --conf spark.memory.offHeap.size=16G \ --conf spark.plugins=org.apache.spark.CometPlugin \ --conf spark.comet.enabled=true \ - --conf spark.comet.exec.enabled=true \ --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \ --conf spark.comet.exec.shuffle.enabled=true \ - --conf spark.comet.exec.shuffle.mode=auto \ - --conf spark.sql.parquet.inferTimestampNTZ.enabled=false \ - --conf spark.sql.session.timeZone="America/Denver" \ - --conf spark.sql.readSideCharPadding=false \ --jars $COMET_JAR \ --conf spark.driver.extraClassPath=$COMET_JAR \ --conf spark.executor.extraClassPath=$COMET_JAR \ From 7c38dfa06a81e416958a96df6c57b3268faa7786 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 22 Oct 2025 15:18:58 -0600 Subject: [PATCH 26/28] address feedback --- .../scala/org/apache/comet/fuzz/Meta.scala | 200 +++++++++--------- 1 file changed, 102 insertions(+), 98 deletions(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala index a0354e8475a..74d13f85ee7 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/Meta.scala @@ -65,7 +65,7 @@ object Meta { (DataTypes.StringType, 0.2), (DataTypes.BinaryType, 0.1)) - private def createFunctionWithInputs(name: String, inputs: Seq[SparkType]): Function = { + private def createFunctionWithInputTypes(name: String, inputs: Seq[SparkType]): Function = { Function(name, Seq(FunctionSignature(inputs))) } @@ -74,11 +74,11 @@ object Meta { } private def createUnaryStringFunction(name: String): Function = { - createFunctionWithInputs(name, Seq(SparkStringType)) + createFunctionWithInputTypes(name, Seq(SparkStringType)) } private def createUnaryNumericFunction(name: String): Function = { - createFunctionWithInputs(name, Seq(SparkNumericType)) + createFunctionWithInputTypes(name, Seq(SparkNumericType)) } // Math expressions (corresponds to mathExpressions in QueryPlanSerde) @@ -87,15 +87,15 @@ object Meta { createUnaryNumericFunction("acos"), createUnaryNumericFunction("asin"), createUnaryNumericFunction("atan"), - createFunctionWithInputs("atan2", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputTypes("atan2", Seq(SparkNumericType, SparkNumericType)), createUnaryNumericFunction("cos"), createUnaryNumericFunction("exp"), createUnaryNumericFunction("expm1"), - createFunctionWithInputs("log", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputTypes("log", Seq(SparkNumericType, SparkNumericType)), createUnaryNumericFunction("log10"), createUnaryNumericFunction("log2"), - createFunctionWithInputs("pow", Seq(SparkNumericType, SparkNumericType)), - createFunctionWithInputs("remainder", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputTypes("pow", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputTypes("remainder", Seq(SparkNumericType, SparkNumericType)), createFunctions( "round", Seq( @@ -107,20 +107,20 @@ object Meta { createUnaryNumericFunction("tan"), createUnaryNumericFunction("ceil"), createUnaryNumericFunction("floor"), - createFunctionWithInputs("unary_minus", Seq(SparkNumericType))) + createFunctionWithInputTypes("unary_minus", Seq(SparkNumericType))) // Hash expressions (corresponds to hashExpressions in QueryPlanSerde) val hashScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("md5", Seq(SparkAnyType)), - createFunctionWithInputs("murmur3_hash", Seq(SparkAnyType)), // TODO can take multiple columns - createFunctionWithInputs("sha2", Seq(SparkAnyType, SparkIntType))) + createFunctionWithInputTypes("md5", Seq(SparkAnyType)), + createFunctionWithInputTypes("murmur3_hash", Seq(SparkAnyType)), // TODO variadic + createFunctionWithInputTypes("sha2", Seq(SparkAnyType, SparkIntType))) // String expressions (corresponds to stringExpressions in QueryPlanSerde) val stringScalarFunc: Seq[Function] = Seq( createUnaryStringFunction("ascii"), createUnaryStringFunction("bit_length"), createUnaryStringFunction("chr"), - createFunctionWithInputs( + createFunctionWithInputTypes( "concat", Seq( SparkTypeOneOf( @@ -137,18 +137,18 @@ object Meta { SparkBinaryType, SparkArrayType( SparkTypeOneOf(Seq(SparkStringType, SparkNumericType, SparkBinaryType))))))), - createFunctionWithInputs("concat_ws", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("contains", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("ends_with", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs( + createFunctionWithInputTypes("concat_ws", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes("contains", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes("ends_with", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes( "hex", Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType, SparkIntType, SparkLongType)))), createUnaryStringFunction("init_cap"), - createFunctionWithInputs("instr", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs( + createFunctionWithInputTypes("instr", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes( "length", Seq(SparkTypeOneOf(Seq(SparkStringType, SparkBinaryType)))), - createFunctionWithInputs("like", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes("like", Seq(SparkStringType, SparkStringType)), createUnaryStringFunction("lower"), createFunctions( "lpad", @@ -162,7 +162,7 @@ object Meta { Seq( FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType)), FunctionSignature(Seq(SparkStringType, SparkStringType, SparkStringType, SparkIntType)))), - createFunctionWithInputs("repeat", Seq(SparkStringType, SparkIntType)), + createFunctionWithInputTypes("repeat", Seq(SparkStringType, SparkIntType)), createFunctions( "replace", Seq( @@ -173,91 +173,93 @@ object Meta { Seq( FunctionSignature(Seq(SparkStringType)), FunctionSignature(Seq(SparkArrayType(SparkAnyType))))), - createFunctionWithInputs("rlike", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes("rlike", Seq(SparkStringType, SparkStringType)), createFunctions( "rpad", Seq( FunctionSignature(Seq(SparkStringType, SparkIntegralType)), FunctionSignature(Seq(SparkStringType, SparkIntegralType, SparkStringType)))), createUnaryStringFunction("rtrim"), - createFunctionWithInputs("starts_with", Seq(SparkStringType, SparkStringType)), - createFunctionWithInputs("string_space", Seq(SparkIntType)), - createFunctionWithInputs("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), - createFunctionWithInputs("translate", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes("starts_with", Seq(SparkStringType, SparkStringType)), + createFunctionWithInputTypes("string_space", Seq(SparkIntType)), + createFunctionWithInputTypes("substring", Seq(SparkStringType, SparkIntType, SparkIntType)), + createFunctionWithInputTypes("translate", Seq(SparkStringType, SparkStringType)), createUnaryStringFunction("trim"), createUnaryStringFunction("btrim"), createUnaryStringFunction("unhex"), createUnaryStringFunction("upper"), - createFunctionWithInputs("xxhash64", Seq(SparkAnyType)), // TODO can take multiple columns - createFunctionWithInputs("sha1", Seq(SparkAnyType))) + createFunctionWithInputTypes("xxhash64", Seq(SparkAnyType)), // TODO variadic + createFunctionWithInputTypes("sha1", Seq(SparkAnyType))) // Conditional expressions (corresponds to conditionalExpressions in QueryPlanSerde) val conditionalScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("if", Seq(SparkBooleanType, SparkAnyType, SparkAnyType))) + createFunctionWithInputTypes("if", Seq(SparkBooleanType, SparkAnyType, SparkAnyType))) // Map expressions (corresponds to mapExpressions in QueryPlanSerde) val mapScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs( + createFunctionWithInputTypes( "map_extract", Seq(SparkMapType(SparkAnyType, SparkAnyType), SparkAnyType)), - createFunctionWithInputs("map_keys", Seq(SparkMapType(SparkAnyType, SparkAnyType))), - createFunctionWithInputs("map_entries", Seq(SparkMapType(SparkAnyType, SparkAnyType))), - createFunctionWithInputs("map_values", Seq(SparkMapType(SparkAnyType, SparkAnyType))), - createFunctionWithInputs( + createFunctionWithInputTypes("map_keys", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputTypes("map_entries", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputTypes("map_values", Seq(SparkMapType(SparkAnyType, SparkAnyType))), + createFunctionWithInputTypes( "map_from_arrays", Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType)))) // Predicate expressions (corresponds to predicateExpressions in QueryPlanSerde) val predicateScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("and", Seq(SparkBooleanType, SparkBooleanType)), - createFunctionWithInputs("or", Seq(SparkBooleanType, SparkBooleanType)), - createFunctionWithInputs("not", Seq(SparkBooleanType)), - createFunctionWithInputs("in", Seq(SparkAnyType, SparkAnyType)) + createFunctionWithInputTypes("and", Seq(SparkBooleanType, SparkBooleanType)), + createFunctionWithInputTypes("or", Seq(SparkBooleanType, SparkBooleanType)), + createFunctionWithInputTypes("not", Seq(SparkBooleanType)), + createFunctionWithInputTypes("in", Seq(SparkAnyType, SparkAnyType)) ) // TODO: variadic // Struct expressions (corresponds to structExpressions in QueryPlanSerde) val structScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs( + createFunctionWithInputTypes( "create_named_struct", Seq(SparkStringType, SparkAnyType) ), // TODO: variadic name/value pairs - createFunctionWithInputs( + createFunctionWithInputTypes( "get_struct_field", Seq(SparkStructType(Seq(SparkAnyType)), SparkStringType))) // Bitwise expressions (corresponds to bitwiseExpressions in QueryPlanSerde) val bitwiseScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("bitwise_and", Seq(SparkIntegralType, SparkIntegralType)), - createFunctionWithInputs("bitwise_count", Seq(SparkIntegralType)), - createFunctionWithInputs("bitwise_get", Seq(SparkIntegralType, SparkIntType)), - createFunctionWithInputs("bitwise_or", Seq(SparkIntegralType, SparkIntegralType)), - createFunctionWithInputs("bitwise_not", Seq(SparkIntegralType)), - createFunctionWithInputs("bitwise_xor", Seq(SparkIntegralType, SparkIntegralType)), - createFunctionWithInputs("shift_left", Seq(SparkIntegralType, SparkIntType)), - createFunctionWithInputs("shift_right", Seq(SparkIntegralType, SparkIntType))) + createFunctionWithInputTypes("bitwise_and", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputTypes("bitwise_count", Seq(SparkIntegralType)), + createFunctionWithInputTypes("bitwise_get", Seq(SparkIntegralType, SparkIntType)), + createFunctionWithInputTypes("bitwise_or", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputTypes("bitwise_not", Seq(SparkIntegralType)), + createFunctionWithInputTypes("bitwise_xor", Seq(SparkIntegralType, SparkIntegralType)), + createFunctionWithInputTypes("shift_left", Seq(SparkIntegralType, SparkIntType)), + createFunctionWithInputTypes("shift_right", Seq(SparkIntegralType, SparkIntType))) // Misc expressions (corresponds to miscExpressions in QueryPlanSerde) val miscScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("isnan", Seq(SparkNumericType)), - createFunctionWithInputs("isnull", Seq(SparkAnyType)), - createFunctionWithInputs("isnotnull", Seq(SparkAnyType)), - createFunctionWithInputs("coalesce", Seq(SparkAnyType, SparkAnyType)) + createFunctionWithInputTypes("isnan", Seq(SparkNumericType)), + createFunctionWithInputTypes("isnull", Seq(SparkAnyType)), + createFunctionWithInputTypes("isnotnull", Seq(SparkAnyType)), + createFunctionWithInputTypes("coalesce", Seq(SparkAnyType, SparkAnyType)) ) // TODO: variadic // Array expressions (corresponds to arrayExpressions in QueryPlanSerde) val arrayScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("array_append", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), - createFunctionWithInputs("array_compact", Seq(SparkArrayType(SparkAnyType))), - createFunctionWithInputs("array_contains", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), - createFunctionWithInputs("array_distinct", Seq(SparkArrayType(SparkAnyType))), - createFunctionWithInputs( + createFunctionWithInputTypes("array_append", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputTypes("array_compact", Seq(SparkArrayType(SparkAnyType))), + createFunctionWithInputTypes( + "array_contains", + Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputTypes("array_distinct", Seq(SparkArrayType(SparkAnyType))), + createFunctionWithInputTypes( "array_except", Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), - createFunctionWithInputs( + createFunctionWithInputTypes( "array_insert", Seq(SparkArrayType(SparkAnyType), SparkIntType, SparkAnyType)), - createFunctionWithInputs( + createFunctionWithInputTypes( "array_intersect", Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), createFunctions( @@ -265,49 +267,51 @@ object Meta { Seq( FunctionSignature(Seq(SparkArrayType(SparkAnyType), SparkStringType)), FunctionSignature(Seq(SparkArrayType(SparkAnyType), SparkStringType, SparkStringType)))), - createFunctionWithInputs("array_max", Seq(SparkArrayType(SparkAnyType))), - createFunctionWithInputs("array_min", Seq(SparkArrayType(SparkAnyType))), - createFunctionWithInputs("array_remove", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), - createFunctionWithInputs("array_repeat", Seq(SparkAnyType, SparkIntType)), - createFunctionWithInputs( + createFunctionWithInputTypes("array_max", Seq(SparkArrayType(SparkAnyType))), + createFunctionWithInputTypes("array_min", Seq(SparkArrayType(SparkAnyType))), + createFunctionWithInputTypes("array_remove", Seq(SparkArrayType(SparkAnyType), SparkAnyType)), + createFunctionWithInputTypes("array_repeat", Seq(SparkAnyType, SparkIntType)), + createFunctionWithInputTypes( "arrays_overlap", Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), - createFunctionWithInputs( + createFunctionWithInputTypes( "array_union", Seq(SparkArrayType(SparkAnyType), SparkArrayType(SparkAnyType))), - createFunctionWithInputs("array", Seq(SparkAnyType, SparkAnyType)), // TODO: variadic - createFunctionWithInputs( + createFunctionWithInputTypes("array", Seq(SparkAnyType, SparkAnyType)), // TODO: variadic + createFunctionWithInputTypes( "element_at", Seq( SparkTypeOneOf( Seq(SparkArrayType(SparkAnyType), SparkMapType(SparkAnyType, SparkAnyType))), SparkAnyType)), - createFunctionWithInputs("flatten", Seq(SparkArrayType(SparkArrayType(SparkAnyType)))), - createFunctionWithInputs("get_array_item", Seq(SparkArrayType(SparkAnyType), SparkIntType))) + createFunctionWithInputTypes("flatten", Seq(SparkArrayType(SparkArrayType(SparkAnyType)))), + createFunctionWithInputTypes( + "get_array_item", + Seq(SparkArrayType(SparkAnyType), SparkIntType))) // Temporal expressions (corresponds to temporalExpressions in QueryPlanSerde) val temporalScalarFunc: Seq[Function] = Seq( - createFunctionWithInputs("date_add", Seq(SparkDateType, SparkIntType)), - createFunctionWithInputs("date_sub", Seq(SparkDateType, SparkIntType)), + createFunctionWithInputTypes("date_add", Seq(SparkDateType, SparkIntType)), + createFunctionWithInputTypes("date_sub", Seq(SparkDateType, SparkIntType)), createFunctions( "from_unixtime", Seq( FunctionSignature(Seq(SparkLongType)), FunctionSignature(Seq(SparkLongType, SparkStringType)))), - createFunctionWithInputs("hour", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("minute", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("second", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("trunc", Seq(SparkDateOrTimestampType, SparkStringType)), - createFunctionWithInputs("year", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("month", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("day", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("dayofmonth", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("dayofweek", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("weekday", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("dayofyear", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("weekofyear", Seq(SparkDateOrTimestampType)), - createFunctionWithInputs("quarter", Seq(SparkDateOrTimestampType))) + createFunctionWithInputTypes("hour", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("minute", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("second", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("trunc", Seq(SparkDateOrTimestampType, SparkStringType)), + createFunctionWithInputTypes("year", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("month", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("day", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("dayofmonth", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("dayofweek", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("weekday", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("dayofyear", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("weekofyear", Seq(SparkDateOrTimestampType)), + createFunctionWithInputTypes("quarter", Seq(SparkDateOrTimestampType))) // Combined in same order as exprSerdeMap in QueryPlanSerde val scalarFunc: Seq[Function] = mathScalarFunc ++ hashScalarFunc ++ stringScalarFunc ++ @@ -316,24 +320,24 @@ object Meta { temporalScalarFunc val aggFunc: Seq[Function] = Seq( - createFunctionWithInputs("min", Seq(SparkAnyType)), - createFunctionWithInputs("max", Seq(SparkAnyType)), - createFunctionWithInputs("count", Seq(SparkAnyType)), + createFunctionWithInputTypes("min", Seq(SparkAnyType)), + createFunctionWithInputTypes("max", Seq(SparkAnyType)), + createFunctionWithInputTypes("count", Seq(SparkAnyType)), createUnaryNumericFunction("avg"), createUnaryNumericFunction("sum"), // first/last are non-deterministic and known to be incompatible with Spark -// createFunctionWithInputs("first", Seq(SparkAnyType)), -// createFunctionWithInputs("last", Seq(SparkAnyType)), - createFunctionWithInputs("var_pop", Seq(SparkNumericType)), - createFunctionWithInputs("var_samp", Seq(SparkNumericType)), - createFunctionWithInputs("covar_pop", Seq(SparkNumericType, SparkNumericType)), - createFunctionWithInputs("covar_samp", Seq(SparkNumericType, SparkNumericType)), - createFunctionWithInputs("stddev_pop", Seq(SparkNumericType)), - createFunctionWithInputs("stddev_samp", Seq(SparkNumericType)), - createFunctionWithInputs("corr", Seq(SparkNumericType, SparkNumericType)), - createFunctionWithInputs("bit_and", Seq(SparkIntegralType)), - createFunctionWithInputs("bit_or", Seq(SparkIntegralType)), - createFunctionWithInputs("bit_xor", Seq(SparkIntegralType))) +// createFunctionWithInputTypes("first", Seq(SparkAnyType)), +// createFunctionWithInputTypes("last", Seq(SparkAnyType)), + createUnaryNumericFunction("var_pop"), + createUnaryNumericFunction("var_samp"), + createFunctionWithInputTypes("covar_pop", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputTypes("covar_samp", Seq(SparkNumericType, SparkNumericType)), + createUnaryNumericFunction("stddev_pop"), + createUnaryNumericFunction("stddev_samp"), + createFunctionWithInputTypes("corr", Seq(SparkNumericType, SparkNumericType)), + createFunctionWithInputTypes("bit_and", Seq(SparkIntegralType)), + createFunctionWithInputTypes("bit_or", Seq(SparkIntegralType)), + createFunctionWithInputTypes("bit_xor", Seq(SparkIntegralType))) val unaryArithmeticOps: Seq[String] = Seq("+", "-") From 871b5716f700fc8745edb65f7d617c57a43088ad Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 22 Oct 2025 15:20:14 -0600 Subject: [PATCH 27/28] improve reporting --- .../src/main/scala/org/apache/comet/fuzz/QueryRunner.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala index 0fc079d21f6..16ee74678f1 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala @@ -110,6 +110,7 @@ object QueryRunner { success = false showSQL(w, sql) showPlans(w, sparkPlan, cometPlan) + w.write(s"[ERROR] Comet did not accelerate any part of the plan\n") } if (success) { From 605842776cce10c3aa0784513c1ac48a9b37e0b0 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 22 Oct 2025 15:32:30 -0600 Subject: [PATCH 28/28] format --- .../src/main/scala/org/apache/comet/fuzz/QueryRunner.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala index 16ee74678f1..bcc9f98d063 100644 --- a/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala +++ b/fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala @@ -110,7 +110,7 @@ object QueryRunner { success = false showSQL(w, sql) showPlans(w, sparkPlan, cometPlan) - w.write(s"[ERROR] Comet did not accelerate any part of the plan\n") + w.write("[ERROR] Comet did not accelerate any part of the plan\n") } if (success) {