Integrate feedback (1)

Author: andrei-ionescu

src/main/scala/com/microsoft/hyperspace/actions/CreateAction.scala

@@ -67,9 +67,12 @@ class CreateAction(
   private def isValidIndexSchema(config: IndexConfig, schema: StructType): Boolean = {
     // First we flatten the schema. Instead of having struct of leaves
     // the flatten method will return a list of field names.
-    // The second step is escaping the field names as Parquet does not works well
-    // with field names that contains the `.` (dot).
-    // Given `struct(nested, struct(nst, struct(field1)))`, the fields variable
+    // The second step is escaping the field names as there are some problems when
+    // using field names with dots. One is `partitionBy` does not works well
+    // with field names that contains the `.` (dot). See more on this Apache Spark
+    // ticket: https://issues.apache.org/jira/browse/SPARK-18084. Other is doing
+    // encountering  `AnalysisException: Cannot resolve column name...` exceptions.
+    // So, given `struct(nested, struct(nst, struct(field1)))`, the fields variable
     // will contain `Seq("nested__nst__field1")`.
     val fields = SchemaUtils.escapeFieldNames(SchemaUtils.flatten(schema))
     // Resolve index config columns from available column names present in the schema.

src/main/scala/com/microsoft/hyperspace/util/SchemaUtils.scala

@@ -16,10 +16,11 @@
 
 package com.microsoft.hyperspace.util
 
-import org.apache.spark.sql.types.{ArrayType, StructField, StructType}
+import org.apache.spark.sql.types.{ArrayType, MapType, StructField, StructType}
 
 object SchemaUtils {
 
+  val BACKTICK_MARKER_REGEX = "^`(.*)`$"
   val NESTED_FIELD_NEEDLE_REGEX = "\\."
   val NESTED_FIELD_REPLACEMENT = "__"
 
@@ -31,6 +32,7 @@ object SchemaUtils {
    * root
    *   |-- id: integer (nullable = true)
    *   |-- name: string (nullable = true)
+   *   |-- nested.nst.field1: string (nullable = true)
    *   |-- nested: struct (nullable = true)
    *   |    |-- field1: string (nullable = true)
    *   |    |-- nst: struct (nullable = true)
@@ -44,12 +46,16 @@ object SchemaUtils {
    * Seq(
    *   "id",
    *   "name",
+   *   "`nested.nst.field1`",
    *   "nested.field1",
    *   "nested.nst.field1",
    *   "nested.nst.field2"
    * )
    * }}}
    *
+   * As observed if there is a field that is not nested but contains `.` (dot)
+   * that will be enclosed in backticks.
+   *
    * @param structFields The struct fields we want to flatten. This can be a [[StructType]] too.
    * @param prefix Option where you can specify a prefix otherwise None.
    * @return The list for leaf fields flattened.
@@ -60,8 +66,18 @@ object SchemaUtils {
         flatten(fields, Some(prefix.map(o => s"$o.$name").getOrElse(name)))
       case StructField(name, ArrayType(StructType(fields), _), _, _) =>
         flatten(fields, Some(prefix.map(o => s"$o.$name").getOrElse(name)))
+      case StructField(name, MapType(StructType(keys), StructType(values), _), _, _) =>
+        flatten(keys, Some(prefix.map(o => s"$o.$name").getOrElse(name))) ++
+          flatten(values, Some(prefix.map(o => s"$o.$name").getOrElse(name)))
       case other =>
-        Seq(prefix.map(o => s"$o.${other.name}").getOrElse(other.name))
+        if (other.name.contains(".")) {
+          // first clean it, then prefix it, then again enclose it with backticks
+          val cleanName = other.name.replaceAll(BACKTICK_MARKER_REGEX, "$1")
+          val prefixed = prefix.map(o => s"$o.$cleanName").getOrElse(cleanName)
+          Seq(s"`$prefixed`")
+        } else {
+          Seq(prefix.map(o => s"$o.${other.name}").getOrElse(other.name))
+        }
     }
   }
 
@@ -76,18 +92,27 @@ object SchemaUtils {
   }
 
   /**
-   * The method escapes the flattened field names.
+   * The method escapes the flattened field name if not enclosed by backticks.
+   *
+   * The field names can escaped by enclosing them by backticks to specify
+   * that the `.` (dot) does not mean a nested field.
    *
    * Given {{{nested.nst.field1}}} it will be escaped to {{{nested__nst__field1}}}.
+   * Given {{{`nested.nst.field1`}}} it will remain as it is.
    *
    * The values used for search and replaced are defined under
-   * [[NESTED_FIELD_NEEDLE_REGEX]] and [[NESTED_FIELD_REPLACEMENT]].
+   * [[NESTED_FIELD_NEEDLE_REGEX]], [[NESTED_FIELD_REPLACEMENT]]
+   * and [[backticksEnclosed]] method.
    *
    * @param field The flattened field name to be escaped.
    * @return The escaped field name.
    */
   def escapeFieldName(field: String): String = {
-    field.replaceAll(NESTED_FIELD_NEEDLE_REGEX, NESTED_FIELD_REPLACEMENT)
+    if (backticksEnclosed(field)) {
+      field
+    } else {
+      field.replaceAll(NESTED_FIELD_NEEDLE_REGEX, NESTED_FIELD_REPLACEMENT)
+    }
   }
 
   /**
@@ -101,10 +126,16 @@ object SchemaUtils {
   }
 
   /**
-   * The method unescapes the field name (returns the original field name).
+   * The method unescapes the field name (returns the original field name)
+   * if the field name is not enclosed by backticks.
+   *
+   * The field names can escaped by enclosing them by backticks to specify
+   * that the `.` (dot) does not mean a nested field.
+   *
    * The method is the inverse operation of [[escapeFieldName]] method.
    *
    * Given {{{nested__nst__field1}}} it will be escaped to {{{nested.nst.field1}}}.
+   * Given {{{`nested__nst__field1`}}} it will remain as it is.
    *
    * The values used for search and replaced are defined under
    * [[NESTED_FIELD_NEEDLE_REGEX]] and [[NESTED_FIELD_REPLACEMENT]].
@@ -113,7 +144,11 @@ object SchemaUtils {
    * @return The original (unescaped) field name.
    */
   def unescapeFieldName(field: String): String = {
-    field.replaceAll(NESTED_FIELD_REPLACEMENT, NESTED_FIELD_NEEDLE_REGEX)
+    if (backticksEnclosed(field)) {
+      field
+    } else {
+      field.replaceAll(NESTED_FIELD_REPLACEMENT, NESTED_FIELD_NEEDLE_REGEX)
+    }
   }
 
   /**
@@ -131,14 +166,32 @@ object SchemaUtils {
    * The method checks if a field name represents a nested field.
    *
    * The check is implemented by checking if the field name string contains
-   * the separator defined for nested field expressions.
+   * the separator defined for nested field expressions and is not enclosed
+   * by backticks.
    *
-   * See [[NESTED_FIELD_NEEDLE_REGEX]].
+   * The field names can escaped by enclosing them by backticks to specify
+   * that the `.` (dot) does not mean a nested field.
    *
-   * @param field The field nme
+   * See [[NESTED_FIELD_NEEDLE_REGEX]] and [[backticksEnclosed]] method.
+   *
+   * @param field The field name
    * @return True if the field name represents a nested field otherwise false.
    */
   def isNestedField(field: String): Boolean = {
+    !backticksEnclosed(field) &&
     NESTED_FIELD_NEEDLE_REGEX.r.findFirstIn(field).isDefined
   }
+
+  /**
+   * The method detects if the field is enclosed by backticks.
+   *
+   * The field names can escaped by enclosing them by backticks to specify
+   * that the `.` (dot) does not mean a nested field.
+   *
+   * @param field The field name
+   * @return True is the field name is enclosed by backticks.
+   */
+  def backticksEnclosed(field: String): Boolean = {
+    BACKTICK_MARKER_REGEX.r.findFirstIn(field).isDefined
+  }
 }

