feat: Support reading Parquet ENUM logical type as String (#7805)

Closes #7723

## Background

Parquet's `ENUM` logical type is physically identical to `STRING`: both
annotate a `BINARY` column with UTF-8 encoded bytes. The only difference
is the label. External tools such as Spark and PyArrow use `ENUM` to
indicate a column holds a finite set of string values, but the wire
format is the same.

Deephaven's read pipeline has three stages where logical type is
dispatched. All three previously had no handling for
`EnumLogicalTypeAnnotation`, causing ENUM-annotated columns from
externally produced files to fail on read.

## Changes

### Stage 1 — Schema to Java type (`ParquetSchemaReader`)

**Before:** `visit(EnumLogicalTypeAnnotation)` set an error string and
returned `Optional.empty()`, so the column was unresolvable.

**After:** Returns `Optional.of(String.class)`, the same result as
`visit(StringLogicalTypeAnnotation)`.

### Stage 2 — Column data to chunk (`ParquetColumnLocation`)

**Before:** No `visit(EnumLogicalTypeAnnotation)` override existed, so
the visitor returned `Optional.empty()` and the read failed at runtime.

**After:** A new override delegates to `ToStringPage.create(...)`, the
same decoder used for `STRING` columns.

### Stage 3 — Pushdown statistics (`MinMaxFromStatistics`)

**Before:** `getMinMaxForStrings` only accepted
`StringLogicalTypeAnnotation`, so ENUM columns returned `false` and
forced a full scan on every filter.

**After:** The condition adds `|| instanceof EnumLogicalTypeAnnotation`,
enabling min/max pushdown for ENUM columns.

## Tests

- `MinMaxFromStatisticsTest.enumLogicalStatisticsAreMaterialised` — unit
test verifying ENUM statistics are extracted as strings.
- `ParquetTableReadWriteTest.testReadEnumLogicalTypeAsString` —
end-to-end test that writes a Parquet file with a `BINARY+ENUM` column
and reads it back, verifying the column materializes as `String` with
correct values.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: leochen4891

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/ParquetSchemaReader.java

@@ -361,22 +361,9 @@ private static LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<Class<?>> getV
         return new LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<Class<?>>() {
             @Override
             public Optional<Class<?>> visit(final LogicalTypeAnnotation.StringLogicalTypeAnnotation stringLogicalType) {
-                final ColumnDescriptor column = currentColumn.getValue();
-                final String columnName = column.getPath()[0];
-                final ColumnTypeInfo columnTypeInfo = nonDefaultTypeColumns.get(columnName);
-                final ColumnTypeInfo.SpecialType specialType =
-                        columnTypeInfo == null ? null : columnTypeInfo.specialType().orElse(null);
-                if (specialType != null) {
-                    if (specialType == ColumnTypeInfo.SpecialType.StringSet) {
-                        return Optional.of(StringSet.class);
-                    }
-                    if (specialType != ColumnTypeInfo.SpecialType.Vector) {
-                        throw new UncheckedDeephavenException("Type " + column.getPrimitiveType()
-                                + " for column " + Arrays.toString(column.getPath())
-                                + " with unknown or incompatible special type " + specialType);
-                    }
-                }
-                return Optional.of(String.class);
+                // Delegate to the shared helper so STRING and ENUM are resolved identically,
+                // including any Deephaven-specific SpecialType metadata (StringSet, Vector).
+                return visitStringLike(currentColumn.getValue(), nonDefaultTypeColumns);
             }
 
             @Override
@@ -393,8 +380,10 @@ public Optional<Class<?>> visit(final LogicalTypeAnnotation.ListLogicalTypeAnnot
 
             @Override
             public Optional<Class<?>> visit(final LogicalTypeAnnotation.EnumLogicalTypeAnnotation enumLogicalType) {
-                errorString.setValue("EnumLogicalType");
-                return Optional.empty();
+                // ENUM is physically identical to STRING (UTF-8 encoded BINARY). Delegate to the
+                // shared helper so any future Deephaven SpecialType metadata on ENUM columns is
+                // handled consistently with STRING columns.
+                return visitStringLike(currentColumn.getValue(), nonDefaultTypeColumns);
             }
 
             @Override
@@ -501,4 +490,30 @@ public Optional<Class<?>> visit(
             }
         };
     }
+
+    /**
+     * Shared resolution logic for string-like logical types (STRING and ENUM). Both are physically BINARY with UTF-8
+     * encoding, so they map to the same Java types. Centralising the logic here ensures that any Deephaven-specific
+     * {@link ColumnTypeInfo.SpecialType} metadata (e.g. {@code StringSet}, {@code Vector}) is respected for both
+     * annotations consistently, including any new special types added in the future.
+     */
+    private static Optional<Class<?>> visitStringLike(
+            final ColumnDescriptor column,
+            final Map<String, ColumnTypeInfo> nonDefaultTypeColumns) {
+        final String columnName = column.getPath()[0];
+        final ColumnTypeInfo columnTypeInfo = nonDefaultTypeColumns.get(columnName);
+        final ColumnTypeInfo.SpecialType specialType =
+                columnTypeInfo == null ? null : columnTypeInfo.specialType().orElse(null);
+        if (specialType != null) {
+            if (specialType == ColumnTypeInfo.SpecialType.StringSet) {
+                return Optional.of(StringSet.class);
+            }
+            if (specialType != ColumnTypeInfo.SpecialType.Vector) {
+                throw new UncheckedDeephavenException("Type " + column.getPrimitiveType()
+                        + " for column " + Arrays.toString(column.getPath())
+                        + " with unknown or incompatible special type " + specialType);
+            }
+        }
+        return Optional.of(String.class);
+    }
 }

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/location/MinMaxFromStatistics.java

