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 69088abb6c9..7355bca73e1 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,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) { 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 cdade62d1e1..b0f22c99e29 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 @@ -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) ) 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 57dec76c556..5cabb9d78ec 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 @@ -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) 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 faccc76f882..ffe7d063e9d 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 @@ -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() 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 fe426552a96..af63a21d533 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 @@ -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")) + } + } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala index c0d8cf2e93c..b221c12990c 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala +++ b/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