src/test/scala/com/microsoft/hyperspace/util/SchemaUtilsTests.scala

@@ -106,6 +106,21 @@ class SchemaUtilsTest extends SparkFunSuite with SparkInvolvedSuite {
     assert(flattened2(4) == "nested.n.n.nf1_b")
     assert(flattened2(5) == "nested.n.n.n.f1")
     assert(flattened2(6) == "nested.n.n.n.f2")
+
+    val df3 = Seq(
+      (1, "f1", "g1", NestedType("n1", 1L)),
+      (2, "f2", "g2", NestedType("n2", 1L)),
+      (3, "f3", "g3", NestedType("n3", 1L))
+    ).toDF("id", "nested.f1", "`other.nested`", "nested")
+
+    val flattened3 = SchemaUtils.flatten(df3.schema)
+
+    assert(flattened3.length == 5)
+    assert(flattened3(0) == "id")
+    assert(flattened3(1) == "`nested.f1`")
+    assert(flattened3(2) == "`other.nested`")
+    assert(flattened3(3) == "nested.f1")
+    assert(flattened3(4) == "nested.f2")
   }
 
   test("flatten - array") {
@@ -176,25 +191,50 @@ class SchemaUtilsTest extends SparkFunSuite with SparkInvolvedSuite {
     assert(SchemaUtils.escapeFieldName("a.b") == "a__b")
     assert(SchemaUtils.escapeFieldName("a.b.c.d") == "a__b__c__d")
     assert(SchemaUtils.escapeFieldName("a_b_c_d") == "a_b_c_d")
+    assert(SchemaUtils.escapeFieldName("`a.b.c.d`") == "`a.b.c.d`")
   }
 
   test("escapeFieldNames") {
     assert(SchemaUtils.escapeFieldNames(
-      Seq("a.b.c.d", "a.b", "A_B")) == Seq("a__b__c__d", "a__b", "A_B"))
+      Seq("a.b.c.d", "a.b", "A_B", "`a.b`")) == Seq("a__b__c__d", "a__b", "A_B", "`a.b`"))
     assert(SchemaUtils.escapeFieldNames(Seq.empty[String]).isEmpty)
   }
 
   test("unescapeFieldName") {
     assert(SchemaUtils.unescapeFieldName("a__b") == "a.b")
     assert(SchemaUtils.unescapeFieldName("a__b__c__d") == "a.b.c.d")
     assert(SchemaUtils.unescapeFieldName("a_b_c_d") == "a_b_c_d")
+    assert(SchemaUtils.unescapeFieldName("`a__b__c__d`") == "`a__b__c__d`")
   }
 
   test("unescapeFieldNames") {
     assert(SchemaUtils.unescapeFieldNames(
-      Seq("a__b__c__d", "a__b", "A_B")) == Seq("a.b.c.d", "a.b", "A_B"))
+      Seq("a__b__c__d", "a__b", "A_B", "`a__b`")) == Seq("a.b.c.d", "a.b", "A_B", "`a__b`"))
     assert(SchemaUtils.escapeFieldNames(Seq.empty[String]).isEmpty)
   }
