apache/beam · error · InvalidSchemaException

Unsupported schema specified for Pubsub source in CREATE TAB

Error message

Unsupported schema specified for Pubsub source in CREATE TABLE.CREATE TABLE for Pubsub topic must not be null

What it means

PubsubSchemaIOProvider.from() validates the data schema supplied in a CREATE TABLE statement via validateDataSchema(). A null schema is rejected with an InvalidSchemaException, because a Pubsub table mapping requires an explicit schema to map message fields.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubSchemaIOProvider.java:158

    validateConfigurationSchema(configuration);
    validateDlq(configuration.getString("deadLetterQueue"));
    validateDataSchema(dataSchema);
    return new PubsubSchemaIO(location, configuration, checkArgumentNotNull(dataSchema));
  }

  @Override
  public boolean requiresDataSchema() {
    return true;
  }

  @Override
  public PCollection.IsBounded isBounded() {
    return PCollection.IsBounded.UNBOUNDED;
  }

  private void validateDataSchema(@Nullable Schema schema) {
    if (schema == null) {
      throw new InvalidSchemaException(
          "Unsupported schema specified for Pubsub source in CREATE TABLE."
              + "CREATE TABLE for Pubsub topic must not be null");
    }
    if (!PubsubSchemaIO.fieldPresent(schema, TIMESTAMP_FIELD, FieldType.DATETIME)) {
      throw new InvalidSchemaException(
          "Unsupported schema specified for Pubsub source in CREATE TABLE."
              + "CREATE TABLE for Pubsub topic must include at least 'event_timestamp' field of "
              + "type 'TIMESTAMP'");
    }
  }

  private void validateDlq(@Nullable String deadLetterQueue) {
    if (deadLetterQueue != null && deadLetterQueue.isEmpty()) {
      throw new InvalidConfigurationException("Dead letter queue topic name is not specified");
    }
  }

  private void validateConfigurationSchema(Row configuration) {

View on GitHub (pinned to 12126d8942)

Solutions

Add a schema (column list with types) to the CREATE TABLE statement for the Pubsub table.
Include the required 'event_timestamp' TIMESTAMP field in that schema (also checked by the same validator).
If using the Pubsub schema service instead, go through the SchemaTransform path rather than CREATE TABLE, which requires an explicit schema.

Example fix

-- before
CREATE TABLE pubsub_table ... WITH OPTIONS (...); -- no schema
-- after
CREATE TABLE pubsub_table (payload VARCHAR, event_timestamp TIMESTAMP) ... WITH OPTIONS (...);

Defensive patterns

Strategy: validation

Validate before calling

if (tableSchema == null) { throw new IllegalArgumentException("CREATE TABLE for a Pubsub source must declare a non-null schema"); }

Try / catch

try { ioProvider.from(tableConfig); } catch (InvalidSchemaException e) { if (e.getMessage().contains("must not be null")) { /* add column list to CREATE TABLE */ } throw e; }

Prevention

Always include an explicit column list in CREATE TABLE statements for Pubsub sources.
Do not rely on schema auto-discovery — the SQL Pubsub provider requires an explicit schema.

When it happens

Trigger: Executing a CREATE TABLE DDL for a Pubsub table (e.g. via Beam SQL / Sqlline / Calcite) without a schema, so the provider's from() receives schema == null and calls validateDataSchema(null).

Common situations: Writing CREATE TABLE without a column list; relying on schema auto-discovery that the SQL provider does not perform; programmatic table config with a null schema field.

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

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