/*
 * 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.hive;

import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.TupleDomain;
import com.facebook.presto.spi.type.TypeManager;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableList;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe;
import org.joda.time.DateTimeZone;

import java.util.List;
import java.util.Optional;
import java.util.Properties;

import static com.facebook.presto.hive.HiveUtil.isDeserializerClass;
import static com.google.common.base.Predicates.not;
import static com.google.common.collect.Iterables.filter;

public class ParquetRecordCursorProvider
        implements HiveRecordCursorProvider
{
    @Override
    public Optional<HiveRecordCursor> createHiveRecordCursor(
            String clientId,
            Configuration configuration,
            ConnectorSession session,
            Path path,
            long start,
            long length,
            Properties schema,
            List<HiveColumnHandle> columns,
            List<HivePartitionKey> partitionKeys,
            TupleDomain<HiveColumnHandle> effectivePredicate,
            DateTimeZone hiveStorageTimeZone,
            TypeManager typeManager)
    {
        if (!isDeserializerClass(schema, ParquetHiveSerDe.class)) {
            return Optional.empty();
        }

        // are all columns supported by Parquet code
        List<HiveColumnHandle> unsupportedColumns = ImmutableList.copyOf(filter(columns, not(isParquetSupportedType())));
        if (!unsupportedColumns.isEmpty()) {
            throw new IllegalArgumentException("Can not read Parquet column: " + unsupportedColumns);
        }

        return Optional.<HiveRecordCursor>of(new ParquetHiveRecordCursor(
                configuration,
                path,
                start,
                length,
                schema,
                partitionKeys,
                columns,
                typeManager));
    }

    private static Predicate<HiveColumnHandle> isParquetSupportedType()
    {
        return new Predicate<HiveColumnHandle>()
        {
            @Override
            public boolean apply(HiveColumnHandle columnHandle)
            {
                HiveType hiveType = columnHandle.getHiveType();
                return hiveType != HiveType.HIVE_TIMESTAMP &&
                        hiveType != HiveType.HIVE_DATE &&
                        hiveType != HiveType.HIVE_BINARY;
            }
        };
    }
}

@@ -817,24 +817,71 @@
 utableList;%0A
+import com.google.common.collect.ImmutableSet;%0A
 import org.a
@@ -951,76 +951,8 @@
 th;%0A
-import org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe;%0A
 impo
@@ -1061,16 +1061,38 @@
 perties;
+%0Aimport java.util.Set;
 %0A%0Aimport
@@ -1133,18 +1133,19 @@
 iveUtil.
-is
+get
 Deserial
@@ -1153,16 +1153,20 @@
 zerClass
+Name
 ;%0Aimport
@@ -1358,16 +1358,271 @@
 vider%0A%7B%0A
+    private static final Set%3CString%3E PARQUET_SERDE_CLASS_NAMES = ImmutableSet.%3CString%3Ebuilder()%0A            .add(%22org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe%22)%0A            .add(%22parquet.hive.serde.ParquetHiveSerDe%22)%0A            .build();%0A%0A
     @Ove
@@ -2158,18 +2158,54 @@
    if (!
-is
+PARQUET_SERDE_CLASS_NAMES.contains(get
 Deserial
@@ -2217,39 +2217,20 @@
 lass
+Name
 (schema
-, ParquetHiveSerDe.class
+)
 )) %7B

Add support for various Parquet serde class names