Skip to content

Commit 86a0655

Browse files
ranadeepsinghCopilot
ranadeepsingh
and
Copilot
authored
fix: route AnalyzeText document errors to errorCol (#2569)
* fix: route AnalyzeText document errors to errorCol Move Azure AI Language document-level errors returned inside HTTP 200 AnalyzeText responses from the response payload into the configured error column after auto-batch flattening. Preserve transport error precedence and add a no-network regression test for mixed document success/error responses. AB#4638662 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * ci: pin PR validation sbt launcher Use the sbt launcher version from project/build.properties instead of installing the latest apt sbt package. This keeps the JDK 11 PR validation job on the repository's sbt 1.10.11 launcher and avoids sbt 2.x rejecting JDK 11 before scalastyle can run. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * ci: use pinned sbt wrapper in PR validation Invoke the downloaded sbt launcher explicitly so the GitHub runner does not resolve its preinstalled sbt 2.x binary under JDK 11. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * ci: prefer pinned sbt on PATH Keep PR validation commands as plain sbt while placing the repository-version launcher first on PATH for subsequent workflow steps. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * test: avoid ordering assumption in AnalyzeText error test Partition collected rows by error nullability instead of relying on collect order, addressing PR review feedback about Spark DataFrames being unordered. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent f3002df commit 86a0655

3 files changed

Lines changed: 148 additions & 15 deletions

File tree

‎.github/workflows/pr-validation.yml‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,10 +41,18 @@ jobs:
4141

4242
- name: Install sbt
4343
run: |
44-
echo "deb https://repo.scala-sbt.org/scalasbt/debian all main" | sudo tee /etc/apt/sources.list.d/sbt.list
45-
curl -sL "https://keyserver.ubuntu.com/pks/lookup?op=get&search=0x2EE0EA64E40A89B84B2DF73499E82A75642AC823" | sudo apt-key add
46-
sudo apt-get update -q
47-
sudo apt-get install -yq sbt
44+
SBT_VERSION="$(sed -n 's/^sbt.version *= *//p' project/build.properties | tr -d ' ')"
45+
mkdir -p "$HOME/.local/bin"
46+
curl -L -o "$HOME/.local/bin/sbt-launch.jar" \
47+
"https://repo1.maven.org/maven2/org/scala-sbt/sbt-launch/${SBT_VERSION}/sbt-launch-${SBT_VERSION}.jar"
48+
cat > "$HOME/.local/bin/sbt" <<EOF
49+
#!/usr/bin/env bash
50+
exec java -Dsbt.version="${SBT_VERSION}" -jar "$HOME/.local/bin/sbt-launch.jar" "\$@"
51+
EOF
52+
chmod +x "$HOME/.local/bin/sbt"
53+
echo "$HOME/.local/bin" >> "$GITHUB_PATH"
54+
export PATH="$HOME/.local/bin:$PATH"
55+
sbt sbtVersion
4856
4957
- name: Scalastyle check
5058
run: sbt scalastyle test:scalastyle

‎cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/language/AnalyzeText.scala‎

Lines changed: 47 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -4,16 +4,18 @@
44
package com.microsoft.azure.synapse.ml.services.language
55

66
import com.microsoft.azure.synapse.ml.logging.{ FeatureNames, SynapseMLLogging }
7+
import com.microsoft.azure.synapse.ml.io.http.ErrorUtils
78
import com.microsoft.azure.synapse.ml.param.ServiceParam
89
import com.microsoft.azure.synapse.ml.services._
910
import com.microsoft.azure.synapse.ml.services.text.{ TADocument, TextAnalyticsAutoBatch }
10-
import com.microsoft.azure.synapse.ml.stages.{ FixedMiniBatchTransformer, FlattenBatch, HasBatchSize, UDFTransformer }
11+
import com.microsoft.azure.synapse.ml.stages.{ FixedMiniBatchTransformer, FlattenBatch, HasBatchSize, Lambda,
12+
UDFTransformer }
1113
import org.apache.http.entity.{ AbstractHttpEntity, StringEntity }
1214
import org.apache.spark.injections.UDFUtils
1315
import org.apache.spark.ml.{ ComplexParamsReadable, NamespaceInjections, PipelineModel }
1416
import org.apache.spark.ml.param.{ Param, ParamValidators }
1517
import org.apache.spark.ml.util.Identifiable
16-
import org.apache.spark.sql.Row
18+
import org.apache.spark.sql.{ Column, Row, functions => F }
1719
import org.apache.spark.sql.expressions.UserDefinedFunction
1820
import org.apache.spark.sql.types.{ ArrayType, DataType, StructType }
1921
import spray.json._
@@ -258,19 +260,47 @@ class AnalyzeText(override val uid: String) extends CognitiveServicesBase(uid)
258260
}
259261
}
260262

