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-8514-acc6dd1139d936d849e2aaa6abb288f8ef3e98ad in repository https://gitbox.apache.org/repos/asf/texera.git
commit 98d3d4b1ec8b02f611801d035c190fa31f56fa47 Author: Kary Zheng <[email protected]> AuthorDate: Wed Sep 30 06:11:49 2026 +0000 feat(source): bound a file scan's window at zero (#8514) ### What changes were proposed in this PR? Every file-scan source takes a Limit and an Offset from `ScanSourceOpDesc`, and neither field said it cannot be negative. Both now declare `minimum: 0`. Neither value means anything below zero, and a negative one is not read the same way twice. The executors take the window with Scala's `drop` and `take`, where a negative drop keeps every row and a negative take keeps none. The scripts the export writes take the same window with `iloc`, which counts from the end instead. So `offset = -1` is every row on one side and the last row on the other, and `limit = -1` is no rows on one side and all but the last on the other. Declaring the bound states what the operators already assume, and the property editor then refuses the value rather than passing it on. It reaches CSV, CSVOld, JSONL, Arrow and the file scans at once, all of which inherit the two fields. The schema is not revalidated on the way in, though, so a plan posted to the API or an imported workflow file still arrives with whatever the field holds. Every reader now takes the window through two accessors on the base class, `windowOffset` and `windowLimit`, which clamp at zero. The fields themselves are left as they came, and the native readers behave exactly as they did: `drop(-1)` and `drop(0)` both keep every row, and `take(-1)` and `take(0)` both keep none. What the clamp settles is that a reader counting a negative from the end cannot answer differently from the one that does not. https://github.com/user-attachments/assets/4692ec26-1c93-4ca3-b3e8-d527cd0d0922 ### Any related issues, documentation, discussions? Closes #8513, the task this change is the whole of. ### How was this PR tested? A case in `ScanSourceOpDescSpec` validates `-1`, `0` and `5` against the generated schema for both fields, through the same validator the property editor uses rather than restating the bound the schema declares. Two more deserialize a window through the same polymorphic mapper a saved workflow goes through, and check that a negative reaches the fields intact while the accessors report the empty or the whole window, and that a non-negative one and an absent one pass through untouched. The executor is covered where the value can actually arrive: `CSVScanSourceOpExecSpec` builds the exec from the descriptor's JSON, as a submitted plan does, and runs it over a real file with each negative in turn. A negative limit emits no rows and a negative offset emits all of them. The 166 tests across the scan-source specs pass. The four numbers quoted above were read off the two runtimes rather than from memory: `Iterator(1,2,3,4,5).drop(-1)` keeps all five and `.take(-1)` keeps none, while pandas' `iloc[-1:]` is the last row and `iloc[:-1]` is the first four. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) <[email protected]> --- .../operator/source/scan/ScanSourceOpDesc.scala | 18 +++++- .../source/scan/arrow/ArrowSourceOpExec.scala | 4 +- .../source/scan/csv/CSVScanSourceOpDesc.scala | 2 +- .../source/scan/csv/CSVScanSourceOpExec.scala | 4 +- .../scan/csv/ParallelCSVScanSourceOpDesc.scala | 2 +- .../scan/csvOld/CSVOldScanSourceOpDesc.scala | 4 +- .../scan/csvOld/CSVOldScanSourceOpExec.scala | 6 +- .../source/scan/json/JSONLScanSourceOpDesc.scala | 4 +- .../source/scan/json/JSONLScanSourceOpExec.scala | 4 +- .../source/scan/ScanSourceOpDescSpec.scala | 65 +++++++++++++++++++++- .../source/scan/csv/CSVScanSourceOpExecSpec.scala | 24 ++++++++ 11 files changed, 120 insertions(+), 17 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDesc.scala index fd69cde2e0..b0a6ee5e20 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDesc.scala @@ -21,7 +21,7 @@ package org.apache.texera.amber.operator.source.scan import com.fasterxml.jackson.annotation.{JsonIgnore, JsonProperty, JsonPropertyDescription} import com.fasterxml.jackson.databind.annotation.JsonDeserialize -import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle +import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} import org.apache.texera.amber.core.storage.FileResolver import org.apache.texera.amber.core.tuple.Schema import org.apache.texera.amber.core.workflow.OutputPort @@ -52,18 +52,34 @@ abstract class ScanSourceOpDesc extends SourceOperatorDescriptor { @JsonIgnore var fileTypeName: Option[String] = None + // Neither field means anything below zero, and a negative one is not read the + // same way twice: the executors take the window with `drop` and `take`, where + // a negative drop keeps every row and a negative take keeps none, while the + // scripts that take it with `iloc` count from the end instead. @JsonProperty() @JsonSchemaTitle("Limit") @JsonPropertyDescription("max output count") @JsonDeserialize(contentAs = classOf[Int]) + @JsonSchemaInject(json = """{"minimum": 0}""") var limit: Option[Int] = None @JsonProperty() @JsonSchemaTitle("Offset") @JsonPropertyDescription("starting point of output") @JsonDeserialize(contentAs = classOf[Int]) + @JsonSchemaInject(json = """{"minimum": 0}""") var offset: Option[Int] = None + // The bound above only reaches the property editor. A plan posted to the API, or + // an imported workflow file, is deserialized without it, so every reader takes + // the window through these two: a negative that arrived that way then means the + // same empty or whole window on all of them. + @JsonIgnore + def windowOffset: Int = offset.getOrElse(0).max(0) + + @JsonIgnore + def windowLimit: Option[Int] = limit.map(_.max(0)) + override def sourceSchema(): Schema = null override def operatorInfo: OperatorInfo = { diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExec.scala index f23833206f..f5d2460025 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExec.scala @@ -88,8 +88,8 @@ class ArrowSourceOpExec( } } - var tupleIterator = rowIterator.drop(desc.offset.getOrElse(0)) - if (desc.limit.isDefined) tupleIterator = tupleIterator.take(desc.limit.get) + var tupleIterator = rowIterator.drop(desc.windowOffset) + if (desc.windowLimit.isDefined) tupleIterator = tupleIterator.take(desc.windowLimit.get) tupleIterator } 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 69088abb6c..3ec9bf9834 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 @@ -115,7 +115,7 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc { parser.beginParsing(inputReader) var data: Array[Array[String]] = Array() - val readLimit = limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) + val readLimit = windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) for (_ <- 0 until readLimit) { val row = CSVScanSourceOpExec.parseNextRow(parser, maxColumns) if (row != null) { diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala index 147238536d..4fe6f72318 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala @@ -61,7 +61,7 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat } var tupleIterator = rowIterator - .drop(desc.offset.getOrElse(0)) + .drop(desc.windowOffset) .map(row => { try { TupleLike( @@ -75,7 +75,7 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat }) .filter(t => t != null) - if (desc.limit.isDefined) tupleIterator = tupleIterator.take(desc.limit.get) + if (desc.windowLimit.isDefined) tupleIterator = tupleIterator.take(desc.windowLimit.get) tupleIterator } 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 cdade62d1e..c011a76cdf 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 @@ -110,7 +110,7 @@ class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc { val attributeTypeList: Array[AttributeType] = inferSchemaFromRows( reader.iterator - .take(limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)) + .take(windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)) .map(seq => seq.toArray) ) 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 57dec76c55..a9eff8534f 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 @@ -102,9 +102,9 @@ class CSVOldScanSourceOpDesc extends ScanSourceOpDesc { // reopen the file to read from the beginning reader = CSVReader.open(file, fileEncoding.getCharset.name())(CustomFormat) - val startOffset = offset.getOrElse(0) + (if (hasHeader) 1 else 0) + val startOffset = windowOffset + (if (hasHeader) 1 else 0) val endOffset = - startOffset + limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) + startOffset + windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) val attributeTypeList: Array[AttributeType] = inferSchemaFromRows( reader.iterator .slice(startOffset, endOffset) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala index e7b1e938c3..87ac8976e1 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala @@ -54,8 +54,8 @@ class CSVOldScanSourceOpExec private[csvOld] ( ) .filter(tuple => tuple != null) - if (desc.limit.isDefined) - tuples.take(desc.limit.get) + if (desc.windowLimit.isDefined) + tuples.take(desc.windowLimit.get) else { tuples } @@ -68,7 +68,7 @@ class CSVOldScanSourceOpExec private[csvOld] ( val filePath = DocumentFactory.openReadonlyDocument(new URI(desc.fileName.get)).asFile().toPath reader = CSVReader.open(filePath.toString, desc.fileEncoding.getCharset.name())(CustomFormat) // skip line if this worker reads the start of a file, and the file has a header line - val startOffset = desc.offset.getOrElse(0) + (if (desc.hasHeader) 1 else 0) + val startOffset = desc.windowOffset + (if (desc.hasHeader) 1 else 0) rows = reader.iterator.drop(startOffset) } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala index faccc76f88..1b0c928929 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala @@ -86,9 +86,9 @@ class JSONLScanSourceOpDesc extends ScanSourceOpDesc { val allFields: ArrayBuffer[Map[String, String]] = ArrayBuffer() - val startOffset = offset.getOrElse(0) + val startOffset = windowOffset val endOffset = - startOffset + limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) + startOffset + windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) reader .lines() .iterator() diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala index 98a6a0bdcc..9b732e73c3 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala @@ -62,9 +62,9 @@ class JSONLScanSourceOpExec private[json] ( reader = new BufferedReader( new InputStreamReader(stream, desc.fileEncoding.getCharset) ) - val offsetValue = desc.offset.getOrElse(0) + val offsetValue = desc.windowOffset var lines = reader.lines().iterator().asScala.drop(offsetValue) - if (desc.limit.isDefined) lines = lines.take(desc.limit.get) + if (desc.windowLimit.isDefined) lines = lines.take(desc.windowLimit.get) val (it1, it2) = lines.duplicate val count: Int = it1.map(_ => 1).sum diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala index 7bd1f619e8..df016eb5ab 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala @@ -19,8 +19,13 @@ package org.apache.texera.amber.operator.source.scan +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.IntNode +import com.github.fge.jsonschema.main.JsonSchemaFactory import org.apache.texera.amber.core.workflow.OutputPort +import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.apache.texera.amber.operator.metadata.OperatorMetadataGenerator import org.apache.texera.amber.operator.source.scan.csv.{ CSVScanSourceOpDesc, ParallelCSVScanSourceOpDesc @@ -195,9 +200,12 @@ class ScanSourceOpDescSpec extends AnyFlatSpec with Matchers { tree.has("fileEncoding") shouldBe true tree.has("limit") shouldBe true tree.has("offset") shouldBe true - // both are @JsonIgnore on the base class + // all @JsonIgnore on the base class; the two window accessors must not reach the + // saved JSON, or a workflow would round-trip with fields the editor never wrote tree.has("INFER_READ_LIMIT") shouldBe false tree.has("fileTypeName") shouldBe false + tree.has("windowLimit") shouldBe false + tree.has("windowOffset") shouldBe false } it should "omit unset optional fields entirely, the shape saved workflows store" in { @@ -207,4 +215,59 @@ class ScanSourceOpDescSpec extends AnyFlatSpec with Matchers { tree.has("limit") shouldBe false tree.has("offset") shouldBe false } + + // The two window fields mean nothing below zero, and a negative one is not read + // the same way twice: `drop(-1)` keeps every row while `iloc[-1:]` keeps the + // last. Validate through the schema the property editor validates against, + // rather than restating the bound it declares. + private def windowSchema(field: String): JsonNode = + OperatorMetadataGenerator + .generateOperatorJsonSchema(classOf[CSVScanSourceOpDesc]: Class[_ <: LogicalOp]) + .path("properties") + .path(field) + + private def schemaTakes(field: String, value: Int): Boolean = + JsonSchemaFactory + .byDefault() + .getJsonSchema(windowSchema(field)) + .validate(IntNode.valueOf(value)) + .isSuccess + + "A scan source's window" should "refuse a negative limit or offset" in { + Seq("limit", "offset").foreach { field => + withClue(s"$field: ") { + schemaTakes(field, -1) shouldBe false + schemaTakes(field, 0) shouldBe true + schemaTakes(field, 5) shouldBe true + } + } + } + + // Nothing revalidates the schema on the way in, so a plan posted to the API or an + // imported workflow file can still carry a value the editor would have refused. + private def deserialize(window: String): ScanSourceOpDesc = + objectMapper + .readValue( + s"""{"operatorType": "CSVFileScan", "fileName": "file:///tmp/a.csv", $window}""", + classOf[LogicalOp] + ) + .asInstanceOf[ScanSourceOpDesc] + + it should "clamp a negative one that was deserialized rather than typed" in { + val d = deserialize(""""limit": -1, "offset": -1""") + d.limit shouldBe Some(-1) + d.offset shouldBe Some(-1) + d.windowLimit shouldBe Some(0) + d.windowOffset shouldBe 0 + } + + it should "pass a non-negative one through, absent included" in { + val set = deserialize(""""limit": 2, "offset": 3""") + set.windowLimit shouldBe Some(2) + set.windowOffset shouldBe 3 + + val unset = deserialize(""""fileEncoding": "UTF_8"""") + unset.windowLimit shouldBe None + unset.windowOffset shouldBe 0 + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala index c5edd3b29f..329f6e9f07 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala @@ -185,6 +185,30 @@ class CSVScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { assert(firstCol == List("2", "3")) } + it should "read a negative window as the empty or whole one, not from the end" in { + // The schema's `minimum` binds the property editor only. `execOver` hands the + // exec the descriptor's JSON, the same way a plan posted to the API arrives, so + // a negative survives to here. A negative limit selects no rows and a negative + // offset skips none, which is what `take` and `drop` already did with it; the + // clamp is what keeps a reader that counts from the end from answering + // differently. + val rows = "a,b\n1,x\n2,y\n3,z\n" + + val emptied = execOver(writeTempCsv(rows), hasHeader = true, limit = Some(-1)) + emptied.open() + val none = + try emptied.produceTuple().toList + finally emptied.close() + assert(none.isEmpty) + + val whole = execOver(writeTempCsv(rows), hasHeader = true, offset = Some(-1)) + whole.open() + val all = + try whole.produceTuple().toList + finally whole.close() + assert(all.map(_.getFields(0).toString) == List("1", "2", "3")) + } + it should "silently drop rows that cannot be parsed into the inferred schema" in { // No header, so every line is data. The schema is inferred from the first // `limit` rows only (INFER_READ_LIMIT is capped by limit); those are integers,
