Refactoring the ParquetTools read/write APIs

extensions/parquet/base/src/main/java/io/deephaven/parquet/base/ParquetFileReader.java

@@ -3,12 +3,17 @@
 //
 package io.deephaven.parquet.base;
 
+import io.deephaven.UncheckedDeephavenException;
+import io.deephaven.util.channel.CachedChannelProvider;
 import io.deephaven.util.channel.SeekableChannelContext;
 import io.deephaven.util.channel.SeekableChannelsProvider;
+import io.deephaven.util.channel.SeekableChannelsProviderLoader;
 import org.apache.parquet.format.*;
 import org.apache.parquet.format.ColumnOrder;
 import org.apache.parquet.format.Type;
 import org.apache.parquet.schema.*;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 
 import java.io.File;
 import java.io.IOException;
@@ -37,6 +42,76 @@ public class ParquetFileReader {
     private final URI rootURI;
     private final MessageType type;
 
+    /**
+     * Make a {@link ParquetFileReader} for the supplied {@link File}. Wraps {@link IOException} as
+     * {@link UncheckedDeephavenException}.
+     *
+     * @param parquetFile The parquet file or the parquet metadata file
+     * @param specialInstructions Optional read instructions to pass to {@link SeekableChannelsProvider} while creating
+     *        channels
+     * @return The new {@link ParquetFileReader}
+     */
+    public static ParquetFileReader create(
+            @NotNull final File parquetFile,
+            @Nullable final Object specialInstructions) {
+        try {
+            return createChecked(parquetFile, specialInstructions);
+        } catch (IOException e) {
+            throw new UncheckedDeephavenException("Failed to create Parquet file reader: " + parquetFile, e);
+        }
+    }
+
+    /**
+     * Make a {@link ParquetFileReader} for the supplied {@link URI}. Wraps {@link IOException} as
+     * {@link UncheckedDeephavenException}.
+     *
+     * @param parquetFileURI The URI for the parquet file or the parquet metadata file
+     * @param specialInstructions Optional read instructions to pass to {@link SeekableChannelsProvider} while creating
+     *        channels
+     * @return The new {@link ParquetFileReader}
+     */
+    public static ParquetFileReader create(
+            @NotNull final URI parquetFileURI,
+            @Nullable final Object specialInstructions) {
+        try {
+            return createChecked(parquetFileURI, specialInstructions);
+        } catch (IOException e) {
+            throw new UncheckedDeephavenException("Failed to create Parquet file reader: " + parquetFileURI, e);
+        }
+    }
+
+    /**
+     * Make a {@link ParquetFileReader} for the supplied {@link File}.
+     *
+     * @param parquetFile The parquet file or the parquet metadata file
+     * @param specialInstructions Optional read instructions to pass to {@link SeekableChannelsProvider} while creating
+     *        channels
+     * @return The new {@link ParquetFileReader}
+     * @throws IOException if an IO exception occurs
+     */
+    public static ParquetFileReader createChecked(
+            @NotNull final File parquetFile,
+            @Nullable final Object specialInstructions) throws IOException {
+        return createChecked(convertToURI(parquetFile, false), specialInstructions);
+    }
+
+    /**
+     * Make a {@link ParquetFileReader} for the supplied {@link URI}.
+     *
+     * @param parquetFileURI The URI for the parquet file or the parquet metadata file
+     * @param specialInstructions Optional read instructions to pass to {@link SeekableChannelsProvider} while creating
+     *        channels
+     * @return The new {@link ParquetFileReader}
+     * @throws IOException if an IO exception occurs
+     */
+    public static ParquetFileReader createChecked(
+            @NotNull final URI parquetFileURI,
+            @Nullable final Object specialInstructions) throws IOException {
+        final SeekableChannelsProvider provider = SeekableChannelsProviderLoader.getInstance().fromServiceLoader(
+                parquetFileURI, specialInstructions);
+        return new ParquetFileReader(parquetFileURI, new CachedChannelProvider(provider, 1 << 7));
+    }
+
     /**
      * Create a new ParquetFileReader for the provided source.
      *

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

@@ -4,12 +4,25 @@
 package io.deephaven.parquet.table;
 
 import io.deephaven.UncheckedDeephavenException;
+import io.deephaven.api.util.NameValidator;
+import io.deephaven.base.ClassUtil;
+import io.deephaven.base.Pair;
+import io.deephaven.engine.table.ColumnDefinition;
 import io.deephaven.stringset.StringSet;
 import io.deephaven.engine.table.impl.locations.TableDataException;
 import io.deephaven.parquet.table.metadata.CodecInfo;
 import io.deephaven.parquet.table.metadata.ColumnTypeInfo;
 import io.deephaven.parquet.table.metadata.TableInfo;
 import io.deephaven.parquet.base.ParquetFileReader;
+import io.deephaven.util.SimpleTypeMap;
+import io.deephaven.vector.ByteVector;
+import io.deephaven.vector.CharVector;
+import io.deephaven.vector.DoubleVector;
+import io.deephaven.vector.FloatVector;
+import io.deephaven.vector.IntVector;
+import io.deephaven.vector.LongVector;
+import io.deephaven.vector.ObjectVector;
+import io.deephaven.vector.ShortVector;
 import org.apache.parquet.format.converter.ParquetMetadataConverter;
 import io.deephaven.util.codec.SimpleByteArrayCodec;
 import io.deephaven.util.codec.UTF8StringAsByteArrayCodec;
@@ -99,7 +112,7 @@ public static ParquetInstructions readParquetSchema(
             @NotNull final ColumnDefinitionConsumer consumer,
             @NotNull final BiFunction<String, Set<String>, String> legalizeColumnNameFunc) throws IOException {
         final ParquetFileReader parquetFileReader =
-                ParquetTools.getParquetFileReaderChecked(new File(filePath), readInstructions);
+                ParquetFileReader.createChecked(new File(filePath), readInstructions.getSpecialInstructions());
         final ParquetMetadata parquetMetadata =
                 new ParquetMetadataConverter().fromParquetMetadata(parquetFileReader.fileMetaData);
         return readParquetSchema(parquetFileReader.getSchema(), parquetMetadata.getFileMetaData().getKeyValueMetaData(),
@@ -282,6 +295,92 @@ public static ParquetInstructions readParquetSchema(
                 : instructionsBuilder.getValue().build();
     }
 
+    /**
+     * Convert schema information from a {@link ParquetMetadata} into {@link ColumnDefinition ColumnDefinitions}.
+     *
+     * @param schema Parquet schema. DO NOT RELY ON {@link ParquetMetadataConverter} FOR THIS! USE
+     *        {@link ParquetFileReader}!
+     * @param keyValueMetadata Parquet key-value metadata map
+     * @param readInstructionsIn Input conversion {@link ParquetInstructions}
+     * @return A {@link Pair} with {@link ColumnDefinition ColumnDefinitions} and adjusted {@link ParquetInstructions}
+     */
+    public static Pair<List<ColumnDefinition<?>>, ParquetInstructions> convertSchema(
+            @NotNull final MessageType schema,
+            @NotNull final Map<String, String> keyValueMetadata,
+            @NotNull final ParquetInstructions readInstructionsIn) {
+        final ArrayList<ColumnDefinition<?>> cols = new ArrayList<>();
+        final ParquetSchemaReader.ColumnDefinitionConsumer colConsumer = makeSchemaReaderConsumer(cols);
+        return new Pair<>(cols, ParquetSchemaReader.readParquetSchema(
+                schema,
+                keyValueMetadata,
+                readInstructionsIn,
+                colConsumer,
+                (final String colName, final Set<String> takenNames) -> NameValidator.legalizeColumnName(colName,
+                        s -> s.replace(" ", "_"), takenNames)));
+    }
+
+    private static ParquetSchemaReader.ColumnDefinitionConsumer makeSchemaReaderConsumer(
+            final ArrayList<ColumnDefinition<?>> colsOut) {
+        return (final ParquetSchemaReader.ParquetMessageDefinition parquetColDef) -> {
+            Class<?> baseType;
+            if (parquetColDef.baseType == boolean.class) {
+                baseType = Boolean.class;
+            } else {
+                baseType = parquetColDef.baseType;
+            }
+            ColumnDefinition<?> colDef;
+            if (parquetColDef.codecType != null && !parquetColDef.codecType.isEmpty()) {
+                final Class<?> componentType =
+                        (parquetColDef.codecComponentType != null && !parquetColDef.codecComponentType.isEmpty())
+                                ? loadClass(parquetColDef.name, "codecComponentType", parquetColDef.codecComponentType)
+                                : null;
+                final Class<?> dataType = loadClass(parquetColDef.name, "codecType", parquetColDef.codecType);
+                colDef = ColumnDefinition.fromGenericType(parquetColDef.name, dataType, componentType);
+            } else if (parquetColDef.dhSpecialType != null) {
+                if (parquetColDef.dhSpecialType == ColumnTypeInfo.SpecialType.StringSet) {
+                    colDef = ColumnDefinition.fromGenericType(parquetColDef.name, StringSet.class, null);
+                } else if (parquetColDef.dhSpecialType == ColumnTypeInfo.SpecialType.Vector) {
+                    final Class<?> vectorType = VECTOR_TYPE_MAP.get(baseType);
+                    if (vectorType != null) {
+                        colDef = ColumnDefinition.fromGenericType(parquetColDef.name, vectorType, baseType);
+                    } else {
+                        colDef = ColumnDefinition.fromGenericType(parquetColDef.name, ObjectVector.class, baseType);
+                    }
+                } else {
+                    throw new UncheckedDeephavenException("Unhandled dbSpecialType=" + parquetColDef.dhSpecialType);
+                }
+            } else {
+                if (parquetColDef.isArray) {
+                    if (baseType == byte.class && parquetColDef.noLogicalType) {
+                        colDef = ColumnDefinition.fromGenericType(parquetColDef.name, byte[].class, byte.class);
+                    } else {
+                        // TODO: ParquetInstruction.loadAsVector
+                        final Class<?> componentType = baseType;
+                        // On Java 12, replace by: dataType = componentType.arrayType();
+                        final Class<?> dataType = java.lang.reflect.Array.newInstance(componentType, 0).getClass();
+                        colDef = ColumnDefinition.fromGenericType(parquetColDef.name, dataType, componentType);
+                    }
+                } else {
+                    colDef = ColumnDefinition.fromGenericType(parquetColDef.name, baseType, null);
+                }
+            }
+            colsOut.add(colDef);
+        };
+    }
+
+    private static final SimpleTypeMap<Class<?>> VECTOR_TYPE_MAP = SimpleTypeMap.create(
+            null, CharVector.class, ByteVector.class, ShortVector.class, IntVector.class, LongVector.class,
+            FloatVector.class, DoubleVector.class, ObjectVector.class);
+
+    private static Class<?> loadClass(final String colName, final String desc, final String className) {
+        try {
+            return ClassUtil.lookupClass(className);
+        } catch (ClassNotFoundException e) {
+            throw new UncheckedDeephavenException(
+                    "Column " + colName + " with " + desc + "=" + className + " that can't be found in classloader");
+        }
+    }
+
     private static LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<Class<?>> getVisitor(
             final Map<String, ColumnTypeInfo> nonDefaultTypeColumns,
             final MutableObject<String> errorString,