{"record":{"id":"704c97886e6c55f4","repo":"apache/iceberg","slug":"unsupported-logical-type-primitive-getoriginalt","errorCode":null,"errorMessage":"Unsupported logical type: ${primitive.getOriginalType()}","messagePattern":"Unsupported logical type: (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java","lineNumber":171,"sourceCode":"    }\n\n    private ParquetValueWriter<?> newOption(Type fieldType, ParquetValueWriter<?> writer) {\n      int maxD = type.getMaxDefinitionLevel(path(fieldType.getName()));\n      return ParquetValueWriters.option(fieldType, maxD, writer);\n    }\n\n    @Override\n    public ParquetValueWriter<?> primitive(LogicalType fType, PrimitiveType primitive) {\n      ColumnDescriptor desc = type.getColumnDescription(currentPath());\n\n      LogicalTypeAnnotation annotation = primitive.getLogicalTypeAnnotation();\n      if (annotation != null) {\n        Optional<ParquetValueWriter<?>> writer =\n            annotation.accept(new LogicalTypeWriterBuilder(fType, desc));\n        if (writer.isPresent()) {\n          return writer.get();\n        } else {\n          throw new UnsupportedOperationException(\n              \"Unsupported logical type: \" + primitive.getOriginalType());\n        }\n      }\n\n      switch (primitive.getPrimitiveTypeName()) {\n        case FIXED_LEN_BYTE_ARRAY:\n        case BINARY:\n          return byteArrays(desc);\n        case BOOLEAN:\n          return ParquetValueWriters.booleans(desc);\n        case INT32:\n          return ints(fType, desc);\n        case INT64:\n          return ParquetValueWriters.longs(desc);\n        case FLOAT:\n          return ParquetValueWriters.floats(desc);\n        case DOUBLE:\n          return ParquetValueWriters.doubles(desc);","sourceCodeStart":153,"sourceCodeEnd":189,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java#L153-L189","documentation":"FlinkParquetWriters.primitive() builds a Parquet value writer for a primitive column. When the column carries a logical type annotation, it delegates to LogicalTypeWriterBuilder; if that visitor returns no writer for the annotation, this UnsupportedOperationException is thrown. It means the Parquet file's logical type is not one Iceberg's Flink writer can map to a Flink data type.","triggerScenarios":"Writing Parquet via FlinkParquetWriters when a primitive column has a logical type annotation (e.g. an unusual or newer logical type like INTERVAL or a map/list annotation variant) that LogicalTypeWriterBuilder.visit methods do not handle, so Optional.empty() is returned.","commonSituations":"Reading/writing Parquet files produced by third-party tools with exotic logical types; a Parquet-runtime upgrade introducing new LogicalTypeAnnotation subclasses not yet handled; mismatched parquet-column versions on the classpath.","solutions":["Identify the offending column's logical type from the message and rewrite the data with a supported logical type (int32/int64/binary/decimal/timestamp/uuid).","Align the parquet-column dependency version with the one required by the Iceberg Flink runtime to ensure all standard logical types are handled.","If the type is genuinely unsupported, project/drop the column before writing, or add a case to LogicalTypeWriterBuilder for the annotation.","Check for duplicate/old parquet jars shading conflicting LogicalTypeAnnotation classes into the Flink job classpath."],"exampleFix":"// before: writing a column annotated with an unsupported logical type\nschema.add(\"weird\", Types.IntegerType.get()); // parquet column has INTERVAL annotation\n// after: convert data to a supported Iceberg type before writing\nschema.add(\"weird\", Types.StringType.get()); // or cast the source column to a supported type","handlingStrategy":"validation","validationCode":"MessageType schema = parquetFileSchema();\nschema.getFields().forEach(f -> {\n  if (f.isPrimitive() && ((PrimitiveType) f).getLogicalTypeAnnotation() != null) {\n    // ensure annotation is one of: decimal, timestamp(micros/nanos), date, string,\n    // enum, uuid, time(micros/nanos), int(bitWidth, signed)\n  }\n});","typeGuard":null,"tryCatchPattern":"try {\n  writer = FlinkParquetWriters.writeBuilder(fileType, schema).build();\n} catch (UnsupportedOperationException e) {\n  throw new IllegalStateException(\"Parquet logical type not supported: \" + e.getMessage(), e);\n}","preventionTips":["Prefer Iceberg-native writes so logical annotations are always Iceberg-generated","Inspect third-party Parquet schemas with parquet-tools before ingesting","Keep parquet-column dependencies aligned with the Iceberg Flink runtime version"],"tags":["flink","parquet","unsupported-type","logical-type"],"backgroundTag":"unsupported-enum-value","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"}