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-7261-6cfd7244a7fc804e73dba9386751399e4a3fb0fd in repository https://gitbox.apache.org/repos/asf/texera.git
commit 69c66a17647f3e530a90486d7acde5b0120cda74 Author: Kary Zheng <[email protected]> AuthorDate: Wed Aug 5 17:01:01 2026 -0700 feat(csv-scan): declare the Delimiter as the single character it is (#7261) ### What changes were proposed in this PR? `customDelimiter` on the three CSV scans is declared as a string with no length constraint, while every reader narrows it with `charAt(0)` — univocity's `setDelimiter` and scala-csv's `DefaultCSVFormat.delimiter` both take a `Char`. Each now declares `maxLength: 1`, and the description says a single character rather than leaving it implied. ### Why are the changes needed? Characters past the first are dropped without a word. Driving each operator's own `sourceSchema()` over a three-column `;`-separated file, `;abc` infers the same three columns as `;`, and `,;` collapses to one column exactly as a bare `,` does — so the second character never participates. A workflow typed that way runs to completion and looks right, and nothing ever says part of the field was discarded. ### Any related issues, documentation, discussions? Closes #7211 ### How was this PR tested? `WorkflowOperator/compile`, `scalafmtCheckAll`, `scalafixAll --check`, and the three operators' descriptor specs (22 tests, three of them added here for the constraint itself). ### Does this PR introduce any user-facing change? Yes, and one worth calling out. A saved workflow whose Delimiter holds more than one character will now show as invalid in the property editor, since the form validates the stored properties against the schema. It still runs — the backend does not enforce the schema — and the run behaves exactly as before, because only the first character ever counted. Surfacing those workflows is the point: today they are silently truncating. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --------- Co-authored-by: Claude Opus 5 (1M context) <[email protected]> --- .../source/scan/csv/CSVScanSourceOpDesc.scala | 7 +- .../scan/csv/ParallelCSVScanSourceOpDesc.scala | 6 +- .../scan/csvOld/CSVOldScanSourceOpDesc.scala | 6 +- .../source/scan/csv/CSVScanSourceOpDescSpec.scala | 89 +++++++++++++++++++++- 4 files changed, 101 insertions(+), 7 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala index 328e3b6c1f..1a784a44ee 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala @@ -20,7 +20,7 @@ package org.apache.texera.amber.operator.source.scan.csv import com.fasterxml.jackson.annotation.{JsonInclude, JsonProperty, JsonPropertyDescription} -import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle +import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} import com.univocity.parsers.csv.{CsvFormat, CsvParser, CsvParserSettings} import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.storage.DocumentFactory @@ -37,10 +37,13 @@ import java.net.URI class CSVScanSourceOpDesc extends ScanSourceOpDesc { + // One character: every reader narrows this with charAt(0), because univocity's + // setDelimiter and scala-csv's DefaultCSVFormat both take a Char. @JsonProperty(defaultValue = ",") @JsonSchemaTitle("Delimiter") - @JsonPropertyDescription("delimiter to separate each line into fields") + @JsonPropertyDescription("single character separating the fields on each line") @JsonInclude(JsonInclude.Include.NON_ABSENT) + @JsonSchemaInject(json = """{ "maxLength": 1 }""") var customDelimiter: Option[String] = None @JsonProperty(defaultValue = "true") diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala index b3d1071b86..cdade62d1e 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala @@ -22,7 +22,7 @@ package org.apache.texera.amber.operator.source.scan.csv import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.fasterxml.jackson.databind.annotation.JsonDeserialize import com.github.tototoshi.csv.{CSVReader, DefaultCSVFormat} -import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle +import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.storage.DocumentFactory import org.apache.texera.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows @@ -37,10 +37,12 @@ import java.net.URI class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc { + // One character -- see CSVScanSourceOpDesc. @JsonProperty(defaultValue = ",") @JsonSchemaTitle("Delimiter") - @JsonPropertyDescription("delimiter to separate each line into fields") + @JsonPropertyDescription("single character separating the fields on each line") @JsonDeserialize(contentAs = classOf[java.lang.String]) + @JsonSchemaInject(json = """{ "maxLength": 1 }""") var customDelimiter: Option[String] = None @JsonProperty(defaultValue = "true") diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala index 2a18d431ff..57dec76c55 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala @@ -21,7 +21,7 @@ package org.apache.texera.amber.operator.source.scan.csvOld import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.github.tototoshi.csv.{CSVReader, DefaultCSVFormat} -import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle +import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.storage.DocumentFactory import org.apache.texera.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows @@ -36,9 +36,11 @@ import java.net.URI class CSVOldScanSourceOpDesc extends ScanSourceOpDesc { + // One character -- see CSVScanSourceOpDesc. @JsonProperty(defaultValue = ",") @JsonSchemaTitle("Delimiter") - @JsonPropertyDescription("delimiter to separate each line into fields") + @JsonPropertyDescription("single character separating the fields on each line") + @JsonSchemaInject(json = """{ "maxLength": 1 }""") var customDelimiter: Option[String] = Some(",") @JsonProperty(defaultValue = "true") diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala index 87b86fea6d..3049d7f02a 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala @@ -19,13 +19,18 @@ package org.apache.texera.amber.operator.source.scan.csv +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.TextNode +import com.github.fge.jsonschema.main.JsonSchemaFactory import org.apache.texera.amber.core.storage.FileResolver import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.workflow.WorkflowContext.{ DEFAULT_EXECUTION_ID, DEFAULT_WORKFLOW_ID } -import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.{LogicalOp, TestOperators} +import org.apache.texera.amber.operator.metadata.OperatorMetadataGenerator +import org.apache.texera.amber.operator.source.scan.ScanSourceOpDesc import org.apache.texera.amber.operator.source.scan.csvOld.CSVOldScanSourceOpDesc import org.scalatest.BeforeAndAfter import org.scalatest.flatspec.AnyFlatSpec @@ -42,6 +47,27 @@ class CSVScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { parallelCsvScanSourceOpDesc = new ParallelCSVScanSourceOpDesc() } + private val delimiterOwners: List[(String, Class[_ <: LogicalOp])] = List( + "CSV" -> classOf[CSVScanSourceOpDesc], + "parallel CSV" -> classOf[ParallelCSVScanSourceOpDesc], + "old CSV" -> classOf[CSVOldScanSourceOpDesc] + ) + + private def delimiterSchema(opDescClass: Class[_ <: LogicalOp]): JsonNode = + OperatorMetadataGenerator + .generateOperatorJsonSchema(opDescClass) + .path("properties") + .path("customDelimiter") + + // The property editor validates a stored delimiter against the schema, so validate + // the same way rather than restating the bound the schema declares. + private def schemaValidates(propertySchema: JsonNode, delimiter: String): Boolean = + JsonSchemaFactory + .byDefault() + .getJsonSchema(propertySchema) + .validate(TextNode.valueOf(delimiter)) + .isSuccess + // Writes a CSV whose header row has an empty column (the third position), // e.g. `id,name,,age`, and returns the absolute path. private def writeCsvWithEmptyHeader(): String = { @@ -54,6 +80,23 @@ class CSVScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { tmpFile.toString } + // Writes a three-column `;`-separated CSV and returns the absolute path. + private def writeSemicolonCsv(): String = { + val tmpFile = Files.createTempFile("semicolon-", ".csv") + tmpFile.toFile.deleteOnExit() + Files.write( + tmpFile, + "id;name;age\n1;Alice;30\n2;Bob;25\n".getBytes(StandardCharsets.UTF_8) + ) + tmpFile.toString + } + + private def columnNames(opDesc: ScanSourceOpDesc, path: String): List[String] = { + opDesc.fileName = Some(path) + opDesc.setResolvedFileName(FileResolver.resolve(path)) + opDesc.sourceSchema().getAttributes.map(_.getName).toList + } + it should "infer schema from single-line-data csv" in { parallelCsvScanSourceOpDesc.fileName = Some(TestOperators.CountrySalesSmallCsvPath) @@ -210,4 +253,48 @@ class CSVScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { assert(names == List("id", "name", "column-3", "age")) } + it should "declare the delimiter as a single character on every CSV scan" in { + delimiterOwners.foreach { + case (name, opDescClass) => + withClue(s"$name: ") { + val propertySchema = delimiterSchema(opDescClass) + assert(propertySchema.path("type").asText() == "string") + assert(propertySchema.path("maxLength").asInt() == 1) + } + } + } + + it should "validate an empty or one-character delimiter and refuse a longer one" in { + delimiterOwners.foreach { + case (name, opDescClass) => + withClue(s"$name: ") { + val propertySchema = delimiterSchema(opDescClass) + // Empty stays valid: the field is optional and every reader resolves an + // empty delimiter to a comma, so clearing it must not be an error. + assert(schemaValidates(propertySchema, "")) + assert(schemaValidates(propertySchema, ",")) + assert(schemaValidates(propertySchema, ";")) + assert(!schemaValidates(propertySchema, ",;")) + assert(!schemaValidates(propertySchema, ";abc")) + } + } + } + + it should "read a multi-character delimiter as its first character" in { + // What the constraint gives up: a saved workflow holding a longer delimiter now + // shows as invalid in the property editor. Its run is unchanged, which is what + // this pins -- the characters past the first never reached a parser. + val path = writeSemicolonCsv() + val csv = new CSVScanSourceOpDesc() + csv.customDelimiter = Some(";abc") + val parallelCsv = new ParallelCSVScanSourceOpDesc() + parallelCsv.customDelimiter = Some(";abc") + val oldCsv = new CSVOldScanSourceOpDesc() + oldCsv.customDelimiter = Some(";abc") + + assert(columnNames(csv, path) == List("id", "name", "age")) + assert(columnNames(parallelCsv, path) == List("id", "name", "age")) + assert(columnNames(oldCsv, path) == List("id", "name", "age")) + } + }
