{"record":{"id":"342426a3ed546cdd","repo":"apache/beam","slug":"field-must-not-be-null","errorCode":null,"errorMessage":"Field  must not be null.","messagePattern":"Field  must not be null\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"critical","filePath":"sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java","lineNumber":230,"sourceCode":"\n          if (!task.isCDC()) {\n            // For ADD files, we need to append the constant CDF columns:\n            // _change_type = \"insert\", _commit_version = task.version, _commit_timestamp =\n            // task.timestamp\n            ColumnarBatch logicalBatch =\n                appendConstantCDFColumns(\n                    currentEngine, batch.getData(), task.getVersion(), task.getTimestamp());\n            // Make sure we use selection vector to considered filtered out or deleted rows.\n            batch = new FilteredColumnarBatch(logicalBatch, batch.getSelectionVector());\n          }\n\n          try (CloseableIterator<io.delta.kernel.data.Row> logicalRows = batch.getRows()) {\n            while (logicalRows.hasNext()) {\n              io.delta.kernel.data.Row deltaRow = logicalRows.next();\n              Row beamRow = DeltaSourceDoFn.toBeamRow(deltaRow, beamSchema);\n              String changeType = beamRow.getString(DeltaIO.CHANGE_TYPE_COLUMN);\n              if (changeType == null) {\n                throw new IllegalStateException(\n                    \"Field \" + DeltaIO.CHANGE_TYPE_COLUMN + \" must not be null.\");\n              }\n              ValueKind kind = getValueKind(changeType);\n              Row publicRow = projectRow(beamRow, publicBeamSchema, task);\n              out.builder(publicRow).setValueKind(kind).output();\n            }\n          }\n        }\n      }\n    }\n  }\n\n  private static Row projectRow(Row row, Schema targetSchema, DeltaCDCReadTask task) {\n    Row.Builder builder = Row.withSchema(targetSchema);\n    for (Schema.Field field : targetSchema.getFields()) {\n      Object value = row.getValue(field.getName());\n      if (value == null) {\n        if (field.getName().equals(DeltaIO.COMMIT_VERSION_COLUMN)) {","sourceCodeStart":212,"sourceCodeEnd":248,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java#L212-L248","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the table has change data feed enabled for the whole read range: ALTER TABLE t SET TBLPROPERTIES ('delta.enableChangeDataFeed'=true).","Verify the source actually reads CDC ranges (readChanges) rather than a plain snapshot, so only CDF batches reach this DoFn.","Check appendCDFColumns/schema handling — confirm the _change_type column is appended and not projected away by column pruning or a custom schema.","Filter out data-file rows before this DoFn; only add-file (CDF) batches should be processed here."],"exampleFix":"// before\nDeltaIO.read().from(\"t\").readChanges(); // table lacks delta.enableChangeDataFeed=true\n// after\n-- run once on the table\nALTER TABLE t SET TBLPROPERTIES ('delta.enableChangeDataFeed' = true);","handlingStrategy":"validation","validationCode":"// verify CDF is enabled before reading\nSparkSession spark = ...;\nboolean cdf = \"true\".equals(spark.sql(\"DESCRIBE TABLE DETAIL \" + path).first().getAs(\"delta.enableChangeDataFeed\"));\nif (!cdf) throw new IllegalStateException(\"Enable delta.enableChangeDataFeed before CDC reads on \" + path);","typeGuard":null,"tryCatchPattern":"try { /* CDC read */ } catch (IllegalStateException e) { if (e.getMessage().contains(\"_change_type must not be null\")) { enableCDFAndRerun(); } else { throw e; } }","preventionTips":["Enable delta.enableChangeDataFeed on the table and keep it enabled across the whole read range.","Only feed CDF/change batches into DeltaCDCSourceDoFn, never plain snapshot scans.","Don't project away or overwrite the _change_type column with custom schemas."],"tags":["java","delta-lake","cdc","null-field","schema"],"backgroundTag":"null-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}