-
Notifications
You must be signed in to change notification settings - Fork 861
fix: accept GeoJSON strings for Edm.GeographyPoint in AzureSearchWriter #2556
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -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, expr, from_json, 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,50 @@ 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 issue #2420) because the writer JSON-escaped the entire string. | ||||||||||||
| * | ||||||||||||
| * For each field declared as `Edm.GeographyPoint` in the index, if the corresponding | ||||||||||||
| * DataFrame column is a `StringType`, parse it into the canonical | ||||||||||||
| * `struct<type:string, coordinates:array<double>>` so that downstream `to_json` | ||||||||||||
| * emits a proper GeoJSON object. Columns that are already structured are left as-is. | ||||||||||||
| * | ||||||||||||
| * @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 def convertGeographyPointToStruct(df: DataFrame, indexJson: String): DataFrame = { | ||||||||||||
| val geoStructType = StructType(Seq( | ||||||||||||
| StructField("type", StringType), | ||||||||||||
| StructField("coordinates", ArrayType(DoubleType)) | ||||||||||||
| )) | ||||||||||||
| val geoFields = parseIndexJson(indexJson).fields | ||||||||||||
| .filter(_.`type` == "Edm.GeographyPoint") | ||||||||||||
| .map(_.name) | ||||||||||||
| geoFields.foldLeft(df) { (currentDF, fieldName) => | ||||||||||||
| if (currentDF.columns.contains(fieldName)) { | ||||||||||||
| currentDF.schema(fieldName).dataType match { | ||||||||||||
| case StringType => | ||||||||||||
| currentDF.withColumn(fieldName, | ||||||||||||
| when(col(fieldName).isNotNull, from_json(col(fieldName), geoStructType)) | ||||||||||||
|
||||||||||||
| when(col(fieldName).isNotNull, from_json(col(fieldName), geoStructType)) | |
| when( | |
| col(fieldName).isNotNull, | |
| from_json(col(fieldName), geoStructType, Map("mode" -> "FAILFAST")) | |
| ) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done in 95a7ae1 -- switched to from_json(..., Map("mode" -> "FAILFAST")) and added a unit test (convertGeographyPointToStruct fails fast on malformed GeoJSON instead of silently nulling) that asserts a SparkException on materialization.
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -171,4 +171,37 @@ 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 | ||
| ) | ||
| ) | ||
|
|
||
| retryWithBackoff(assertSize(in, 2)) | ||
|
|
||
|
Comment on lines
+207
to
+208
|
||
| } | ||
|
|
||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The Scaladoc says this runs "for each field declared as
Edm.GeographyPointin the index", but the implementation only inspectsparseIndexJson(indexJson).fields(top-level fields) and won’t convert GeographyPoint values nested inside complex types. Either clarify the doc to state it’s top-level only, or extend the traversal to nestedfieldsso behavior matches the comment.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done in 95a7ae1 -- updated the Scaladoc to state '''top-level''' explicitly and noted that the scope mirrors
convertDateTimeToISO8601(which has the same top-level-only limitation). Extending nested-field traversal felt out-of-scope for this bug fix; happy to do it as a follow-up if preferred.