{"record":{"id":"9ecd903bf3060399","repo":"apache/iceberg","slug":"cannot-convert-unknown-type-to-flink-s","errorCode":null,"errorMessage":"Cannot convert unknown type to Flink: %s","messagePattern":"Cannot convert unknown type to Flink: (.+?)","errorType":"exception","errorClass":"java.lang.UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/TypeToFlinkType.java","lineNumber":148,"sourceCode":"        } else {\n          // NANOS\n          return new TimestampType(9);\n        }\n      case STRING:\n        return new VarCharType(VarCharType.MAX_LENGTH);\n      case UUID:\n        // UUID length is 16\n        return new BinaryType(16);\n      case FIXED:\n        Types.FixedType fixedType = (Types.FixedType) primitive;\n        return new BinaryType(fixedType.length());\n      case BINARY:\n        return new VarBinaryType(VarBinaryType.MAX_LENGTH);\n      case DECIMAL:\n        Types.DecimalType decimal = (Types.DecimalType) primitive;\n        return new DecimalType(decimal.precision(), decimal.scale());\n      default:\n        throw new UnsupportedOperationException(\n            \"Cannot convert unknown type to Flink: \" + primitive);\n    }\n  }\n}\n","sourceCodeStart":130,"sourceCodeEnd":153,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/TypeToFlinkType.java#L130-L153","documentation":"TypeToFlinkType.primitive maps Iceberg primitive Types to Flink DataTypes; an unrecognized primitive hits the default branch and throws this message with the type's toString. It means an Iceberg type reached the converter that this Iceberg-Flink version does not know how to map (often a newer type than the runtime supports).","triggerScenarios":"Converting an Iceberg schema to Flink types where a column's primitive type is missing from the switch (e.g. Types.TimestampNanoType/unknown Variant on older runtimes) — via FlinkSchemaUtil.convert(schema) or table schema-to-Flink conversion at scan/sink setup.","commonSituations":"Iceberg tables written by newer engines with nano-timestamp or unknown future types being read by an older iceberg-flink runtime; mixed Iceberg versions on the classpath.","solutions":["Upgrade the iceberg-flink runtime to a version that maps the offending primitive type","Migrate the column to a supported type (e.g. timestamp -> micros timestamp) with add/update-schema rewrite","Check for mixed Iceberg jar versions on the classpath and align them"],"exampleFix":"// before\nTable column: timestamp_ns with old iceberg-flink runtime -> throws\n// after\nUpgrade iceberg-flink runtime, or rewrite column to Types.TimestampType (micros)","handlingStrategy":"validation","validationCode":"for (Types.NestedField f : schema.columns()) {\n  if (f.type().isPrimitiveType()) {\n    Type.PrimitiveType p = f.type().asPrimitiveType();\n    // e.g. reject TimestampNanoType / unknown types before conversion\n  }\n}","typeGuard":"boolean isSupportedIcebergPrimitive(Type t) {\n  return switch (t.typeId()) {\n    case BOOLEAN, INT, LONG, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP,\n         STRING, UUID, FIXED, BINARY, DECIMAL -> true;\n    default -> false;\n  };\n}","tryCatchPattern":"try {\n  DataType dt = TypeToFlinkType.toFlinkType(primitive);\n} catch (UnsupportedOperationException e) {\n  throw new IllegalStateException(\"Upgrade iceberg-flink or migrate column: \" + e.getMessage(), e);\n}","preventionTips":["Align Iceberg runtime versions across writers and the Flink job","Avoid nano-timestamp/new types until the runtime supports them","Run schema compatibility checks in CI for cross-engine tables"],"tags":["flink","type-conversion","version-mismatch"],"backgroundTag":"incompatible-source-type","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}