apache/beam · error · RuntimeException

input must be PBegin or PCollection

Error message

input must be PBegin or PCollection

What it means

SpannerIO.CreateTransaction accepts input only of type PBegin (no signal) or PCollection (used as a Wait signal before creating the read-only transaction). Any other PInput type triggers RuntimeException 'input must be PBegin or PCollection'. The type check is done with instanceof in SpannerIO's public inner transform.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java:1247

  public abstract static class CreateTransaction
      extends PTransform<PInput, PCollectionView<Transaction>> {

    abstract SpannerConfig getSpannerConfig();

    abstract @Nullable TimestampBound getTimestampBound();

    abstract Builder toBuilder();

    @Override
    public PCollectionView<Transaction> expand(PInput input) {
      getSpannerConfig().validate();

      PCollection<?> collection = input.getPipeline().apply(Create.of(1));

      if (input instanceof PCollection) {
        collection = collection.apply(Wait.on((PCollection<?>) input));
      } else if (!(input instanceof PBegin)) {
        throw new RuntimeException("input must be PBegin or PCollection");
      }

      return collection
          .apply(
              "Create transaction",
              ParDo.of(new CreateTransactionFn(this.getSpannerConfig(), this.getTimestampBound())))
          .apply("As PCollectionView", View.asSingleton());
    }

    /** Specifies the Cloud Spanner configuration. */
    public CreateTransaction withSpannerConfig(SpannerConfig spannerConfig) {
      return toBuilder().setSpannerConfig(spannerConfig).build();
    }

    /** Specifies the Cloud Spanner project. */
    public CreateTransaction withProjectId(String projectId) {
      return withProjectId(ValueProvider.StaticValueProvider.of(projectId));
    }

View on GitHub (pinned to 12126d8942)

Solutions

1. Apply SpannerIO.createTransaction() directly on pipeline.begin() (PBegin) or a single PCollection
2. Use Wait.on(pcollection) by passing the PCollection, or apply the transform to a PBegin and combine with Wait
3. Unwrap the desired PCollection from multi-output results before applying

Example fix

// before
pipeline.apply(multiOutputTransform).apply(SpannerIO.createTransaction());
// after
PCollection<T> out = pipeline.apply(multiOutputTransform).get(mainTag);
out.apply(Wait.on(out)) /* via SpannerIO.read().withWait... */ ;

if (!(input instanceof PBegin) && !(input instanceof PCollection)) {
  throw new IllegalArgumentException("CreateTransaction input must be PBegin or PCollection");
}

boolean validCreateTransactionInput(PInput in) { return in instanceof PBegin || in instanceof PCollection; }

try { in.apply(SpannerIO.createTransaction()); } catch (RuntimeException e) { /* unwrap correct PCollection and retry */ }

When it happens

Trigger: Applying CreateTransaction (e.g. via SpannerIO.createTransaction() chained off a PCollectionList, PBegin-derived custom input, or a tagged/multiple-output result) that is neither PBegin nor PCollection.

Common situations: Chaining createTransaction after a transform that returns PCollectionTuple or a multi-output POutput; passing a PCollectionList; wiring Wait signals incorrectly.

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

• Unhandled input type ${input.getClass()}
• Illegal access to pipeline after visitor traversal was compl
• Pipeline update will not be possible because the following t
• Unable to provide coder for %s, this factory can only provid
• The input schema must have exactly one field of type byte.

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