fix(workflow-operator): sample at least one row when Limit is 0

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala

@@ -115,7 +115,9 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc {
     parser.beginParsing(inputReader)
 
     var data: Array[Array[String]] = Array()
-    val readLimit = limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)
+    // A Limit of 0 asks for zero output rows, not zero rows to infer the schema from,
+    // so the sample always includes at least one row regardless of the output limit.
+    val readLimit = math.max(limit.getOrElse(INFER_READ_LIMIT), 1).min(INFER_READ_LIMIT)
     for (_ <- 0 until readLimit) {
       val row = CSVScanSourceOpExec.parseNextRow(parser, maxColumns)
       if (row != null) {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala

@@ -108,9 +108,11 @@ class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc {
     if (hasHeader)
       reader.readNext()
 
+    // A Limit of 0 asks for zero output rows, not zero rows to infer the schema from,
+    // so the sample always includes at least one row regardless of the output limit.
     val attributeTypeList: Array[AttributeType] = inferSchemaFromRows(
       reader.iterator
-        .take(limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT))
+        .take(math.max(limit.getOrElse(INFER_READ_LIMIT), 1).min(INFER_READ_LIMIT))
         .map(seq => seq.toArray)
     )
 

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala

@@ -103,8 +103,10 @@ class CSVOldScanSourceOpDesc extends ScanSourceOpDesc {
     reader = CSVReader.open(file, fileEncoding.getCharset.name())(CustomFormat)
 
     val startOffset = offset.getOrElse(0) + (if (hasHeader) 1 else 0)
+    // A Limit of 0 asks for zero output rows, not zero rows to infer the schema from,
+    // so the sample always includes at least one row regardless of the output limit.
     val endOffset =
-      startOffset + limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)
+      startOffset + math.max(limit.getOrElse(INFER_READ_LIMIT), 1).min(INFER_READ_LIMIT)
     val attributeTypeList: Array[AttributeType] = inferSchemaFromRows(
       reader.iterator
         .slice(startOffset, endOffset)

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala

@@ -87,8 +87,10 @@ class JSONLScanSourceOpDesc extends ScanSourceOpDesc {
     val allFields: ArrayBuffer[Map[String, String]] = ArrayBuffer()
 
     val startOffset = offset.getOrElse(0)
+    // A Limit of 0 asks for zero output rows, not zero rows to infer the schema from,
+    // so the sample always includes at least one row regardless of the output limit.
     val endOffset =
-      startOffset + limit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)
+      startOffset + math.max(limit.getOrElse(INFER_READ_LIMIT), 1).min(INFER_READ_LIMIT)
     reader
       .lines()
       .iterator()

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala

@@ -97,6 +97,17 @@ class CSVScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     opDesc.sourceSchema().getAttributes.map(_.getName).toList
   }
 
+  // Writes a two-column, two-data-row CSV and returns the absolute path.
+  private def writeSimpleCsv(): String = {
+    val tmpFile = Files.createTempFile("simple-", ".csv")
+    tmpFile.toFile.deleteOnExit()
+    Files.write(
+      tmpFile,
+      "id,name\n1,Alice\n2,Bob\n".getBytes(StandardCharsets.UTF_8)
+    )
+    tmpFile.toString
+  }
+
   // Writes a numeric column with one blank cell and returns the absolute path.
   private def writeCsvWithBlankNumericCell(): String = {
     val tmpFile = Files.createTempFile("blank-cell-", ".csv")
@@ -321,4 +332,24 @@ class CSVScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(columnNames(oldCsv, path) == List("id", "name", "age"))
   }
 
+  // Limit bounds the operator's output row count, not the sample sourceSchema() reads
+  // to infer columns. A Limit of 0 used to leave that sample empty: CSV and parallel
+  // CSV then reported a schema with no attributes, and old CSV threw
+  // ArrayIndexOutOfBoundsException reading the (empty) inferred type of its first
+  // header, before a single row was read.
+  it should "still infer the file's columns when Limit is 0" in {
+    val path = writeSimpleCsv()
+
+    val csv = new CSVScanSourceOpDesc()
+    csv.limit = Some(0)
+    val parallelCsv = new ParallelCSVScanSourceOpDesc()
+    parallelCsv.limit = Some(0)
+    val oldCsv = new CSVOldScanSourceOpDesc()
+    oldCsv.limit = Some(0)
+
+    assert(columnNames(csv, path) == List("id", "name"))
+    assert(columnNames(parallelCsv, path) == List("id", "name"))
+    assert(columnNames(oldCsv, path) == List("id", "name"))
+  }
+
 }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala

@@ -20,6 +20,7 @@
 package org.apache.texera.amber.operator.source.scan.json
 
 import org.apache.texera.amber.core.executor.OpExecWithClassName
+import org.apache.texera.amber.core.storage.FileResolver
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
@@ -28,6 +29,9 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
 class JSONLScanSourceOpDescSpec extends AnyFlatSpec with Matchers {
 
   private val workflowId = WorkflowIdentity(1L)
@@ -72,6 +76,28 @@ class JSONLScanSourceOpDescSpec extends AnyFlatSpec with Matchers {
     physical.outputPorts.keySet shouldBe d.operatorInfo.outputPorts.map(_.id).toSet
   }
 
+  // Limit bounds the operator's output row count, not the sample sourceSchema() reads
+  // to infer columns. A Limit of 0 used to leave that sample empty, so the operator
+  // reported a schema with no attributes at all.
+  "JSONLScanSourceOpDesc.sourceSchema" should "still infer the file's columns when Limit is 0" in {
+    val tmpFile = Files.createTempFile("jsonl-", ".jsonl")
+    tmpFile.toFile.deleteOnExit()
+    Files.write(
+      tmpFile,
+      """{"id": 1, "name": "Alice"}
+        |{"id": 2, "name": "Bob"}
+        |""".stripMargin.getBytes(StandardCharsets.UTF_8)
+    )
+    val path = tmpFile.toString
+
+    val d = new JSONLScanSourceOpDesc
+    d.limit = Some(0)
+    d.fileName = Some(path)
+    d.setResolvedFileName(FileResolver.resolve(path))
+
+    d.sourceSchema().getAttributes.map(_.getName).toList shouldBe List("id", "name")
+  }
+
   "JSONLScanSourceOpDesc" should "round-trip its config fields through the polymorphic base" in {
     val d = new JSONLScanSourceOpDesc
     d.flatten = true