CSV scan does not auto-rename empty column headers; downstream Iceberg writer crashes with "Empty name" at flush time #5279

What happened?

A CSV with an empty column header (e.g. a trailing comma in the header row — common in Excel exports, third-party CSVs, and auto-generated dumps) crashes the operator output writer at flush time. The entire execution result for that operator port is lost.

Stack trace:

java.lang.IllegalArgumentException: Empty name
    at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:143)
    at org.apache.iceberg.avro.AvroSchemaUtil.validAvroName(AvroSchemaUtil.java:492)
    at org.apache.iceberg.avro.AvroSchemaUtil.makeCompatibleName(AvroSchemaUtil.java:484)
    at org.apache.iceberg.parquet.TypeToMessageType.primitive(TypeToMessageType.java:122)
    at org.apache.iceberg.parquet.TypeToMessageType.field(TypeToMessageType.java:87)
    at org.apache.iceberg.parquet.TypeToMessageType.convert(TypeToMessageType.java:64)
    at org.apache.iceberg.parquet.ParquetSchemaUtil.convert(ParquetSchemaUtil.java:41)
    at org.apache.iceberg.parquet.Parquet$WriteBuilder.build(Parquet.java:311)
    at org.apache.iceberg.parquet.Parquet$DataWriteBuilder.build(Parquet.java:760)
    at o.a.t.amber.core.storage.result.iceberg.IcebergTableWriter.flushBuffer(IcebergTableWriter.scala)
    at o.a.t.amber.core.storage.result.iceberg.IcebergTableWriter.close(IcebergTableWriter.scala)
    at o.a.t.amber.engine.architecture.worker.managers.OutputPortResultWriterThread.run(OutputPortResultWriterThread.scala)

Root cause

CSVScanSourceOpDesc.scala:107-110 feeds the raw header string straight into Attribute:

new Attribute(
  if (hasHeader) firstRow(i) else s"column-${i + 1}",
  attributeTypeList(i)
)

If firstRow(i) is "" (CSV with id,name,,age → position 2), the resulting Attribute has name "". Same in ParallelCSVScanSourceOpDesc.scala. Neither sanitizes.

The empty name flows through Schema → Iceberg Table → OutputPortResultWriterThread.run → IcebergTableWriter.close → flushBuffer → Parquet.build → ParquetSchemaUtil.convert → AvroSchemaUtil.validAvroName, where Avro forbids empty names.

Regression context

This is a regression introduced by #3295 ("Use Output Ports of an Operator to Write Storage", merged 2025-03-09). Before #3295 sink operators wrote results on a path that did not trigger the Parquet/Avro name check, so empty-name attributes propagated harmlessly. After #3295 every output port writes via OutputPortResultWriterThread → IcebergTableWriter → Parquet and the validation fires. The CSV scan side has always allowed empty headers to propagate — the bug pre-existed but was latent.

Proposed fix

Auto-rename empty header positions to column-<index> at CSV scan, following the precedent set by pandas (Unnamed: N), Spark (_cN), R (VN), and DuckDB (columnN).

Concretely, in CSVScanSourceOpDesc.scala:107-110:

new Attribute(
  if (hasHeader && firstRow(i).nonEmpty) firstRow(i) else s"column-${i + 1}",
  attributeTypeList(i)
)

And the same in ParallelCSVScanSourceOpDesc.scala.

A defensive complementary check at Schema.add / Attribute(...) (reject empty names with a clear message naming the offending operator) would make future regressions impossible.

How to reproduce?

Add this test at common/workflow-operator/src/test/scala/org/apache/amber/operator/source/scan/csv/CSVScanSourceEmptyHeaderSpec.scala:

package org.apache.amber.operator.source.scan.csv

import org.apache.amber.core.storage.FileResolver
import org.apache.amber.core.tuple.Schema
import org.scalatest.flatspec.AnyFlatSpec
import java.nio.charset.StandardCharsets
import java.nio.file.Files

class CSVScanSourceEmptyHeaderSpec extends AnyFlatSpec {
  it should "auto-rename empty CSV column headers to column-N" in {
    val tmpFile = Files.createTempFile("empty-header-", ".csv")
    tmpFile.toFile.deleteOnExit()
    val content = "id,name,,age\n1,Alice,x,30\n2,Bob,y,25\n"
    Files.write(tmpFile, content.getBytes(StandardCharsets.UTF_8))

    val op = new CSVScanSourceOpDesc()
    op.fileName = Some(tmpFile.toString)
    op.customDelimiter = Some(",")
    op.hasHeader = true
    op.setResolvedFileName(FileResolver.resolve(op.fileName.get))

    val schema: Schema = op.sourceSchema()
    val attrNames = schema.getAttributes.map(_.getName).toList
    val emptyPositions =
      attrNames.zipWithIndex.collect { case (n, i) if n.isEmpty => i }

    assert(emptyPositions.isEmpty,
      s"Schema has empty attribute names at positions $emptyPositions; got $attrNames")
  }
}

Run:

sbt "WorkflowOperator/testOnly org.apache.amber.operator.source.scan.csv.CSVScanSourceEmptyHeaderSpec"

Output on current master:

[info] CSVScanSourceEmptyHeaderSpec:
[info] - should auto-rename empty CSV column headers to column-N *** FAILED ***
[info]   List(2) was not empty Schema produced by CSV scan contains empty attribute name(s)
[info]   at positions List(2). Full attribute names: List(id, name, , age).
[info] *** 1 TEST FAILED ***

The full-pipeline crash (the stack trace above) is reproducible by running the same CSV through any workflow that materializes results via Iceberg.

Version

1.1.0-incubating (Pre-release/Master)