Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,15 @@ import org.apache.spark.SparkConf
import org.apache.spark.internal.Logging
import org.apache.spark.network.util.ByteUnit
import org.apache.spark.sql.{SparkSession, SparkSessionExtensions}
import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionInfo, Literal}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreeNode
import org.apache.spark.sql.comet._
import org.apache.spark.sql.execution._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StringType
import org.apache.spark.unsafe.types.UTF8String

import org.apache.comet.CometConf._
import org.apache.comet.rules.{CometExecRule, CometPlanAdaptiveDynamicPruningFilters, CometReuseSubquery, CometScanRule, CometSpark34AqeDppFallbackRule, EliminateRedundantTransitions, RevertNativeForTransitionHeavyStages}
Expand Down Expand Up @@ -99,6 +103,7 @@ class CometSparkSessionExtensions
extensions.injectQueryStagePrepRule { session => CometExecRule(session) }
injectQueryStageOptimizerRuleShim(extensions, CometPlanAdaptiveDynamicPruningFilters)
injectQueryStageOptimizerRuleShim(extensions, CometReuseSubquery)
extensions.injectFunction(CometSparkSessionExtensions.cometVersionFunction)
}

case class CometScanColumnar(session: SparkSession) extends ColumnarRule {
Expand All @@ -119,6 +124,40 @@ class CometSparkSessionExtensions
object CometSparkSessionExtensions extends Logging {
lazy val isBigEndian: Boolean = ByteOrder.nativeOrder().equals(ByteOrder.BIG_ENDIAN)

/**
* SQL function `comet_version()` returning the Comet build version as a string. Registered via
* `SparkSessionExtensions.injectFunction` so users can query the loaded Comet version, e.g.
* `SELECT comet_version()`.
*/
private[comet] val cometVersionFunction
: (FunctionIdentifier, ExpressionInfo, Seq[Expression] => Expression) = {
val name = "comet_version"
val versionLiteral = Literal(UTF8String.fromString(COMET_VERSION), StringType)
// Populate the full ExpressionInfo so that Spark's ExpressionInfoSuite (SPARK-32870, and the
// example-output check) is satisfied for this injected function. The example output is derived
// from COMET_VERSION so it matches the value the function actually returns across releases.
val info = new ExpressionInfo(
classOf[Literal].getCanonicalName,
"",
name,
"_FUNC_() - Returns the Apache DataFusion Comet version.",
"",
s"\n Examples:\n > SELECT _FUNC_();\n $COMET_VERSION\n ",
"",
"misc_funcs",
"1.0.0",
"",
"")
val builder = (args: Seq[Expression]) => {
if (args.nonEmpty) {
throw new IllegalArgumentException(
s"Function $name does not accept arguments (got ${args.length})")
}
versionLiteral
}
(FunctionIdentifier(name), info, builder)
}

/**
* Checks whether Comet extension should be loaded for Spark.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -130,4 +130,20 @@ class CometSparkSessionExtensionsSuite extends CometTestBase {
getCometShuffleMemorySize(conf, sqlConf) ==
getBytesFromMib((1024 * 0.2).toLong))
}

test("comet_version() returns the Comet build version") {
withSQLConf(CometConf.COMET_EXEC_ENABLED.key -> "false") {
val result = spark.sql("SELECT comet_version()").collect()
assert(result.length == 1)
val version = result(0).getString(0)
assert(version == COMET_VERSION, s"Expected $COMET_VERSION but got $version")
}
}

test("comet_version() rejects arguments") {
val ex = intercept[Exception] {
spark.sql("SELECT comet_version(1)").collect()
}
assert(ex.getMessage.contains("comet_version"))
}
}
Loading