This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8605-8dbf049bd814fde00c0ce1407a1f31c7de03c645 in repository https://gitbox.apache.org/repos/asf/texera.git
commit be8fc4f8fd76c3330b43defa131b2defbfa5e055 Author: Suyash Jain <[email protected]> AuthorDate: Thu Sep 24 03:29:05 2026 +0000 feat(workflow-operator): honor the File Scan Encoding field (#8605) ### What changes were proposed in this PR? File Scan offers an **Encoding** field, and picking anything other than UTF-8 changed nothing. A UTF-16 file came back decoded as UTF-8 rather than as its text. `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 that compounded. First, `encoding` was `private val`, so the executor could not read it even if it wanted to. Second, `FileScanSourceOpExec` instead read `desc.fileEncoding`, the inherited field that the annotation above strips during serialization, so it always came back as its `UTF_8` default. The fix makes `encoding` a `var`, matching `ScanSourceOpDesc.fileEncoding` and `FileScanOpDesc.fileEncoding`, and reads it in the executor: ``` before: fileEncoding = desc.fileEncoding after: fileEncoding = desc.encoding ``` `FileScanOpDesc` (the non-source variant) declares its own public `fileEncoding` with no `@JsonIgnoreProperties`, so it was never affected and is untouched here. ### Any related issues, documentation, discussions? Closes #8596 ### How was this PR tested? `FileScanSourceOpDescSpec` already had a "with US_ASCII encoding" case, but it set the inherited `fileEncoding`, so it exercised the encoding path without being able to detect this bug: ASCII and UTF-8 agree on ASCII bytes, so it passed either way. It now sets `encoding`. The `before` block was pointed at `encoding` for the same reason. Two cases were added: | Case | Asserts | | --- | --- | | carry the Encoding field through serialization into the executor | the charset survives the `writeValueAsString` to `readValue` round trip that `getPhysicalOp` uses to reach the executor | | decode a UTF-16 file with the charset the Encoding field names | a real UTF-16 temp file decodes to its lines, not to a BOM plus NUL-interleaved characters | The second case is a genuine regression test. It was confirmed to fail on the old wiring before the fix was applied, with the executor line reverted to `desc.fileEncoding`: ``` [info] should decode a UTF-16 file with the charset the Encoding field names *** FAILED *** [info] processedTuple.next().getField[Nothing]("line").equals("line1") was false [info] (FileScanSourceOpDescSpec.scala:251) [info] Tests: succeeded 13, failed 1 ``` and to pass with the fix in place: ``` [info] Tests: succeeded 14, failed 0, canceled 0, ignored 0, pending 0 ``` No regressions across the sibling scan operators: ``` sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.source.scan.*" [info] Suites: completed 18, aborted 0 [info] Tests: succeeded 164, failed 0, canceled 0, ignored 0, pending 0 ``` `sbt WorkflowOperator/scalafmtAll` and `sbt WorkflowOperator/scalafixAll` produce no further diff. ### Was this PR authored or co-authored using generative AI tooling? Yes, partially. I (Suyash Jain) worked on this PR together with Claude Code as a pair-programming assistant. The fix was verified locally by confirming the new regression test fails before it and passes after, and by running the full scan-operator suite. --- .../source/scan/file/FileScanSourceOpDesc.scala | 6 ++- .../source/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 82997632d1..c49a616c33 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 d47cf3681c..254aa086fd 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 fafb696f13..6ac320e99c 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 =
