{"record":{"id":"e6f50ad0e8fed06b","repo":"apache/beam","slug":"type-of-element-must-match-the-dofn-type","errorCode":null,"errorMessage":"Type of @Element must match the DoFn type","messagePattern":"Type of @Element must match the DoFn type","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java","lineNumber":654,"sourceCode":"      throw new UnsupportedOperationException(\n          String.format(\n              \"%s is splittable and uses timer family, but these are not compatible\",\n              fn.getClass().getName()));\n    }\n  }\n\n  /**\n   * Extract information on how the DoFn uses schemas. In particular, if the schema of an element\n   * parameter does not match the input PCollection's schema, convert.\n   */\n  @Internal\n  public static DoFnSchemaInformation getDoFnSchemaInformation(\n      DoFn<?, ?> fn, PCollection<?> input) {\n    DoFnSignature signature = DoFnSignatures.getSignature(fn.getClass());\n    DoFnSignature.ProcessElementMethod processElementMethod = signature.processElement();\n    if (!processElementMethod.getSchemaElementParameters().isEmpty()) {\n      if (!input.hasSchema()) {\n        throw new IllegalArgumentException(\"Type of @Element must match the DoFn type\" + input);\n      }\n    }\n\n    SchemaRegistry schemaRegistry = input.getPipeline().getSchemaRegistry();\n    DoFnSchemaInformation doFnSchemaInformation = DoFnSchemaInformation.create();\n    for (SchemaElementParameter parameter : processElementMethod.getSchemaElementParameters()) {\n      TypeDescriptor<?> elementT = parameter.elementT();\n      FieldAccessDescriptor accessDescriptor =\n          getFieldAccessDescriptorFromParameter(\n              parameter.fieldAccessString(),\n              input.getSchema(),\n              signature.fieldAccessDeclarations(),\n              fn);\n      doFnSchemaInformation = doFnSchemaInformation.withFieldAccessDescriptor(accessDescriptor);\n      Schema selectedSchema = SelectHelpers.getOutputSchema(input.getSchema(), accessDescriptor);\n      ConvertHelpers.ConvertedSchemaInformation converted =\n          ConvertHelpers.getConvertedSchemaInformation(selectedSchema, elementT, schemaRegistry);\n      if (converted.outputSchemaCoder != null) {","sourceCodeStart":636,"sourceCodeEnd":672,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java#L636-L672","documentation":"When a DoFn's @ProcessElement declares schema element parameters (e.g. @SchemaElementParameter annotations on @Element), the input PCollection must have a schema so Beam can map schema fields onto the element type. getDoFnSchemaInformation throws IllegalArgumentException if schema parameters are present but the input PCollection has no schema.","triggerScenarios":"Calling ParDo.getDoFnSchemaInformation(fn, input) (during pipeline construction/validation) where fn's @Element parameter carries schema element parameters but input.hasSchema() is false — e.g. input is a KV, primitive, or unregistered type rather than a schema-bearing Row/POJO.","commonSituations":"Using @SchemaElementParameter (e.g. field access ordering hints) on a DoFn fed by a non-schema PCollection; feeding an Avro/Row-typed pipeline stage into a DoFn expecting schema conversion of a non-schema input.","solutions":["Make the input PCollection schema-bearing: use a schema-registered type (POJO with @DefaultSchema, Avro record, or Row) via PCollection.setSchema(...)/beam converters (SetSchema/Convert.to(Row.class))","Remove the schema element parameters from the @Element parameter if schema mapping is not needed","Convert the input to Row first (e.g. apply Convert.toRow()) before the DoFn"],"exampleFix":"// before\nPCollection<KV<String, V>> kv = ...;\nkv.apply(ParDo.of(doFnWithSchemaElementParams));\n// after\nPCollection<Row> rows = kv.apply(Convert.toRow());\nrows.apply(ParDo.of(doFnWithSchemaElementParams));","handlingStrategy":"validation","validationCode":"if (hasSchemaElementParameters(fn) && !input.hasSchema()) {\n  throw new IllegalArgumentException(\"Input PCollection must have a schema for @SchemaElementParameter\");\n}","typeGuard":"boolean schemaCompatible(DoFn<?, ?> fn, PCollection<?> input) {\n  DoFnSignature sig = DoFnSignatures.getSignature(fn.getClass());\n  return sig.processElement().getSchemaElementParameters().isEmpty() || input.hasSchema();\n}","tryCatchPattern":"try {\n  return ParDo.getDoFnSchemaInformation(fn, input);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().startsWith(\"Type of @Element must match the DoFn type\")) {\n    throw new IllegalStateException(\"Convert input to a schema-bearing PCollection (Row/POJO) first\", e);\n  }\n  throw e;\n}","preventionTips":["Only annotate @Element with schema element parameters when the input is a schema-bearing PCollection","Register schemas for custom types (beam:schemas) before such DoFns","Convert non-schema inputs with Convert.toRow() upstream of schema-dependent DoFns"],"tags":["java","apache-beam","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"}