+
+  test("isNestedField") {
+    assert(SchemaUtils.isNestedField("a.b.c"))
+    assert(SchemaUtils.isNestedField("`a.b.c"))
+    assert(SchemaUtils.isNestedField("a.b.c`"))
+    assert(!SchemaUtils.isNestedField("`a.b.c`"))
+    assert(!SchemaUtils.isNestedField("`abc`"))
+    assert(!SchemaUtils.isNestedField("`abc"))
+    assert(!SchemaUtils.isNestedField("abc`"))
+  }
+
+  test("hasNestedField") {
+    assert(SchemaUtils.hasNestedFields(Seq("a.b.c", "abc")))
+    assert(!SchemaUtils.hasNestedFields(Seq("`a.b.c`", "abc")))
+    assert(!SchemaUtils.hasNestedFields(Seq.empty[String]))
+  }
+
+  test("backticksEnclosed") {
+    assert(!SchemaUtils.backticksEnclosed("a.b.c"))
+    assert(!SchemaUtils.backticksEnclosed("`a.b.c"))
+    assert(!SchemaUtils.backticksEnclosed("a.b.c`"))
+    assert(SchemaUtils.backticksEnclosed("`a.b.c`"))
+  }
 }
 
 case class NestedType4(nf1_b: String, n: NestedType)