fix: accept GeoJSON strings for Edm.GeographyPoint in AzureSearchWriter

cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/search/AzureSearch.scala

@@ -20,7 +20,8 @@ import org.apache.spark.ml.util._
 import org.apache.spark.ml.{ComplexParamsReadable, NamespaceInjections, PipelineModel}
 import org.apache.spark.ml.linalg.SQLDataTypes.VectorType
 import org.apache.spark.ml.functions.vector_to_array
-import org.apache.spark.sql.functions.{col, expr, struct, to_json, to_utc_timestamp, date_format, when}
+import org.apache.spark.sql.functions.{col, concat, expr, forall, from_json, lit, raise_error, size,
+  struct, to_json, to_utc_timestamp, date_format, when}
 import org.apache.spark.sql.streaming.DataStreamWriter
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{DataFrame, Dataset, Row}
@@ -249,6 +250,74 @@ object AzureSearchWriter extends IndexParser with IndexJsonGetter with SLogging
     }
   }
 
+  /**
+   * Converts string columns containing GeoJSON to the proper struct shape required for
+   * Azure Search `Edm.GeographyPoint` fields.
+   *
+   * Azure AI Search expects spatial values to be sent as a GeoJSON object
+   * (e.g. `{"type":"Point","coordinates":[lon, lat]}`), not as a JSON-encoded string.
+   * Users frequently have their GeoJSON readily available as a string column, and
+   * passing it as a `StringType` previously caused a `400 Bad Request`
+   * (see [[https://github.com/microsoft/SynapseML/issues/2420]]) because the writer
+   * JSON-escaped the entire string.
+   *
+   * For each '''top-level''' field declared as `Edm.GeographyPoint` in the index, if the
+   * corresponding DataFrame column is a `StringType`, parse it into the canonical
+   * `StructType(type: StringType, coordinates: ArrayType(DoubleType))` so that downstream
+   * `to_json` emits a proper GeoJSON object. Columns that are already structured are
+   * left as-is. GeographyPoint fields nested inside complex types are not auto-converted
+   * (mirrors the existing top-level-only handling in `convertDateTimeToISO8601`).
+   *
+   * Parsing uses Spark's `FAILFAST` mode so malformed GeoJSON surfaces an explicit
+   * exception instead of being silently coerced to `null` and shipped to Azure Search.
+   * `FAILFAST` alone only rejects syntactically invalid JSON, so the parsed value is
+   * additionally validated to be a genuine GeoJSON Point (`type == "Point"` with exactly
+   * two non-null coordinates). Anything else raises an error naming the column and the
+   * offending value rather than indexing a silently-null location. NULL inputs are
+   * preserved as NULL.
+   *
+   * @param df DataFrame with potential GeographyPoint columns
+   * @param indexJson JSON string containing the index schema
+   * @return DataFrame with string GeographyPoint columns converted to GeoJSON structs
+   */
+  private[ml] def convertGeographyPointToStruct(df: DataFrame, indexJson: String): DataFrame = {
+   // Derived from edmTypeToSparkType so the parsed shape can never drift from the type
+   // checkSchemaParity expects for Edm.GeographyPoint
+   val geoStructType = edmTypeToSparkType(GeographyPointEdmType, None)
+   val parseOptions = Map("mode" -> "FAILFAST")
+   val geoFields = parseIndexJson(indexJson).fields
+     .filter(_.`type` == GeographyPointEdmType)
+     .map(_.name)
+   geoFields.foldLeft(df) { (currentDF, fieldName) =>
+     if (currentDF.columns.contains(fieldName)) {
+       currentDF.schema(fieldName).dataType match {
+         case StringType =>
+           val parsed = from_json(col(fieldName), geoStructType, parseOptions)
+           val coordinates = parsed.getField("coordinates")
+           val isValidPoint = parsed.getField("type") === lit("Point") &&
+             coordinates.isNotNull &&
+             size(coordinates) === lit(GeographyPointCoordinateCount) &&
+             forall(coordinates, c => c.isNotNull)
+           val invalidValueError = raise_error(concat(
+             lit(s"AzureSearchWriter: column '$fieldName' is mapped to an " +
+               s"$GeographyPointEdmType field but the value is not a valid GeoJSON Point " +
+               """(expected {"type":"Point","coordinates":[longitude,latitude]}). """ +
+               "Offending value: "),
+             col(fieldName)))
+           currentDF.withColumn(fieldName,
+             when(col(fieldName).isNull || isValidPoint, parsed)
+               .otherwise(invalidValueError.cast(geoStructType))
+           )
+         case _ =>
+           // Already a struct (or otherwise compatible); checkSchemaParity will validate.
+           currentDF
+       }
+     } else {
+       currentDF
+     }
+   }
+  }
+
   private def dfToIndexJson(schema: StructType,
                             indexName: String,
                             keyCol: String,
@@ -367,17 +436,18 @@ object AzureSearchWriter extends IndexParser with IndexJsonGetter with SLogging
       SearchIndex.createIfNoneExists(auth, serviceName, indexJson, apiVersion)
     }
     val dateConvertedDF = convertDateTimeToISO8601(preppedDF, indexJson)
