feat: expose Comet version as spark.comet.version runtime config

docs/source/user-guide/latest/installation.md

@@ -166,6 +166,33 @@ CometNativeColumnarToRow
    +- CometNativeScan parquet [a#6] Batched: true, DataFilters: [isnotnull(a#6), (a#6 > 5)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/test], PartitionFilters: [], PushedFilters: [IsNotNull(a), GreaterThan(a,5)], ReadSchema: struct<a:int>
 ```
 
+## Checking the Comet Version
+
+When the Comet plugin is loaded, it exposes its build version as the Spark config
+`spark.comet.version`. This can be queried at runtime from any supported language, for example:
+
+```scala
+scala> spark.conf.get("spark.comet.version")
+```
+
+```sql
+SET spark.comet.version;
+```
+
+The same value is available programmatically on the JVM classpath, along with additional build
+metadata that is useful when reporting issues:
+
+```scala
+scala> import org.apache.comet.{COMET_VERSION, COMET_BRANCH, COMET_REVISION}
+scala> println(COMET_VERSION)
+```
+
+Comet also logs its version when the native library is initialized:
+
+```shell
+INFO core/src/lib.rs: Comet native library version <version> initialized
+```
+
 ## Additional Configuration
 
 Depending on your deployment mode you may also need to set the driver & executor class path(s) to

spark/src/main/scala/org/apache/spark/Plugins.scala

@@ -28,7 +28,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.{EXECUTOR_MEMORY, EXECUTOR_MEMORY_OVERHEAD, EXECUTOR_MEMORY_OVERHEAD_FACTOR}
 import org.apache.spark.sql.internal.StaticSQLConf
 
-import org.apache.comet.{CometSparkSessionExtensions, NativeBase}
+import org.apache.comet.{COMET_VERSION, CometSparkSessionExtensions, NativeBase}
 import org.apache.comet.CometConf.{COMET_METRICS_ENABLED, COMET_ONHEAP_ENABLED}
 
 /**
@@ -48,6 +48,12 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl
   override def init(sc: SparkContext, pluginContext: PluginContext): ju.Map[String, String] = {
     logInfo("CometDriverPlugin init")
 
+    // Expose the Comet build version as a Spark config so it can be queried at runtime, e.g.
+    // `spark.conf.get("spark.comet.version")` or `SET spark.comet.version` in SQL. This is set
+    // before the off-heap check below so the version is reported even when Comet is otherwise
+    // disabled.
+    sc.conf.set(CometDriverPlugin.COMET_VERSION_CONFIG, COMET_VERSION)
+
     if (!CometSparkSessionExtensions.isOffHeapEnabled(sc.getConf) &&
       !sc.getConf.getBoolean(COMET_ONHEAP_ENABLED.key, false)) {
       logWarning("Comet plugin is disabled because Spark is not running in off-heap mode.")
@@ -106,6 +112,10 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl
 }
 
 object CometDriverPlugin extends Logging {
+
+  /** Spark config key under which the loaded Comet version is exposed at runtime. */
+  val COMET_VERSION_CONFIG = "spark.comet.version"
+
   def registerCometMetrics(sc: SparkContext): Unit = {
     if (sc.getConf.getBoolean(
         COMET_METRICS_ENABLED.key,

spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala

@@ -24,6 +24,8 @@ import java.io.File
 import org.apache.spark.sql.{CometTestBase, SaveMode}
 import org.apache.spark.sql.internal.StaticSQLConf
 
+import org.apache.comet.COMET_VERSION
+
 class CometPluginsSuite extends CometTestBase {
   override protected def sparkConf: SparkConf = {
     val conf = new SparkConf()
@@ -83,6 +85,13 @@ class CometPluginsSuite extends CometTestBase {
     }
   }
 
+  test("Comet version is exposed as a Spark config") {
+    // The driver plugin sets spark.comet.version, which is then visible both on the SparkContext
+    // conf and through the session runtime config (SET / spark.conf.get).
+    assert(spark.sparkContext.conf.get(CometDriverPlugin.COMET_VERSION_CONFIG) == COMET_VERSION)
+    assert(spark.conf.get(CometDriverPlugin.COMET_VERSION_CONFIG) == COMET_VERSION)
+  }
+
   test("CometSource metrics are recorded") {
     val nativeBefore = CometSource.NATIVE_OPERATORS.getCount
     val queriesBefore = CometSource.QUERIES_PLANNED.getCount