Add ORC support for iceberg connector

pom.xml

@@ -608,7 +608,7 @@
             <dependency>
                 <groupId>com.facebook.presto.orc</groupId>
                 <artifactId>orc-protobuf</artifactId>
-                <version>11</version>
+                <version>12</version>
             </dependency>
 
             <dependency>

presto-hive-metastore/src/main/java/com/facebook/presto/hive/HiveType.java

@@ -194,7 +194,7 @@ public static List<HiveType> toHiveTypes(String hiveTypes)
                 .collect(toList()));
     }
 
-    private static HiveType toHiveType(TypeInfo typeInfo)
+    public static HiveType toHiveType(TypeInfo typeInfo)
     {
         requireNonNull(typeInfo, "typeInfo is null");
         return new HiveType(typeInfo);

presto-hive-metastore/src/main/java/com/facebook/presto/hive/metastore/file/FileHiveMetastore.java

@@ -432,7 +432,7 @@ public synchronized void replaceTable(MetastoreContext metastoreContext, String
         checkArgument(!newTable.getTableType().equals(TEMPORARY_TABLE), "temporary tables must never be stored in the metastore");
 
         Table table = getRequiredTable(metastoreContext, databaseName, tableName);
-        if (!table.getTableType().equals(VIRTUAL_VIEW) || !newTable.getTableType().equals(VIRTUAL_VIEW)) {
+        if ((!table.getTableType().equals(VIRTUAL_VIEW) || !newTable.getTableType().equals(VIRTUAL_VIEW)) && !isIcebergTable(table.getParameters())) {
             throw new PrestoException(HIVE_METASTORE_ERROR, "Only views can be updated with replaceTable");
         }
         if (!table.getDatabaseName().equals(databaseName) || !table.getTableName().equals(tableName)) {

presto-hive/src/main/java/com/facebook/presto/hive/OrcFileWriter.java

@@ -29,6 +29,7 @@
 import com.facebook.presto.orc.OrcWriterOptions;
 import com.facebook.presto.orc.WriterStats;
 import com.facebook.presto.orc.metadata.CompressionKind;
+import com.facebook.presto.orc.metadata.OrcType;
 import com.facebook.presto.spi.PrestoException;
 import com.google.common.collect.ImmutableList;
 import org.joda.time.DateTimeZone;
@@ -58,7 +59,7 @@ public class OrcFileWriter
     private static final int INSTANCE_SIZE = ClassLayout.parseClass(OrcFileWriter.class).instanceSize();
     private static final ThreadMXBean THREAD_MX_BEAN = ManagementFactory.getThreadMXBean();
 
-    private final OrcWriter orcWriter;
+    protected final OrcWriter orcWriter;
     private final Callable<Void> rollbackAction;
     private final int[] fileInputColumnIndexes;
     private final List<Block> nullBlocks;
@@ -83,6 +84,43 @@ public OrcFileWriter(
             WriterStats stats,
             DwrfEncryptionProvider dwrfEncryptionProvider,
             Optional<DwrfWriterEncryption> dwrfWriterEncryption)
+    {
+        this(
+                dataSink,
+                rollbackAction,
+                orcEncoding,
+                columnNames,
+                fileColumnTypes,
+                Optional.empty(),
+                compression,
+                options,
+                fileInputColumnIndexes,
+                metadata,
+                hiveStorageTimeZone,
+                validationInputFactory,
+                validationMode,
+                stats,
+                dwrfEncryptionProvider,
+                dwrfWriterEncryption);
+    }
+
+    public OrcFileWriter(
+            DataSink dataSink,
+            Callable<Void> rollbackAction,
+            OrcEncoding orcEncoding,
+            List<String> columnNames,
+            List<Type> fileColumnTypes,
+            Optional<List<OrcType>> fileColumnOrcTypes,
+            CompressionKind compression,
+            OrcWriterOptions options,
+            int[] fileInputColumnIndexes,
+            Map<String, String> metadata,
+            DateTimeZone hiveStorageTimeZone,
+            Optional<Supplier<OrcDataSource>> validationInputFactory,
+            OrcWriteValidationMode validationMode,
+            WriterStats stats,
+            DwrfEncryptionProvider dwrfEncryptionProvider,
+            Optional<DwrfWriterEncryption> dwrfWriterEncryption)
     {
         requireNonNull(dataSink, "dataSink is null");
 
@@ -91,6 +129,7 @@ public OrcFileWriter(
                     dataSink,
                     columnNames,
                     fileColumnTypes,
+                    fileColumnOrcTypes,
                     orcEncoding,
                     compression,
                     dwrfWriterEncryption,

presto-iceberg/pom.xml

@@ -21,11 +21,6 @@
             <artifactId>concurrent</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>com.facebook.presto</groupId>
-            <artifactId>presto-client</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>com.facebook.presto</groupId>
             <artifactId>presto-hive-common</artifactId>
@@ -104,6 +99,11 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>com.facebook.presto</groupId>
+            <artifactId>presto-orc</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>com.facebook.presto</groupId>
             <artifactId>presto-plugin-toolkit</artifactId>

presto-iceberg/src/main/java/com/facebook/presto/iceberg/FilesTable.java

@@ -0,0 +1,187 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.iceberg;
+
+import com.facebook.presto.common.Page;
+import com.facebook.presto.common.predicate.TupleDomain;
+import com.facebook.presto.common.type.ArrayType;
+import com.facebook.presto.common.type.StandardTypes;
+import com.facebook.presto.common.type.TypeManager;
+import com.facebook.presto.common.type.TypeSignatureParameter;
+import com.facebook.presto.iceberg.util.PageListBuilder;
+import com.facebook.presto.spi.ColumnMetadata;
+import com.facebook.presto.spi.ConnectorPageSource;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.ConnectorTableMetadata;
+import com.facebook.presto.spi.FixedPageSource;
+import com.facebook.presto.spi.SchemaTableName;
+import com.facebook.presto.spi.SystemTable;
+import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.airlift.slice.Slices;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.facebook.presto.common.type.BigintType.BIGINT;
+import static com.facebook.presto.common.type.IntegerType.INTEGER;
+import static com.facebook.presto.common.type.VarbinaryType.VARBINARY;
+import static com.facebook.presto.common.type.VarcharType.VARCHAR;
+import static com.facebook.presto.iceberg.IcebergUtil.getTableScan;
+import static com.facebook.presto.iceberg.util.PageListBuilder.forTable;
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static java.util.Objects.requireNonNull;
+
+public class FilesTable
+        implements SystemTable
+{
+    private final ConnectorTableMetadata tableMetadata;
+    private final Table icebergTable;
+    private final Optional<Long> snapshotId;
+
+    public FilesTable(SchemaTableName tableName, Table icebergTable, Optional<Long> snapshotId, TypeManager typeManager)
+    {
+        this.icebergTable = requireNonNull(icebergTable, "icebergTable is null");
+
+        tableMetadata = new ConnectorTableMetadata(requireNonNull(tableName, "tableName is null"),
+                ImmutableList.<ColumnMetadata>builder()
+                        .add(new ColumnMetadata("file_path", VARCHAR))
+                        .add(new ColumnMetadata("file_format", VARCHAR))
+                        .add(new ColumnMetadata("record_count", BIGINT))
+                        .add(new ColumnMetadata("file_size_in_bytes", BIGINT))
+                        .add(new ColumnMetadata("column_sizes", typeManager.getParameterizedType(StandardTypes.MAP, ImmutableList.of(
+                                TypeSignatureParameter.of(INTEGER.getTypeSignature()),
+                                TypeSignatureParameter.of(BIGINT.getTypeSignature())))))
+                        .add(new ColumnMetadata("value_counts", typeManager.getParameterizedType(StandardTypes.MAP, ImmutableList.of(
+                                TypeSignatureParameter.of(INTEGER.getTypeSignature()),
+                                TypeSignatureParameter.of(BIGINT.getTypeSignature())))))
+                        .add(new ColumnMetadata("null_value_counts", typeManager.getParameterizedType(StandardTypes.MAP, ImmutableList.of(
+                                TypeSignatureParameter.of(INTEGER.getTypeSignature()),
+                                TypeSignatureParameter.of(BIGINT.getTypeSignature())))))
+                        .add(new ColumnMetadata("lower_bounds", typeManager.getParameterizedType(StandardTypes.MAP, ImmutableList.of(
+                                TypeSignatureParameter.of(INTEGER.getTypeSignature()),
+                                TypeSignatureParameter.of(VARCHAR.getTypeSignature())))))
+                        .add(new ColumnMetadata("upper_bounds", typeManager.getParameterizedType(StandardTypes.MAP, ImmutableList.of(
+                                TypeSignatureParameter.of(INTEGER.getTypeSignature()),
+                                TypeSignatureParameter.of(VARCHAR.getTypeSignature())))))
+                        .add(new ColumnMetadata("key_metadata", VARBINARY))
+                        .add(new ColumnMetadata("split_offsets", new ArrayType(BIGINT)))
+                        .build());
+        this.snapshotId = requireNonNull(snapshotId, "snapshotId is null");
+    }
+
+    @Override
+    public Distribution getDistribution()
+    {
+        return Distribution.SINGLE_COORDINATOR;
+    }
+
+    @Override
+    public ConnectorTableMetadata getTableMetadata()
+    {
+        return tableMetadata;
+    }
+
+    @Override
+    public ConnectorPageSource pageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
+    {
+        return new FixedPageSource(buildPages(tableMetadata, session, icebergTable, snapshotId));
+    }
+
+    private static List<Page> buildPages(ConnectorTableMetadata tableMetadata, ConnectorSession session, Table icebergTable, Optional<Long> snapshotId)
+    {
+        PageListBuilder pagesBuilder = forTable(tableMetadata);
+        TableScan tableScan = getTableScan(TupleDomain.all(), snapshotId, icebergTable).includeColumnStats();
+        Map<Integer, Type> idToTypeMap = getIdToTypeMap(icebergTable.schema());
+
+        tableScan.planFiles().forEach(fileScanTask -> {
+            DataFile dataFile = fileScanTask.file();
+            pagesBuilder.beginRow();
+            pagesBuilder.appendVarchar(dataFile.path().toString());
+            pagesBuilder.appendVarchar(dataFile.format().name());
+            pagesBuilder.appendBigint(dataFile.recordCount());
+            pagesBuilder.appendBigint(dataFile.fileSizeInBytes());
+            if (checkNonNull(dataFile.columnSizes(), pagesBuilder)) {
+                pagesBuilder.appendIntegerBigintMap(dataFile.columnSizes());
+            }
+            if (checkNonNull(dataFile.valueCounts(), pagesBuilder)) {
+                pagesBuilder.appendIntegerBigintMap(dataFile.valueCounts());
+            }
+            if (checkNonNull(dataFile.nullValueCounts(), pagesBuilder)) {
+                pagesBuilder.appendIntegerBigintMap(dataFile.nullValueCounts());
+            }
+            if (checkNonNull(dataFile.lowerBounds(), pagesBuilder)) {
+                pagesBuilder.appendIntegerVarcharMap(dataFile.lowerBounds().entrySet().stream()
+                        .collect(toImmutableMap(
+                                Map.Entry<Integer, ByteBuffer>::getKey,
+                                entry -> Transforms.identity(idToTypeMap.get(entry.getKey())).toHumanString(
+                                        Conversions.fromByteBuffer(idToTypeMap.get(entry.getKey()), entry.getValue())))));
+            }
+            if (checkNonNull(dataFile.upperBounds(), pagesBuilder)) {
+                pagesBuilder.appendIntegerVarcharMap(dataFile.upperBounds().entrySet().stream()
+                        .collect(toImmutableMap(
+                                Map.Entry<Integer, ByteBuffer>::getKey,
+                                entry -> Transforms.identity(idToTypeMap.get(entry.getKey())).toHumanString(
+                                        Conversions.fromByteBuffer(idToTypeMap.get(entry.getKey()), entry.getValue())))));
+            }
+            if (checkNonNull(dataFile.keyMetadata(), pagesBuilder)) {
+                pagesBuilder.appendVarbinary(Slices.wrappedBuffer(dataFile.keyMetadata()));
+            }
+            if (checkNonNull(dataFile.splitOffsets(), pagesBuilder)) {
+                pagesBuilder.appendBigintArray(dataFile.splitOffsets());
+            }
+            pagesBuilder.endRow();
+        });
+
+        return pagesBuilder.build();
+    }
+
+    private static Map<Integer, Type> getIdToTypeMap(Schema schema)
+    {
+        ImmutableMap.Builder<Integer, Type> idToTypeMap = ImmutableMap.builder();
+        for (Types.NestedField field : schema.columns()) {
+            populateIdToTypeMap(field, idToTypeMap);
+        }
+        return idToTypeMap.build();
+    }
+
+    private static void populateIdToTypeMap(Types.NestedField field, ImmutableMap.Builder<Integer, Type> idToTypeMap)
+    {
+        Type type = field.type();
+        idToTypeMap.put(field.fieldId(), type);
+        if (type instanceof Type.NestedType) {
+            type.asNestedType().fields().forEach(child -> populateIdToTypeMap(child, idToTypeMap));
+        }
+    }
+
+    private static boolean checkNonNull(Object object, PageListBuilder pagesBuilder)
+    {
+        if (object == null) {
+            pagesBuilder.appendNull();
+            return false;
+        }
+        return true;
+    }
+}