@@ -392,7 +392,8 @@ static boolean getMinMaxForStrings(
             @NotNull final Consumer<String> maxSetter) {
         final PrimitiveType parquetColType = statistics.type();
         final LogicalTypeAnnotation logicalType = parquetColType.getLogicalTypeAnnotation();
-        if (logicalType instanceof LogicalTypeAnnotation.StringLogicalTypeAnnotation) {
+        if (logicalType instanceof LogicalTypeAnnotation.StringLogicalTypeAnnotation
+                || logicalType instanceof LogicalTypeAnnotation.EnumLogicalTypeAnnotation) {
             verifyPrimitive(statistics, PrimitiveType.PrimitiveTypeName.BINARY);
             final String minString = statistics.minAsString();
             final String maxString = statistics.maxAsString();

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/location/ParquetColumnLocation.java

@@ -535,6 +535,12 @@ private static class LogicalTypeVisitor<ATTR extends Any>
             return Optional.of(ToStringPage.create(pageType, columnChunkReader.getDictionarySupplier()));
         }
 
+        @Override
+        public Optional<ToPage<ATTR, ?>> visit(
+                final LogicalTypeAnnotation.EnumLogicalTypeAnnotation enumLogicalType) {
+            return Optional.of(ToStringPage.create(pageType, columnChunkReader.getDictionarySupplier()));
+        }
+
         @Override
         public Optional<ToPage<ATTR, ?>> visit(
                 final LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {

extensions/parquet/table/src/test/java/io/deephaven/parquet/table/ParquetTableReadWriteTest.java

@@ -50,8 +50,12 @@
 import io.deephaven.engine.util.file.TrackedFileHandleFactory;
 import io.deephaven.parquet.base.BigDecimalParquetBytesCodec;
 import io.deephaven.parquet.base.BigIntegerParquetBytesCodec;
+import io.deephaven.parquet.base.ColumnWriter;
 import io.deephaven.parquet.base.InvalidParquetFileException;
+import io.deephaven.parquet.base.NullParquetMetadataFileWriter;
 import io.deephaven.parquet.base.NullStatistics;
+import io.deephaven.parquet.base.ParquetFileWriter;
+import io.deephaven.parquet.base.RowGroupWriter;
 import io.deephaven.parquet.base.materializers.ParquetMaterializerUtils;
 import io.deephaven.parquet.table.location.ParquetTableLocation;
 import io.deephaven.parquet.table.location.ParquetTableLocationKey;
@@ -76,6 +80,7 @@
 import org.apache.commons.lang3.mutable.MutableDouble;
 import org.apache.commons.lang3.mutable.MutableFloat;
 import org.apache.commons.lang3.mutable.MutableObject;
+import org.apache.parquet.bytes.HeapByteBufferAllocator;
 import org.apache.parquet.column.Encoding;
 import org.apache.parquet.column.statistics.DoubleStatistics;
 import org.apache.parquet.column.statistics.IntStatistics;
@@ -133,6 +138,7 @@
 import static io.deephaven.util.QueryConstants.*;
 import static org.apache.parquet.schema.LogicalTypeAnnotation.decimalType;
 import static org.apache.parquet.schema.LogicalTypeAnnotation.intType;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
@@ -5002,6 +5008,31 @@ public void testSelectValidateDefinition() {
         assertTrue(result.getDefinition().getColumn("Double").isDirect());
     }
 
+    @Test
+    public void testReadEnumLogicalTypeAsString() throws IOException {
+        final MessageType schema = Types.buildMessage()
+                .required(BINARY).as(LogicalTypeAnnotation.enumType()).named("status")
+                .named("schema");
+        final File dest = new File(rootFile, "enum_logical_type.parquet");
+        final Binary[] values = {Binary.fromString("RED"), Binary.fromString("GREEN"), Binary.fromString("BLUE")};
+        final Statistics<?> stats = Statistics.createStats(schema.getType("status"));
+        for (final Binary v : values) {
+            stats.updateStats(v);
+        }
+        // ParquetFileWriter is AutoCloseable, so nest it in its own try-with-resources to
+        // guarantee the file is finalised even if an exception is thrown mid-write.
+        try (final java.io.OutputStream os = Files.newOutputStream(dest.toPath());
+                final ParquetFileWriter fileWriter = new ParquetFileWriter(dest.toURI(), os,
+                        ParquetInstructions.EMPTY.getTargetPageSize(), new HeapByteBufferAllocator(), schema,
+                        "UNCOMPRESSED", Collections.emptyMap(), NullParquetMetadataFileWriter.INSTANCE, true)) {
+            final RowGroupWriter rowGroupWriter = fileWriter.addRowGroup(values.length);
+            try (final ColumnWriter columnWriter = rowGroupWriter.addColumn("status")) {
+                columnWriter.addPageNoNulls(values, values.length, stats);
+            }
+        }
+        checkSingleTable(newTable(stringCol("status", "RED", "GREEN", "BLUE")), dest);
+    }
+
     private void assertTableStatistics(Table inputTable, File dest) {
         // Verify that the columns have the correct statistics.
         final ParquetMetadata metadata =

extensions/parquet/table/src/test/java/io/deephaven/parquet/table/location/MinMaxFromStatisticsTest.java

@@ -648,6 +648,21 @@ public void stringLogicalStatisticsAreMaterialised() {
                 max, "zzz");
     }
 
+    @Test
+    public void enumLogicalStatisticsAreMaterialised() {
+        final PrimitiveType colType = Types.required(PrimitiveType.PrimitiveTypeName.BINARY)
+                .as(LogicalTypeAnnotation.enumType())
+                .named("enumBinary");
+        final Statistics<?> stats = buildStats(
+                colType, "ALPHA".getBytes(StandardCharsets.UTF_8), "ZETA".getBytes(StandardCharsets.UTF_8), 0L);
+        final MutableObject<String> min = new MutableObject<>();
+        final MutableObject<String> max = new MutableObject<>();
+        assertMatches(
+                MinMaxFromStatistics.getMinMaxForStrings(stats, min::setValue, max::setValue),
+                min, "ALPHA",
+                max, "ZETA");
+    }
+
     @Test
     public void binaryWithoutStringLogicalTypeIsRejected() {
         final PrimitiveType colType = Types.required(PrimitiveType.PrimitiveTypeName.BINARY).named("rawBinary");