This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 98d3d4b1ec feat(source): bound a file scan's window at zero (#8514)
98d3d4b1ec is described below
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,
