From b55f9176db79e5102fb7c7d325432e4688b3a6c1 Mon Sep 17 00:00:00 2001 From: suyashj1231 <149845903+suyashj1231@users.noreply.github.com> Date: Fri, 18 Sep 2026 18:14:16 -0700 Subject: [PATCH] fix(workflow-operator): honor the File Scan Encoding field MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FileScanSourceOpDesc re-declares the inherited fileEncoding as its own `encoding` property so the field can carry a hide annotation, and suppresses the inherited one with @JsonIgnoreProperties(value = Array("limit", "offset", "fileEncoding")) That is the same pattern TextSourceOpDesc uses for fileScanLimit and fileScanOffset, and those two work. Encoding did not, for two reasons: `encoding` was `private val`, so the executor could not read it, and the executor instead read `desc.fileEncoding`, which the annotation strips during serialization and which therefore always came back as its UTF_8 default. So the Encoding field on a File Scan changed nothing. A UTF-16 file was decoded as UTF-8 whatever the user picked. Make `encoding` a `var`, matching ScanSourceOpDesc.fileEncoding and FileScanOpDesc.fileEncoding, and read it in the executor. FileScanOpDesc declares its own public fileEncoding and was never affected. The existing US_ASCII case set the inherited field, so it exercised the encoding path without being able to detect this: ASCII and UTF-8 agree on ASCII bytes, so it passed either way. It now sets `encoding`, and two cases are added — one that the charset survives the descriptor round trip getPhysicalOp uses to reach the executor, and one that a real UTF-16 file decodes to its text. The latter fails on the old wiring. Closes #8596 Claude-Session: https://claude.ai/code/session_01EeaEYRdhRYWL7ya7w8LJux --- .../scan/file/FileScanSourceOpDesc.scala | 6 ++- .../scan/file/FileScanSourceOpExec.scala | 2 +- .../scan/file/FileScanSourceOpDescSpec.scala | 53 ++++++++++++++++++- 3 files changed, 57 insertions(+), 4 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala index 82997632d14..c49a616c339 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala @@ -45,7 +45,11 @@ class FileScanSourceOpDesc extends ScanSourceOpDesc with TextSourceOpDesc { new JsonSchemaString(path = HideAnnotation.hideExpectedValue, value = "binary") ) ) - private val encoding: FileDecodingMethod = FileDecodingMethod.UTF_8 + // Re-declared here rather than inherited so the field can carry the hide + // annotation above; `fileEncoding` from ScanSourceOpDesc is suppressed by the + // @JsonIgnoreProperties on this class, so this is the only charset that + // survives into the executor. + var encoding: FileDecodingMethod = FileDecodingMethod.UTF_8 @JsonProperty(defaultValue = "false") @JsonSchemaTitle("Extract") diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExec.scala index d47cf3681c2..254aa086fd5 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpExec.scala @@ -36,7 +36,7 @@ class FileScanSourceOpExec private[scan] ( FileScanUtils.createTuplesFromFile( fileName = desc.fileName.get, attributeType = desc.attributeType, - fileEncoding = desc.fileEncoding, + fileEncoding = desc.encoding, extract = desc.extract, outputFileName = desc.outputFileName, fileScanOffset = desc.fileScanOffset, diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala index fafb696f131..6ac320e99cf 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala @@ -30,6 +30,9 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.BeforeAndAfter import org.scalatest.flatspec.AnyFlatSpec +import java.nio.charset.StandardCharsets +import java.nio.file.Files + class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { var fileScanSourceOpDesc: FileScanSourceOpDesc = _ @@ -37,7 +40,7 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { before { fileScanSourceOpDesc = new FileScanSourceOpDesc() fileScanSourceOpDesc.setResolvedFileName(FileResolver.resolve(TestOperators.TestTextFilePath)) - fileScanSourceOpDesc.fileEncoding = FileDecodingMethod.UTF_8 + fileScanSourceOpDesc.encoding = FileDecodingMethod.UTF_8 } it should "infer schema with single column representing each line of text in normal text scan mode" in { @@ -188,7 +191,7 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { fileScanSourceOpDesc.setResolvedFileName( FileResolver.resolve(TestOperators.TestCRLFTextFilePath) ) - fileScanSourceOpDesc.fileEncoding = FileDecodingMethod.ASCII + fileScanSourceOpDesc.encoding = FileDecodingMethod.ASCII fileScanSourceOpDesc.attributeType = FileAttributeType.STRING fileScanSourceOpDesc.fileScanLimit = Option(5) val FileScanSourceOpExec = @@ -209,6 +212,52 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { FileScanSourceOpExec.close() } + it should "carry the Encoding field through serialization into the executor" in { + fileScanSourceOpDesc.encoding = FileDecodingMethod.UTF_16 + + // getPhysicalOp hands the executor objectMapper.writeValueAsString(this), and + // FileScanSourceOpExec reads the descriptor back out of that string, so the + // charset only reaches the executor if it survives the round trip. + val roundTripped = objectMapper.readValue( + objectMapper.writeValueAsString(fileScanSourceOpDesc), + classOf[FileScanSourceOpDesc] + ) + + assert(roundTripped.encoding == FileDecodingMethod.UTF_16) + } + + it should "decode a UTF-16 file with the charset the Encoding field names" in { + val utf16File = Files.createTempFile("file-scan-utf16", ".txt") + try { + Files.write(utf16File, "line1\nline2\nline3".getBytes(StandardCharsets.UTF_16)) + + fileScanSourceOpDesc.setResolvedFileName(FileResolver.resolve(utf16File.toString)) + fileScanSourceOpDesc.encoding = FileDecodingMethod.UTF_16 + fileScanSourceOpDesc.attributeType = FileAttributeType.STRING + + val fileScanSourceOpExec = + new FileScanSourceOpExec(objectMapper.writeValueAsString(fileScanSourceOpDesc)) + fileScanSourceOpExec.open() + val processedTuple: Iterator[Tuple] = fileScanSourceOpExec + .produceTuple() + .map(tupleLike => + tupleLike + .asInstanceOf[SchemaEnforceable] + .enforceSchema(fileScanSourceOpDesc.sourceSchema()) + ) + + // Decoded as UTF-8 these bytes come back as the byte-order mark followed by + // NUL-interleaved characters, so this is the assertion the old wiring failed. + assert(processedTuple.next().getField("line").equals("line1")) + assert(processedTuple.next().getField("line").equals("line2")) + assert(processedTuple.next().getField("line").equals("line3")) + assertThrows[java.util.NoSuchElementException](processedTuple.next().getField("line")) + fileScanSourceOpExec.close() + } finally { + Files.deleteIfExists(utf16File) + } + } + "FileScanSourceOpDesc.getPhysicalOp" should "wire the FileScanSourceOpExec class as a source op and propagate its schema" in { val physical =