apache/beam · critical · IllegalStateException

Field must not be null.

Error message

Field  must not be null.

What it means

DeltaCDCSourceDoFn.processElement converts each Delta CDC row to a Beam Row and reads the mandatory _change_type metadata column. A null value means the row lacks the change-type field the CDF/CDC protocol guarantees, so the row cannot be classified as insert/update_preimage/update_postimage/delete and the DoFn fails fast with an IllegalStateException.

Source

Thrown at sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java:230

          if (!task.isCDC()) {
            // For ADD files, we need to append the constant CDF columns:
            // _change_type = "insert", _commit_version = task.version, _commit_timestamp =
            // task.timestamp
            ColumnarBatch logicalBatch =
                appendConstantCDFColumns(
                    currentEngine, batch.getData(), task.getVersion(), task.getTimestamp());
            // Make sure we use selection vector to considered filtered out or deleted rows.
            batch = new FilteredColumnarBatch(logicalBatch, batch.getSelectionVector());
          }

          try (CloseableIterator<io.delta.kernel.data.Row> logicalRows = batch.getRows()) {
            while (logicalRows.hasNext()) {
              io.delta.kernel.data.Row deltaRow = logicalRows.next();
              Row beamRow = DeltaSourceDoFn.toBeamRow(deltaRow, beamSchema);
              String changeType = beamRow.getString(DeltaIO.CHANGE_TYPE_COLUMN);
              if (changeType == null) {
                throw new IllegalStateException(
                    "Field " + DeltaIO.CHANGE_TYPE_COLUMN + " must not be null.");
              }
              ValueKind kind = getValueKind(changeType);
              Row publicRow = projectRow(beamRow, publicBeamSchema, task);
              out.builder(publicRow).setValueKind(kind).output();
            }
          }
        }
      }
    }
  }

  private static Row projectRow(Row row, Schema targetSchema, DeltaCDCReadTask task) {
    Row.Builder builder = Row.withSchema(targetSchema);
    for (Schema.Field field : targetSchema.getFields()) {
      Object value = row.getValue(field.getName());
      if (value == null) {
        if (field.getName().equals(DeltaIO.COMMIT_VERSION_COLUMN)) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure the table has change data feed enabled for the whole read range: ALTER TABLE t SET TBLPROPERTIES ('delta.enableChangeDataFeed'=true).
2. Verify the source actually reads CDC ranges (readChanges) rather than a plain snapshot, so only CDF batches reach this DoFn.
3. Check appendCDFColumns/schema handling — confirm the _change_type column is appended and not projected away by column pruning or a custom schema.
4. Filter out data-file rows before this DoFn; only add-file (CDF) batches should be processed here.

Example fix

// before
DeltaIO.read().from("t").readChanges(); // table lacks delta.enableChangeDataFeed=true
// after
-- run once on the table
ALTER TABLE t SET TBLPROPERTIES ('delta.enableChangeDataFeed' = true);

// verify CDF is enabled before reading
SparkSession spark = ...;
boolean cdf = "true".equals(spark.sql("DESCRIBE TABLE DETAIL " + path).first().getAs("delta.enableChangeDataFeed"));
if (!cdf) throw new IllegalStateException("Enable delta.enableChangeDataFeed before CDC reads on " + path);

try { /* CDC read */ } catch (IllegalStateException e) { if (e.getMessage().contains("_change_type must not be null")) { enableCDFAndRerun(); } else { throw e; } }

When it happens

Trigger: A batch of rows read from the Delta scan does not contain the _change_type column value — e.g. reading a non-CDF table or a source where CDF was disabled, or a Kernel scan batch (non-CDF data rows mixed into the output) passed to the CDC DoFn.

Common situations: Enabling CDC reads against a table where delta.enableChangeDataFeed was never turned on (or was disabled for some commits); schema evolution dropping the appended CDF columns; wiring DeltaSourceDoFn.toBeamRow output of a plain snapshot scan into the CDC DoFn.

Related errors

• Starting version or timestamp must be specified.
• Resolved start version %d is greater than resolved end versi
• Unsupported change type:
• Unsupported Delta type: <deltaType.getClass()>
• Unsupported metadata column %s. Supported columns are: %s, %

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