[SPARK] Handle special characters correctly in validateDataSkippingType

spark/src/main/scala/org/apache/spark/sql/delta/stats/StatisticsCollection.scala

@@ -47,6 +47,7 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.parser.{AbstractSqlParser, AstBuilder, ParseException, ParserUtils}
 import org.apache.spark.sql.catalyst.parser.SqlBaseParser.MultipartIdentifierListContext
+import org.apache.spark.sql.catalyst.util.quoteIfNeeded
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.functions.lit
 import org.apache.spark.sql.internal.SQLConf
@@ -477,10 +478,18 @@ object StatisticsCollection extends DeltaCommand {
       insideStruct: Boolean = false): Unit = dataType match {
     case s: StructType =>
       s.foreach { field =>
-        validateDataSkippingType(name + "." + field.name, field.dataType, columnPaths,
-          insideStruct = true)
+        // we need to make sure we quote the field if needed otherwise we will not handle
+        // column names with special characters correctly.
+        validateDataSkippingType(name + "." +
+          quoteIfNeeded(field.name), field.dataType, columnPaths, insideStruct = true)
+      }
+    case SkippingEligibleDataType(_) =>
+      if (insideStruct) {
+        // If this is inside the struct we are already quoting the nested field name.
+        columnPaths.append(name)
+      } else {
+        columnPaths.append(quoteIfNeeded(name))
       }
-    case SkippingEligibleDataType(_) => columnPaths.append(name)
     case _ if insideStruct =>
       logWarning(s"Data skipping is not supported for column $name of type $dataType")
       columnPaths.append(name)
@@ -517,7 +526,15 @@ object StatisticsCollection extends DeltaCommand {
         transformSchema(schema, Some(columnName.head)) {
           case (`prefixPath`, struct @ StructType(_), _) =>
             val columnField = struct(columnName.head)
-            validateDataSkippingType(columnAttribute.name, columnField.dataType, visitedColumns)
+            // We need to figure out if the column is top-level column
+            // or a column inside a struct, we support collecting null count stats
+            // on nested columns part of a struct.
+            val fieldInsideStruct = prefixPath.size > 0
+            validateDataSkippingType(
+              columnAttribute.name,
+              columnField.dataType,
+              visitedColumns,
+              insideStruct = fieldInsideStruct)
             struct
           case (_, other, _) => other
         }

spark/src/test/scala/org/apache/spark/sql/delta/stats/StatsCollectionSuite.scala

@@ -526,32 +526,6 @@ class StatsCollectionSuite
         )
       }
     }
-
-    test(s"Delta statistic column: invalid data type $invalidType in nested column") {
-      withTable(tableName1, tableName2) {
-        val columnName = "c2.c21"
-        val exceptOne = intercept[DeltaIllegalArgumentException] {
-          sql(
-            s"create table $tableName1 (c1 long, c2 STRUCT<c20:INT, c21:$invalidType>) " +
-              s"using delta TBLPROPERTIES('delta.dataSkippingStatsColumns' = 'c2.c21')"
-          )
-        }
-        assert(
-          exceptOne.getErrorClass == "DELTA_COLUMN_DATA_SKIPPING_NOT_SUPPORTED_TYPE" &&
-            exceptOne.getMessageParametersArray.toSeq == Seq(columnName, typename)
-        )
-        sql(s"create table $tableName2 (c1 long, c2 STRUCT<c20:INT, c21:$invalidType>) using delta")
-        val exceptThree = interceptWithUnwrapping[DeltaIllegalArgumentException] {
-          sql(
-            s"ALTER TABLE $tableName2 SET TBLPROPERTIES('delta.dataSkippingStatsColumns'='c2.c21')"
-          )
-        }
-        assert(
-          exceptThree.getErrorClass == "DELTA_COLUMN_DATA_SKIPPING_NOT_SUPPORTED_TYPE" &&
-          exceptThree.getMessageParametersArray.toSeq == Seq(columnName, typename)
-        )
-      }
-    }
   }
 
   test(s"Delta statistic column: mix case column name") {
@@ -858,6 +832,21 @@ class StatsCollectionSuite
     }
   }
 
+  test("handle special nested characters in column name") {
+    withTable("t") {
+      sql(
+        s"create table t (`|` long, c struct<s struct<a int, b INT>, `s.a` int>) using delta " +
+          s"TBLPROPERTIES('delta.columnMapping.mode' = 'name')")
+      // Have this test to make sure there are no collisions with c.`s.a` and c.s.a.
+      sql(
+        s"ALTER TABLE t SET TBLPROPERTIES('delta.dataSkippingStatsColumns' = 'c')")
+      val deltaLog = DeltaLog.forTable(spark, TableIdentifier("t"))
+      val tblProperty = DeltaConfigs.DATA_SKIPPING_STATS_COLUMNS
+        .fromMetaData(deltaLog.update().metadata)
+      assert(tblProperty.get == "c")
+    }
+  }
+
   private def recordsScanned(df: DataFrame): Long = {
     val scan = df.queryExecution.executedPlan.find {
       case FileScanExecNode(_) => true