{"record":{"id":"fc527dab8342747f","repo":"apache/beam","slug":"sql-can-only-run-over-pcollections-that-have-schemas","errorCode":null,"errorMessage":"SQL can only run over PCollections that have schemas.","messagePattern":"SQL can only run over PCollections that have schemas\\.","errorType":"exception","errorClass":"java.lang.IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamPCollectionTable.java","lineNumber":37,"sourceCode":"\nimport org.apache.beam.sdk.extensions.sql.meta.SchemaBaseBeamTable;\nimport org.apache.beam.sdk.schemas.transforms.Convert;\nimport org.apache.beam.sdk.values.PBegin;\nimport org.apache.beam.sdk.values.PCollection;\nimport org.apache.beam.sdk.values.POutput;\nimport org.apache.beam.sdk.values.Row;\n\n/**\n * {@code BeamPCollectionTable} converts a {@code PCollection<Row>} as a virtual table, then a\n * downstream query can query directly.\n */\npublic class BeamPCollectionTable<InputT> extends SchemaBaseBeamTable {\n  private transient PCollection<InputT> upstream;\n\n  public BeamPCollectionTable(PCollection<InputT> upstream) {\n    super(upstream.getSchema());\n    if (!upstream.hasSchema()) {\n      throw new IllegalArgumentException(\"SQL can only run over PCollections that have schemas.\");\n    }\n    this.upstream = upstream;\n  }\n\n  @Override\n  public PCollection.IsBounded isBounded() {\n    return upstream.isBounded();\n  }\n\n  @Override\n  public PCollection<Row> buildIOReader(PBegin begin) {\n    assert begin.getPipeline() == upstream.getPipeline();\n    return upstream.apply(Convert.toRows());\n  }\n\n  @Override\n  public POutput buildIOWriter(PCollection<Row> input) {\n    throw new IllegalArgumentException(\"cannot use [BeamPCollectionTable] as target\");","sourceCodeStart":19,"sourceCodeEnd":55,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamPCollectionTable.java#L19-L55","documentation":"BeamPCollectionTable wraps an existing PCollection so SQL can query it as a table, but SQL requires a schema to map columns. The constructor throws IllegalArgumentException if the input PCollection does not have a schema attached.","triggerScenarios":"Calling sqlContext.registerTable (or constructing BeamPCollectionTable directly) with a PCollection<POJO>/PCollection<Row> created without apply(Schema) — e.g. a PCollection<String> from a text source registered as-is.","commonSituations":"Registering a raw text/JSON PCollection without converting to typed rows or applying a schema; using Java 8 lambdas producing untyped types that Beam cannot infer schemas for.","solutions":["Convert the PCollection to Rows and apply a schema: pcoll.apply(Convert.toRows(...)).setSchema(...)","Use BeamSQL's built-in connectors (e.g. BeamTextTable) instead of wrapping a raw PCollection","Ensure the element type is a class with schema-inferable fields (annotated @DefaultSchema or a simple POJO) and call setRowSchema/setSchema"],"exampleFix":"// before\npcoll.apply(ParDo.of(new ParseFn()));\nsqlContext.registerTable(\"t\", new BeamPCollectionTable<>(pcoll));\n// after\nPCollection<Row> rows = pcoll.apply(...).setRowSchema(mySchema);\nsqlContext.registerTable(\"t\", new BeamPCollectionTable<>(rows));","handlingStrategy":"validation","validationCode":"if (!pcollection.hasSchema()) { throw new IllegalArgumentException(\"apply a schema before registering table\"); }","typeGuard":"boolean isSqlReady(PCollection<?> p) { return p.hasSchema(); }","tryCatchPattern":"try { table = new BeamPCollectionTable<>(pcoll); } catch (IllegalArgumentException e) { pcoll = pcoll.apply(Convert.toRows(...)).setRowSchema(schema); }","preventionTips":["Always call setSchema/setRowSchema on PCollections before SQL registration","Use typed POJOs with @DefaultSchema annotations","Prefer built-in table connectors for raw text sources"],"tags":["sql","ptransform","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"}