apache/beam · error · java.lang.IllegalArgumentException

input is expected to be empty

Error message

%s %s input is expected to be empty

What it means

SingleStoreSchemaTransformReadProvider is a source transform: it takes no input PCollection. Its expand() throws IllegalArgumentException when the incoming PCollectionRowTuple contains any input, because a read provider must be a pipeline root.

Solutions

1. Apply the read schema transform with no inputs (empty PCollectionRowTuple / as a root of the pipeline).
2. Remove any accidental input tags from the tuple passed to expand.
3. If you want to enrich existing data with SingleStore data, read first, then join the resulting PCollections.

Example fix

// before
PCollectionRowTuple.of("input", existingPc).apply(singleStoreReadTransform)

// after
PCollectionRowTuple.empty(pipeline).apply(singleStoreReadTransform)

if (!inputTuple.getAll().isEmpty()) {
  throw new IllegalStateException("SingleStore read transform must be a pipeline root (no inputs)");
}

When it happens

Trigger: Applying the SingleStore read schema transform to a non-empty PCollectionRowTuple, e.g. composing it downstream of another transform or passing an accidentally named/extra tag in the input map.

Common situations: Chaining the read transform after a PBegin-to-PCollection conversion; YAML pipeline DSL wiring an input into the read transform; reusing a write-config object for a read transform.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• is missing expected tag
• configuration with is not compatible with a format
• Converting to Beam schema type is not supported
• Failed to get number of partitions in the database
• Failed to validate

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

Appendix: source

Thrown at sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/schematransform/SingleStoreSchemaTransformReadProvider.java:92

  public List<String> outputCollectionNames() {
    return Collections.singletonList(OUTPUT_TAG);
  }

  /**
   * An implementation of {@link SchemaTransform} for SingleStoreDB read jobs configured using
   * {@link SingleStoreSchemaTransformReadConfiguration}.
   */
  private static class SingleStoreReadSchemaTransform extends SchemaTransform {
    private final SingleStoreSchemaTransformReadConfiguration configuration;

    SingleStoreReadSchemaTransform(SingleStoreSchemaTransformReadConfiguration configuration) {
      this.configuration = configuration;
    }

    @Override
    public PCollectionRowTuple expand(PCollectionRowTuple input) {
      if (!input.getAll().isEmpty()) {
        throw new IllegalArgumentException(
            String.format(
                "%s %s input is expected to be empty",
                input.getClass().getSimpleName(), getClass().getSimpleName()));
      }
      SingleStoreIO.DataSourceConfiguration dataSourceConfiguration =
          configuration.getDataSourceConfiguration();
      String table = configuration.getTable();
      String query = configuration.getQuery();
      Boolean outputParallelization = configuration.getOutputParallelization();
      Boolean withPartitions = configuration.getWithPartitions();

      Preconditions.checkArgument(
          !(outputParallelization != null && withPartitions != null && withPartitions),
          "outputParallelization parameter is not supported for partitioned read");

      if (withPartitions != null && withPartitions) {
        SingleStoreIO.ReadWithPartitions<Row> readWithPartitions =
            SingleStoreIO.readWithPartitionsRows();

View on GitHub (pinned to 12126d8942)