261-
protected def postprocessResponseUdf: UserDefinedFunction = {
263+
private def postprocessedOutputType: StructType = {
262264
val responseType = responseDataType.asInstanceOf[StructType]
263265
val results = responseType("results").dataType.asInstanceOf[StructType]
264-
val outputType = ArrayType(
265-
new StructType()
266-
.add("statistics", results("statistics").dataType)
267-
.add("documents", results("documents").dataType.asInstanceOf[ArrayType].elementType)
268-
.add("errors", results("errors").dataType.asInstanceOf[ArrayType].elementType)
269-
.add("modelVersion", results("modelVersion").dataType)
270-
)
266+
new StructType()
267+
.add("statistics", results("statistics").dataType)
268+
.add("documents", results("documents").dataType.asInstanceOf[ArrayType].elementType)
269+
.add("errors", results("errors").dataType.asInstanceOf[ArrayType].elementType)
270+
.add("modelVersion", results("modelVersion").dataType)
271+
}
272+
273+
protected def postprocessResponseUdf: UserDefinedFunction = {
274+
val outputType = ArrayType(postprocessedOutputType)
271275
UDFUtils.oldUdf(postprocessResponse _, outputType)
272276
}
273277

278+
private def responseErrorToErrorCol(error: Column): Column = {
279+
F.when(error.isNotNull, F.struct(
280+
F.to_json(error).as("response"),
281+
F.lit(null).cast(ErrorUtils.ErrorSchema("status").dataType).as("status") // scalastyle:ignore null
282+
))
283+
}
284+
285+
private def outputWithoutResponseError(output: Column): Column = {
286+
val outputType = postprocessedOutputType
287+
F.when(output.isNotNull, F.struct(
288+
output.getField("statistics").as("statistics"),
289+
output.getField("documents").as("documents"),
290+
F.lit(null).cast(outputType("errors").dataType).as("errors"), // scalastyle:ignore null
291+
output.getField("modelVersion").as("modelVersion")
292+
)).otherwise(F.lit(null).cast(outputType)) // scalastyle:ignore null
293+
}
294+
295+
private def moveResponseErrorsToErrorCol(
296+
dataset: org.apache.spark.sql.Dataset[_]): org.apache.spark.sql.DataFrame = {
297+
val df = dataset.toDF
298+
val output = F.col(getOutputCol)
299+
val responseError = output.getField("errors")
300+
df.withColumn(getErrorCol, F.coalesce(F.col(getErrorCol), responseErrorToErrorCol(responseError)))
301+
.withColumn(getOutputCol, F.when(responseError.isNotNull, outputWithoutResponseError(output)).otherwise(output))
302+
}
303+
274304
override protected def getInternalTransformer(schema: StructType): PipelineModel = {
275305

276306
val batcher = if (shouldAutoBatch(schema)) {
@@ -293,8 +323,14 @@ class AnalyzeText(override val uid: String) extends CognitiveServicesBase(uid)
293323
None
294324
}
295325

326+
val moveResponseErrors = if (shouldAutoBatch(schema)) {
327+
Some(Lambda(moveResponseErrorsToErrorCol _).setTransformSchema((schema: StructType) => schema))
328+
} else {
329+
None
330+
}
331+
296332
NamespaceInjections.pipelineModel(
297-
Array(batcher, Some(pipe), Some(postprocess), flatten).flatten
333+
Array(batcher, Some(pipe), Some(postprocess), flatten, moveResponseErrors).flatten
298334
)
299335
}
300336

‎cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/language/AnalyzeTextSuite.scala‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,102 @@
33

44
package com.microsoft.azure.synapse.ml.services.language
55

6+
import com.microsoft.azure.synapse.ml.core.test.base.TestBase
67
import com.microsoft.azure.synapse.ml.services.text.{SentimentAssessment, TextEndpoint}
78
import com.microsoft.azure.synapse.ml.core.test.fuzzing.{TestObject, TransformerFuzzing}
9+
import com.microsoft.azure.synapse.ml.io.http.{HTTPRequestData, HTTPResponseData, HTTPSchema}
10+
import org.apache.http.impl.client.CloseableHttpClient
811
import org.apache.spark.ml.util.MLReadable
912
import org.apache.spark.sql.{DataFrame, Row}
1013
import org.apache.spark.sql.functions.{col, flatten, map}
1114
import org.scalactic.{Equality, TolerantNumerics}
1215

16+
object AnalyzeTextErrorRoutingTestData extends Serializable {
17+
val ResponseWithDocumentError: String =
18+
"""{
19+
| "kind": "PiiEntityRecognition",
20+
| "results": {
21+
| "documents": [
22+
| {
23+
| "id": "0",
24+
| "redactedText": "My SSN is ***********",
25+
| "entities": [],
26+
| "warnings": [],
27+
| "statistics": null
28+
| }
29+
| ],
30+
| "errors": [
31+
| {
32+
| "id": "1",
33+
| "error": {
34+
| "code": "InvalidArgument",
35+
| "message": "Document exceeds the service character limit.",
36+
| "target": "documents.1",
37+
| "details": null,
38+
| "innererror": {
39+
| "code": "InvalidDocument",
40+
| "innerError": "DocumentTooLong"
41+
| }
42+
| }
43+
| }
44+
| ],
45+
| "modelVersion": "test",
46+
| "statistics": {
47+
| "documentsCount": 2,
48+
| "validDocumentsCount": 1,
49+
| "erroneousDocumentsCount": 1,
50+
| "transactionsCount": 2
51+
| }
52+
| }
53+
|}""".stripMargin
54+
55+
def okResponseHandler(
56+
client: CloseableHttpClient,
57+
request: HTTPRequestData): HTTPResponseData = {
58+
HTTPSchema.stringToResponse(ResponseWithDocumentError, 200, "OK")
59+
}
60+
}
61+
62+
class AnalyzeTextErrorRoutingSuite extends TestBase {
63+
import spark.implicits._
64+
65+
test("AnalyzeText moves document-level 200 response errors to errorCol") {
66+
val model = new AnalyzeText()
67+
.setSubscriptionKey("unused")
68+
.setLocation("eastus")
69+
.setTextCol("text")
70+
.setLanguage("en")
71+
.setKind("PiiEntityRecognition")
72+
.setOutputCol("response")
73+
.setErrorCol("error")
74+
.setHandler(AnalyzeTextErrorRoutingTestData.okResponseHandler _)
75+
76+
val rows = model.transform(Seq("valid text", "too long").toDF("text").coalesce(1))
77+
.select("response", "error")
78+
.collect()
79+
80+
assert(rows.length == 2)
81+
val (failedRows, successRows) = rows.partition(row => row.getAs[Row]("error") != null)
82+
assert(successRows.length == 1)
83+
assert(failedRows.length == 1)
84+
85+
val successResponse = successRows.head.getAs[Row]("response")
86+
assert(successResponse.getAs[Row]("documents") != null)
87+
assert(successResponse.getAs[Row]("errors") == null)
88+
89+
val failedResponse = failedRows.head.getAs[Row]("response")
90+
assert(failedResponse.getAs[Row]("documents") == null)
91+
assert(failedResponse.getAs[Row]("errors") == null)
92+
93+
val error = failedRows.head.getAs[Row]("error")
94+
assert(error != null)
95+
val errorResponse = error.getAs[String]("response")
96+
assert(errorResponse.contains("InvalidArgument"))
97+
assert(errorResponse.contains("Document exceeds the service character limit."))
98+
assert(error.getAs[Row]("status") == null)
99+
}
100+
}
101+
13102
class EntityLinkingSuite extends TransformerFuzzing[AnalyzeText] with TextEndpoint {
14103
override val compareDataInSerializationTest: Boolean = false
15104

0 commit comments

Comments
 (0)