You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/09/24 03:41:14 UTC
(texera) 01/01: feat(workflow-operator): honor the File Scan Encoding field (#8605)
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 <14...@users.noreply.github.com>
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 =