apache/beam · error · RuntimeException

The input to FillGaps must have a schema.

Error message

The input to FillGaps must have a schema.

What it means

FillGaps.expand requires its input PCollection to have a Beam schema, because gap-filling operates on typed Row keys and timestamp fields. If the input has no schema (e.g. plain strings or KV without registered schema), it throws RuntimeException immediately at pipeline construction.

Solutions

1. Use an input PCollection of a schema-registered type (POJO with @DefaultSchema, Avro record, or Row).
2. Call PCollection.setSchema(...) / apply a transform that produces a schemaful output before FillGaps.
3. Verify with input.hasSchema() during pipeline construction and fail with a clear message early.

Example fix

// before
PCollection<String> input = pipeline.apply(Create.of("a", "b"));
input.apply(FillGaps.create(...)); // throws
// after
PCollection<Row> input = pipeline
    .apply(Create.of("a", "b"))
    .apply(MapElements.into(TypeDescriptor.of(Row.class)).via(...))
    .setSchema(schema);
input.apply(FillGaps.create(...));

if (!input.hasSchema()) {
  throw new IllegalStateException("FillGaps input must have a schema; apply setSchema() first");
}

try {
  input.apply(FillGaps.create(...));
} catch (RuntimeException e) {
  if (e.getMessage() != null && e.getMessage().contains("must have a schema")) {
    // convert input to Row/POJO and set a schema before retrying
  }
}

When it happens

Trigger: Applying FillGaps.create(...) to a PCollection created via pipeline.apply(Create.of(...)) of simple types, or a PCollections whose element class has no @DefaultSchema/GetSchema registered.

Common situations: Reading untyped records from text/Kafka without a schema; passing PCollection<String> instead of a schemaful POJO/Row; forgetting to call setSchema or use schema-aware coders.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• ArrowFlightIO.write(): e.getMessage()
• BeamRowMapper does not have support for fields of type
• can't generate row coder for type
• Cannot find Spanner table.
• cannot make schema for type

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

Appendix: source

Thrown at sdks/java/extensions/timeseries/src/main/java/org/apache/beam/sdk/extensions/timeseries/FillGaps.java:247

      SerializableBiFunction<
              TimestampedValue<ValueT>, TimestampedValue<ValueT>, TimestampedValue<ValueT>>
          mergeFunction) {
    return toBuilder().setMergeValues(mergeFunction).build();
  }

  /**
   * This function can be used to modify elements before propagating to the next bucket. A common
   * use case is to modify a contained timestamp to match that of the new bucket.
   */
  public FillGaps<ValueT> withInterpolateFunction(
      SerializableFunction<InterpolateData<ValueT>, ValueT> interpolateFunction) {
    return toBuilder().setInterpolateFunction(interpolateFunction).build();
  }

  @Override
  public PCollection<ValueT> expand(PCollection<ValueT> input) {
    if (!input.hasSchema()) {
      throw new RuntimeException("The input to FillGaps must have a schema.");
    }

    FixedWindows bucketWindows = FixedWindows.of(getTimeseriesBucketDuration());
    // TODO(reuvenlax, BEAM-12795): We need to create KVs to use state/timers. Once BEAM-12795 is
    // fixed we can dispense with the KVs here.
    PCollection<KV<Row, ValueT>> keyedValues =
        input
            .apply("FixedWindow", Window.into(bucketWindows))
            .apply("withKeys", WithKeys.of(getKeyDescriptor()));

    WindowFn<ValueT, BoundedWindow> originalWindowFn =
        (WindowFn<ValueT, BoundedWindow>) input.getWindowingStrategy().getWindowFn();
    return keyedValues
        .apply("globalWindow", Window.into(new GlobalWindows()))
        .apply(
            "fillGaps",
            ParDo.of(
                new FillGapsDoFn<>(

View on GitHub (pinned to 12126d8942)