fix: backtick-escape column names in pruneColumns to prevent V2 DataSource function-name collision (#748)

Co-authored-by: Kyra Mangasarian <kyraman@amazon.com>

Author: kyraman

src/main/scala/com/amazon/deequ/analyzers/runners/AnalysisRunner.scala

@@ -20,6 +20,7 @@ import com.amazon.deequ.analyzers._
 import com.amazon.deequ.io.DfsUtils
 import com.amazon.deequ.metrics.{DoubleMetric, Metric}
 import com.amazon.deequ.repository.{MetricsRepository, ResultKey}
+import com.amazon.deequ.utilities.ColumnUtil.escapeColumn
 import org.apache.spark.sql.Column
 import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.types.StructType
@@ -409,7 +410,7 @@ object AnalysisRunner {
         // All analyzers are dataset-level (e.g. Size), no column selection needed
         data
       } else {
-        data.select(neededColumns.map(col): _*)
+        data.select(neededColumns.map(c => col(escapeColumn(c))): _*)
       }
     }
   }

src/test/scala/com/amazon/deequ/analyzers/runners/AnalysisRunnerTests.scala

@@ -532,5 +532,32 @@ class AnalysisRunnerTests extends AnyWordSpec
         assert(result.metric(mean).get.value === mean.calculate(data).value)
         assert(result.metric(minimum).get.value === minimum.calculate(data).value)
       }
+
+    "handle column pruning when column name collides with Spark function name" in
+      withSparkSession { session =>
+        import session.implicits._
+        val data = Seq(
+          ("setosa", 5),
+          ("setosa", 4),
+          ("versicolor", 6),
+          ("versicolor", 7),
+          ("virginica", 6)
+        ).toDF("flower_type", "length")
+
+        val completeness = Completeness("length")
+        val mean = Mean("length")
+        val maximum = Maximum("length")
+
+        val analysis = Analysis()
+          .addAnalyzer(completeness)
+          .addAnalyzer(mean)
+          .addAnalyzer(maximum)
+
+        val result = AnalysisRunner.run(data, analysis)
+
+        assert(result.metric(completeness).get.value.isSuccess)
+        assert(result.metric(mean).get.value.isSuccess)
+        assert(result.metric(maximum).get.value.get === 7.0)
+      }
   }
 }