+    val geoConvertedDF = convertGeographyPointToStruct(dateConvertedDF, indexJson)
 
     logInfo("checking schema parity")
-    checkSchemaParity(dateConvertedDF.schema, indexJson, actionCol)
+    checkSchemaParity(geoConvertedDF.schema, indexJson, actionCol)
 
     val df1 = if (filterNulls) {
       val collectionColumns = parseIndexJson(indexJson).fields
         .filter(_.`type`.startsWith("Collection"))
         .map(_.name)
-      collectionColumns.foldLeft(dateConvertedDF) { (ndf, c) => filterOutNulls(ndf, c) }
+      collectionColumns.foldLeft(geoConvertedDF) { (ndf, c) => filterOutNulls(ndf, c) }
     } else {
-      dateConvertedDF
+      geoConvertedDF
     }
 
     // Convert date/timestamp columns to ISO8601 strings for Azure AI Search
@@ -451,6 +521,11 @@ object AzureSearchWriter extends IndexParser with IndexJsonGetter with SLogging
     t.substring("Collection(".length).dropRight(1)
   }
 
+  private[ml] val GeographyPointEdmType = "Edm.GeographyPoint"
+
+  // GeoJSON Points are always [longitude, latitude]
+  private[ml] val GeographyPointCoordinateCount = 2
+
   private[ml] def edmTypeToSparkType(dt: String,  //scalastyle:ignore cyclomatic.complexity
                                      fields: Option[Seq[IndexField]]): DataType = dt match {
     case t if isEdmCollection(t) =>
@@ -462,7 +537,7 @@ object AzureSearchWriter extends IndexParser with IndexJsonGetter with SLogging
     case "Edm.Double" => DoubleType
     case "Edm.Single" => FloatType
     case "Edm.DateTimeOffset" => StringType // We convert date/time to ISO8601 strings
-    case "Edm.GeographyPoint"   =>
+    case GeographyPointEdmType =>
       StructType(Seq(
         StructField("type", StringType),
         StructField("coordinates", ArrayType(DoubleType))

cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/search/split2/SearchWriterSuitePart2.scala

@@ -171,4 +171,197 @@ class SearchWriterSuite extends SearchWriterSuiteUtilities {
 
   }
 
+  test("Handle GeoJSON GeographyPoint fields supplied as strings") {
+
+    val in = generateIndexName()
+    val df = spark.createDataFrame(Seq(
+      ("upload", "0", """{"type":"Point","coordinates":[-122.3493, 47.6205]}"""),
+      ("upload", "1", """{"type":"Point","coordinates":[-122.3351, 47.6080]}""")
+    )).toDF("searchAction", "id", "location")
+
+    val indexJson =
+      s"""
+         |{
+         |  "name": "$in",
+         |  "fields": [
+         |    { "name": "id", "type": "Edm.String", "key": true, "searchable": true, "retrievable": true },
+         |    { "name": "location", "type": "Edm.GeographyPoint", "searchable": false,
+         |     "filterable": true, "retrievable": true, "sortable": true }
+         |  ]
+         |}
+         |""".stripMargin
+
+    AzureSearchWriter.write(df,
+      Map(
+        "subscriptionKey" -> azureSearchKey,
+        "actionCol" -> "searchAction",
+        "serviceName" -> testServiceName,
+        "indexJson" -> indexJson
+      )
+    )
+
+    // With fatalErrors=true (default) any 400 from Azure Search becomes a thrown
+    // RuntimeException, so reaching this `assertSize` proves the documents were
+    // accepted as valid spatial objects -- a count of 2 is only achievable if the
+    // GeoJSON strings were correctly parsed and serialized as GeoJSON objects.
+    retryWithBackoff(assertSize(in, 2))
+
+  }
+
+  test("convertGeographyPointToStruct parses GeoJSON strings into structs") {
+    val df = spark.createDataFrame(Seq(
+      ("0", """{"type":"Point","coordinates":[-122.3493, 47.6205]}"""),
+      ("1", null)
+    )).toDF("id", "location")
+
+    val indexJson =
+      """
+        |{
+        |  "name": "unit-test-geo",
+        |  "fields": [
+        |    { "name": "id", "type": "Edm.String", "key": true },
+        |    { "name": "location", "type": "Edm.GeographyPoint" }
+        |  ]
+        |}
+        |""".stripMargin
+
+    val converted = AzureSearchWriter.convertGeographyPointToStruct(df, indexJson)
+    val expected = StructType(Seq(
+      StructField("type", StringType),
+      StructField("coordinates", ArrayType(DoubleType))
+    ))
+    assert(converted.schema("location").dataType == expected)
+
+    val rows = converted.orderBy("id").collect()
+    val parsed = rows.head.getStruct(rows.head.fieldIndex("location"))
+    assert(parsed.getString(0) == "Point")
+    assert(parsed.getSeq[Double](1) == Seq(-122.3493, 47.6205))
+    assert(rows(1).isNullAt(rows(1).fieldIndex("location")))
+  }
+
+  test("convertGeographyPointToStruct leaves struct columns untouched") {
+    val schema = StructType(Seq(
+      StructField("id", StringType),
+      StructField("location", StructType(Seq(
+        StructField("type", StringType, nullable = false),
+        StructField("coordinates", ArrayType(DoubleType, containsNull = false), nullable = false)
+      )))
+    ))
+    val df = spark.createDataFrame(
+      spark.sparkContext.parallelize(Seq(Row("0", Row("Point", Seq(-122.3493, 47.6205))))),
+      schema
+    )
+
+    val indexJson =
+      """
+        |{
+        |  "name": "unit-test-geo",
+        |  "fields": [
+        |    { "name": "id", "type": "Edm.String", "key": true },
+        |    { "name": "location", "type": "Edm.GeographyPoint" }
+        |  ]
+        |}
+        |""".stripMargin
+
+    val converted = AzureSearchWriter.convertGeographyPointToStruct(df, indexJson)
+    assert(converted.schema("location").dataType == schema("location").dataType)
+  }
+
+  test("convertGeographyPointToStruct fails fast on malformed GeoJSON instead of silently nulling") {
+    val df = spark.createDataFrame(Seq(
+      ("0", "{not valid json")
+    )).toDF("id", "location")
+
+    val indexJson =
+      """
+        |{
+        |  "name": "unit-test-geo",
+        |  "fields": [
+        |    { "name": "id", "type": "Edm.String", "key": true },
+        |    { "name": "location", "type": "Edm.GeographyPoint" }
+        |  ]
+        |}
+        |""".stripMargin
+
+    val converted = AzureSearchWriter.convertGeographyPointToStruct(df, indexJson)
+    // FAILFAST surfaces parse errors when the row is materialized, not at plan time.
+    // The concrete wrapper type varies (SparkException, ExecutionException, or a bare
+    // RuntimeException when Spark folds the LocalRelation on the driver), so assert on
+    // the flattened cause chain instead.
+    val caught = intercept[Exception] {
+      converted.collect()
+    }
+    assert(causeChain(caught).contains("Malformed records are detected"),
+      s"expected a FAILFAST parse failure but got: ${causeChain(caught)}")
+  }
+
+  test("convertGeographyPointToStruct rejects valid JSON that is not a GeoJSON Point") {
+    val indexJson =
+      """
+        |{
+        |  "name": "unit-test-geo",
+        |  "fields": [
+        |    { "name": "id", "type": "Edm.String", "key": true },
+        |    { "name": "location", "type": "Edm.GeographyPoint" }
+        |  ]
+        |}
+        |""".stripMargin
+
+    // Every one of these is syntactically valid JSON (or blank), so Spark's FAILFAST parser
+    // accepts it and yields a partially-null struct. Without explicit shape validation these
+    // would be silently indexed as a null location instead of failing the write.
+    val wrongShapes = Seq(
+      """{"foo":"bar"}""",
+      """{"type":"Point"}""",
+      """{"coordinates":[-122.3493, 47.6205]}""",
+      """{"type":"Polygon","coordinates":[-122.3493, 47.6205]}""",
+      """{"type":"Point","coordinates":[-122.3493]}""",
+      """{"type":"Point","coordinates":[-122.3493, 47.6205, 12.0]}""",
+      """{"type":"Point","coordinates":[-122.3493, null]}""",
+      "",
+      "   "
+    )
+
+    wrongShapes.foreach { badValue =>
+      val df = spark.createDataFrame(Seq(("0", badValue))).toDF("id", "location")
+      val converted = AzureSearchWriter.convertGeographyPointToStruct(df, indexJson)
+      val caught = intercept[Exception] {
+        converted.collect()
+      }
+      val message = causeChain(caught)
+      assert(message.contains("not a valid GeoJSON Point"),
+        s"expected a GeoJSON validation failure for '$badValue' but got: $message")
+      assert(message.contains("location"),
+        s"expected the error to name the offending column for '$badValue'")
+    }
+  }
+
+  test("convertGeographyPointToStruct preserves nulls without raising") {
+    val df = spark.createDataFrame(Seq(
+      ("0", """{"type":"Point","coordinates":[-122.3493, 47.6205]}"""),
+      ("1", null),
+      ("2", null)
+    )).toDF("id", "location")
+
+    val indexJson =
+      """
+        |{
+        |  "name": "unit-test-geo",
+        |  "fields": [
+        |    { "name": "id", "type": "Edm.String", "key": true },
+        |    { "name": "location", "type": "Edm.GeographyPoint" }
+        |  ]
+        |}
+        |""".stripMargin
+
+    val rows = AzureSearchWriter.convertGeographyPointToStruct(df, indexJson).orderBy("id").collect()
+    assert(rows.length == 3)
+    assert(!rows.head.isNullAt(rows.head.fieldIndex("location")))
+    assert(rows(1).isNullAt(rows(1).fieldIndex("location")))
+    assert(rows(2).isNullAt(rows(2).fieldIndex("location")))
+  }
+
+  private def causeChain(t: Throwable): String =
+    Iterator.iterate(t)(_.getCause).takeWhile(_ != null).map(_.toString).mkString(" | ")
+
 }