diff --git a/cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/search/AzureSearch.scala b/cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/search/AzureSearch.scala index 7535e17006..15dfaf98d9 100644 --- a/cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/search/AzureSearch.scala +++ b/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)) diff --git a/cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/search/split2/SearchWriterSuitePart2.scala b/cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/search/split2/SearchWriterSuitePart2.scala index e24f77ae58..dec4874cb8 100644 --- a/cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/search/split2/SearchWriterSuitePart2.scala +++ b/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(" | ") + }