{"record":{"id":"516622ff9eecb15d","repo":"apache/beam","slug":"unexpected-logical-type","errorCode":null,"errorMessage":"Unexpected logical type: {}","messagePattern":"Unexpected logical type: (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/ApplyWatermarkColumn.java","lineNumber":86,"sourceCode":"      return null;\n    }\n    switch (field.getType().getTypeName()) {\n      case INT64:\n        return Instant.ofEpochMilli(timeUnit.toMillis((Long) value));\n      case DATETIME:\n        return (Instant) value;\n      case LOGICAL_TYPE:\n        String logicalType =\n            Preconditions.checkStateNotNull(field.getType().getLogicalType()).getIdentifier();\n        if (logicalType.equals(SqlTypes.DATETIME.getIdentifier())) {\n          return Instant.ofEpochMilli(\n              MICROSECONDS.toMillis(DateTimeUtil.microsFromTimestamp((LocalDateTime) value)));\n        } else if (logicalType.equals(SqlTypes.TIMESTAMP.getIdentifier())\n            || logicalType.equals(org.apache.beam.sdk.schemas.logicaltypes.Timestamp.IDENTIFIER)) {\n          return Instant.ofEpochMilli(\n              MICROSECONDS.toMillis(DateTimeUtil.microsFromInstant((java.time.Instant) value)));\n        } else {\n          throw new UnsupportedOperationException(\"Unexpected logical type: \" + logicalType);\n        }\n      default:\n        throw new UnsupportedOperationException(\"Unexpected Beam type: \" + field.getType());\n    }\n  }\n\n  @Override\n  public Duration getAllowedTimestampSkew() {\n    // Generous skew to cover backfill of historical data and late-arriving CDC patterns.\n    return Duration.standardDays(365);\n  }\n}\n","sourceCodeStart":68,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/ApplyWatermarkColumn.java#L68-L99","documentation":"ApplyWatermarkColumn.getInstant converts a record field value into an Instant to use as the event watermark. It recognizes only a fixed set of logical types (timestamp-with-local-tz variants). When a different logicalType reaches the converter inside a recognized switch branch, it throws UnsupportedOperationException naming the unexpected logical type.","triggerScenarios":"A Beam schema field whose logicalType identifier is neither SqlTypes.TIMESTAMP.getIdentifier() nor org.apache.beam.sdk.schemas.logicaltypes.Timestamp.IDENTIFIER is passed to getInstant, e.g. a DATE, TIME, or custom logical type used as the watermark column.","commonSituations":"Configuring the CDC/watermark column as a DATE or naive TIMESTAMP (LocalDateTime without tz) variant not in the accepted set, or schema evolution changing the field's logical type; using a Beam logical type from a different SDK version whose identifier differs.","solutions":["Ensure the watermark column's Beam logical type is SqlTypes.TIMESTAMP or beam Timestamp logical type; convert the field's schema accordingly when building the output schema.","Pre-convert DATE/TIME or custom logical types to a supported timestamp logical type in an earlier PTransform.","Update or patch ApplyWatermarkColumn to handle the needed logical type and re-check with your Beam version.","Log/print field.getType() and its logical type identifier before applying the watermark transform to confirm what is actually arriving."],"exampleFix":"// before\nSchema.Field watermarkField = schema.getField(\"event_time\"); // DATE logical type\n// after\nSchema.Field watermarkField =\n    Schema.Field.of(\"event_time\", Schema.FieldType.logicalType(SqlTypes.TIMESTAMP));","handlingStrategy":"type-guard","validationCode":"Schema.LogicalTypeSupplier lt = schema.getField(col).getType().getLogicalType();\nif (lt == null || !(isSqlTimestamp(lt) || isBeamTimestamp(lt))) {\n  throw new IllegalArgumentException(\"Watermark column must use a timestamp logical type\");\n}","typeGuard":"static boolean isSupportedTimestamp(FieldType t) {\n  return t.getLogicalType() != null && (\n      SqlTypes.TIMESTAMP.getIdentifier().equals(t.getLogicalType().getIdentifier()) ||\n      org.apache.beam.sdk.schemas.logicaltypes.Timestamp.IDENTIFIER.equals(t.getLogicalType().getIdentifier()));\n}","tryCatchPattern":"try {\n  records.apply(ApplyWatermarkColumn.create(col));\n} catch (UnsupportedOperationException e) {\n  if (e.getMessage().startsWith(\"Unexpected logical type\")) {\n    throw new IllegalArgumentException(\"Coerce column to a supported timestamp logical type first\");\n  }\n  throw e;\n}","preventionTips":["Declare watermark columns with SqlTypes.TIMESTAMP logical types.","Coerce DATE/TIME/custom logical types upstream before the watermark transform.","Re-verify schema after any upstream schema evolution."],"tags":["java","beam","iceberg","cdc","watermark","logical-type"],"backgroundTag":"unsupported-operation","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"}