{"record":{"id":"aa6af17ad16244f8","repo":"apache/beam","slug":"the-input-to-fillgaps-must-have-a-schema","errorCode":null,"errorMessage":"The input to FillGaps must have a schema.","messagePattern":"The input to FillGaps must have a schema\\.","errorType":"validation","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/timeseries/src/main/java/org/apache/beam/sdk/extensions/timeseries/FillGaps.java","lineNumber":247,"sourceCode":"      SerializableBiFunction<\n              TimestampedValue<ValueT>, TimestampedValue<ValueT>, TimestampedValue<ValueT>>\n          mergeFunction) {\n    return toBuilder().setMergeValues(mergeFunction).build();\n  }\n\n  /**\n   * This function can be used to modify elements before propagating to the next bucket. A common\n   * use case is to modify a contained timestamp to match that of the new bucket.\n   */\n  public FillGaps<ValueT> withInterpolateFunction(\n      SerializableFunction<InterpolateData<ValueT>, ValueT> interpolateFunction) {\n    return toBuilder().setInterpolateFunction(interpolateFunction).build();\n  }\n\n  @Override\n  public PCollection<ValueT> expand(PCollection<ValueT> input) {\n    if (!input.hasSchema()) {\n      throw new RuntimeException(\"The input to FillGaps must have a schema.\");\n    }\n\n    FixedWindows bucketWindows = FixedWindows.of(getTimeseriesBucketDuration());\n    // TODO(reuvenlax, BEAM-12795): We need to create KVs to use state/timers. Once BEAM-12795 is\n    // fixed we can dispense with the KVs here.\n    PCollection<KV<Row, ValueT>> keyedValues =\n        input\n            .apply(\"FixedWindow\", Window.into(bucketWindows))\n            .apply(\"withKeys\", WithKeys.of(getKeyDescriptor()));\n\n    WindowFn<ValueT, BoundedWindow> originalWindowFn =\n        (WindowFn<ValueT, BoundedWindow>) input.getWindowingStrategy().getWindowFn();\n    return keyedValues\n        .apply(\"globalWindow\", Window.into(new GlobalWindows()))\n        .apply(\n            \"fillGaps\",\n            ParDo.of(\n                new FillGapsDoFn<>(","sourceCodeStart":229,"sourceCodeEnd":265,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/timeseries/src/main/java/org/apache/beam/sdk/extensions/timeseries/FillGaps.java#L229-L265","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Use an input PCollection of a schema-registered type (POJO with @DefaultSchema, Avro record, or Row).","Call PCollection.setSchema(...) / apply a transform that produces a schemaful output before FillGaps.","Verify with input.hasSchema() during pipeline construction and fail with a clear message early."],"exampleFix":"// before\nPCollection<String> input = pipeline.apply(Create.of(\"a\", \"b\"));\ninput.apply(FillGaps.create(...)); // throws\n// after\nPCollection<Row> input = pipeline\n    .apply(Create.of(\"a\", \"b\"))\n    .apply(MapElements.into(TypeDescriptor.of(Row.class)).via(...))\n    .setSchema(schema);\ninput.apply(FillGaps.create(...));","handlingStrategy":"validation","validationCode":"if (!input.hasSchema()) {\n  throw new IllegalStateException(\"FillGaps input must have a schema; apply setSchema() first\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  input.apply(FillGaps.create(...));\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"must have a schema\")) {\n    // convert input to Row/POJO and set a schema before retrying\n  }\n}","preventionTips":["Prefer schema-aware sources (Avro, POJO with @DefaultSchema, Row).","Call setSchema explicitly when building PCollections from untyped data.","Assert hasSchema() in pipeline-building unit tests."],"tags":["beam","timeseries","schema"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}