What causes "%s expects a single %s tagged PCollection input" in apache/beam?

Select exactly one output tag from the producing transform before applying FileWriteSchemaTransform

apache/beam · error · IllegalArgumentException

expects a single tagged PCollection input

Error message

%s expects a single %s tagged PCollection<Row> input

What it means

FileWriteSchemaTransform.expand requires exactly one input PCollection<Row>, tagged with the transform's INPUT_TAG, inside the PCollectionRowTuple. If the tuple is empty or contains more than one tagged collection, it throws IllegalArgumentException. This guards the transform against ambiguous or missing row input.

Solutions

Select exactly one output tag from the producing transform before applying FileWriteSchemaTransform
Ensure the input tag matches the expected INPUT_TAG for the write transform
Convert your PCollection<Row> into a single-entry PCollectionRowTuple: PCollectionRowTuple.of(INPUT_TAG, rows)
Log input.getAll().keySet() before expand to debug what tags are present

Example fix

// before
PCollectionRowTuple tuple = PCollectionRowTuple.of("rows", rows)
    .apply(FileWriteSchemaTransform...); // wrong tag
// after
PCollectionRowTuple.of(INPUT_TAG, rows)
    .apply(FileWriteSchemaTransform...);

Defensive patterns

Strategy: validation

Validate before calling

if (tuple.getAll().size() != 1 || !tuple.getAll().containsKey(INPUT_TAG)) {
  throw new IllegalArgumentException("FileWriteSchemaTransform needs exactly one " + INPUT_TAG + " tagged PCollection<Row>");
}

Try / catch

if (input.getAll().size() != 1) {
  throw new IllegalStateException("Wire exactly one " + INPUT_TAG + " PCollection<Row> into FileWriteSchemaTransform; got: "
      + input.getAll().keySet());
}

Prevention

Always construct input tuples with PCollectionRowTuple.of(INPUT_TAG, rows)
When consuming multi-output transforms, select a single named output first
Add an assert in the composing code that tuple.getAll().keySet() equals Set.of(INPUT_TAG)

When it happens

Trigger: Applying FileWriteSchemaTransform via SchemaTransform providers with a PCollectionRowTuple that has zero tagged PCollections or multiple tags (e.g. feeding output of a multi-output transform without selecting one tag).

Common situations: Wiring the wrong transform output (multi-output PCollectionTuple) into the write transform; forgetting to .get() a specific tag before applying; building the tuple programmatically without adding the INPUT_TAG entry.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/547041b56cbca053. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/file-schema-transform/src/main/java/org/apache/beam/sdk/io/fileschematransform/FileWriteSchemaTransformProvider.java:114

  /**
   * A {@link PTransform} that converts a {@link PCollectionRowTuple} of {@link
   * #inputCollectionNames()} tagged {@link Row}s into a {@link PCollectionRowTuple} of {@link
   * #outputCollectionNames()} tagged {@link Row}s.
   */
  static class FileWriteSchemaTransform extends SchemaTransform {

    final FileWriteSchemaTransformConfiguration configuration;

    FileWriteSchemaTransform(FileWriteSchemaTransformConfiguration configuration) {
      validateConfiguration(configuration);
      this.configuration = configuration;
    }

    @Override
    public PCollectionRowTuple expand(PCollectionRowTuple input) {
      if (input.getAll().isEmpty() || input.getAll().size() > 1) {
        throw new IllegalArgumentException(
            String.format(
                "%s expects a single %s tagged PCollection<Row> input",
                FileWriteSchemaTransform.class.getName(), INPUT_TAG));
      }

      PCollection<Row> rowInput = input.get(INPUT_TAG);

      PTransform<PCollection<Row>, PCollectionTuple> transform =
          getProvider().buildTransform(configuration, rowInput.getSchema());

      PCollectionTuple files = rowInput.apply("Write Rows", transform);
      PCollection<Row> output =
          files
              .get(RESULT_TAG)
              .apply(
                  "Filenames to Rows",
                  MapElements.into(rows())
                      .via(

View on GitHub (pinned to 12126d8942)