@Override
    public ConnectorPageSource createPageSource(
            ConnectorTransactionHandle transaction,
            ConnectorSession session,
            ConnectorSplit split,
            ConnectorTableLayoutHandle layout,
            List<ColumnHandle> columns,
            SplitContext splitContext,
            RuntimeStats runtimeStats)
    {
        HiveTableLayoutHandle hiveLayout = (HiveTableLayoutHandle) layout;

        List<HiveColumnHandle> selectedColumns = columns.stream()
                .map(HiveColumnHandle.class::cast)
                .collect(toList());

        HiveSplit hiveSplit = (HiveSplit) split;
        Path path = new Path(hiveSplit.getFileSplit().getPath());

        Configuration configuration = hdfsEnvironment.getConfiguration(
                new HdfsContext(
                        session,
                        hiveSplit.getDatabase(),
                        hiveSplit.getTable(),
                        hiveLayout.getTablePath(),
                        false),
                path);

        Optional<EncryptionInformation> encryptionInformation = hiveSplit.getEncryptionInformation();
        CacheQuota cacheQuota = generateCacheQuota(hiveSplit);
        HiveFileContext fileContext = new HiveFileContext(
                splitContext.isCacheable(),
                cacheQuota,
                hiveSplit.getFileSplit().getExtraFileInfo().map(BinaryExtraHiveFileInfo::new),
                OptionalLong.of(hiveSplit.getFileSplit().getFileSize()),
                OptionalLong.of(hiveSplit.getFileSplit().getStart()),
                OptionalLong.of(hiveSplit.getFileSplit().getLength()),
                hiveSplit.getFileSplit().getFileModifiedTime(),
                HiveSessionProperties.isVerboseRuntimeStatsEnabled(session),
                runtimeStats);

        if (columns.stream().anyMatch(columnHandle -> ((HiveColumnHandle) columnHandle).getColumnType().equals(AGGREGATED))) {
            checkArgument(columns.stream().allMatch(columnHandle -> ((HiveColumnHandle) columnHandle).getColumnType().equals(AGGREGATED)), "Not all columns are of 'AGGREGATED' type");

            if (hiveLayout.isFooterStatsUnreliable()) {
                throw new PrestoException(HIVE_UNSUPPORTED_FORMAT, format("Partial aggregation pushdown is not supported when footer stats are unreliable. " +
                                "Table %s has file %s with unreliable footer stats. " +
                                "Set session property [catalog-name].pushdown_partial_aggregations_into_scan=false and execute query again.",
                        hiveLayout.getSchemaTableName(),
                        hiveSplit.getFileSplit().getPath()));
            }

            return createAggregatedPageSource(aggregatedPageSourceFactories, configuration, session, hiveSplit, hiveLayout, selectedColumns, fileContext, encryptionInformation);
        }
        if (hiveLayout.isPushdownFilterEnabled()) {
            Optional<ConnectorPageSource> selectivePageSource = createSelectivePageSource(
                    selectivePageSourceFactories,
                    configuration,
                    session,
                    hiveSplit,
                    hiveLayout,
                    selectedColumns,
                    hiveStorageTimeZone,
                    typeManager,
                    optimizedRowExpressionCache,
                    splitContext,
                    fileContext,
                    encryptionInformation);
            if (selectivePageSource.isPresent()) {
                return selectivePageSource.get();
            }
        }

        TupleDomain<HiveColumnHandle> effectivePredicate = hiveLayout.getDomainPredicate()
                .transform(Subfield::getRootName)
                .transform(hiveLayout.getPredicateColumns()::get);

        if (shouldSkipBucket(hiveLayout, hiveSplit, splitContext, isLegacyTimestampBucketing(session))) {
            return new HiveEmptySplitPageSource();
        }

        if (shouldSkipPartition(typeManager, hiveLayout, hiveStorageTimeZone, hiveSplit, splitContext)) {
            return new HiveEmptySplitPageSource();
        }

        Optional<ConnectorPageSource> pageSource = createHivePageSource(
                cursorProviders,
                pageSourceFactories,
                configuration,
                session,
                hiveSplit.getFileSplit(),
                hiveSplit.getTableBucketNumber(),
                hiveSplit.getStorage(),
                splitContext.getDynamicFilterPredicate().map(filter -> filter.transform(handle -> (HiveColumnHandle) handle).intersect(effectivePredicate)).orElse(effectivePredicate),
                selectedColumns,
                hiveLayout.getPredicateColumns(),
                hiveSplit.getPartitionKeys(),
                hiveStorageTimeZone,
                typeManager,
                hiveLayout.getSchemaTableName(),
                hiveLayout.getPartitionColumns().stream().map(HiveColumnHandle.class::cast).collect(toList()),
                hiveLayout.getDataColumns(),
                hiveLayout.getTableParameters(),
                hiveSplit.getPartitionDataColumnCount(),
                hiveSplit.getTableToPartitionMapping(),
                hiveSplit.getBucketConversion(),
                hiveSplit.isS3SelectPushdownEnabled(),
                fileContext,
                hiveLayout.getRemainingPredicate(),
                hiveLayout.isPushdownFilterEnabled(),
                rowExpressionService,
                encryptionInformation,
                hiveSplit.getRowIdPartitionComponent());
        if (pageSource.isPresent()) {
            return pageSource.get();
        }
        throw new IllegalStateException("Could not find a file reader for split " + hiveSplit);
    }
@Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Not all columns are of 'AGGREGATED' type")
    public void testFailsWhenMixOfAggregatedAndRegularColumns()
    {
        HivePageSourceProvider pageSourceProvider = createPageSourceProvider();
        pageSourceProvider.createPageSource(
                new HiveTransactionHandle(),
                SESSION,
                getHiveSplit(ORC),
                getHiveTableLayout(false, true, false),
                ImmutableList.of(LONG_COLUMN, LONG_AGGREGATED_COLUMN),
                new SplitContext(false),
                new RuntimeStats());
    }