@Override
    public ConnectorSplitSource getSplits(
            ConnectorTransactionHandle transaction,
            ConnectorSession session,
            ConnectorTableLayoutHandle layoutHandle,
            SplitSchedulingContext splitSchedulingContext)
    {
        HiveTableLayoutHandle layout = (HiveTableLayoutHandle) layoutHandle;
        SchemaTableName tableName = layout.getSchemaTableName();

        // get table metadata
        TransactionalMetadata metadata = hiveTransactionManager.get(transaction);
        if (metadata == null) {
            throw new PrestoException(HIVE_TRANSACTION_NOT_FOUND, format("Transaction not found: %s", transaction));
        }
        SemiTransactionalHiveMetastore metastore = metadata.getMetastore();
        MetastoreContext metastoreContext = new MetastoreContext(
                session.getIdentity(),
                session.getQueryId(),
                session.getClientInfo(),
                session.getClientTags(),
                session.getSource(),
                getMetastoreHeaders(session),
                isUserDefinedTypeEncodingEnabled(session),
                metastore.getColumnConverterProvider(),
                session.getWarningCollector(),
                session.getRuntimeStats());
        Table table = layout.getTable(metastore, metastoreContext);

        if (!isOfflineDataDebugModeEnabled(session)) {
            // verify table is not marked as non-readable
            String tableNotReadable = table.getParameters().get(OBJECT_NOT_READABLE);
            if (!isNullOrEmpty(tableNotReadable)) {
                throw new HiveNotReadableException(tableName, Optional.empty(), tableNotReadable);
            }
        }

        // get partitions
        List<HivePartition> partitions = layout.getPartitions()
                .orElseThrow(() -> new PrestoException(GENERIC_INTERNAL_ERROR, "Layout does not contain partitions"));

        // short circuit if we don't have any partitions
        HivePartition partition = Iterables.getFirst(partitions, null);
        if (partition == null) {
            return new FixedSplitSource(ImmutableList.of());
        }

        Optional<HiveBucketFilter> bucketFilter = layout.getBucketFilter();

        // validate bucket bucketed execution
        Optional<HiveBucketHandle> bucketHandle = layout.getBucketHandle();
        if ((splitSchedulingContext.getSplitSchedulingStrategy() == GROUPED_SCHEDULING) && !bucketHandle.isPresent()) {
            throw new PrestoException(GENERIC_INTERNAL_ERROR, "SchedulingPolicy is bucketed, but BucketHandle is not present");
        }

        if (bucketHandle.isPresent()) {
            if (bucketHandle.get().getReadBucketCount() > bucketHandle.get().getTableBucketCount()) {
                throw new PrestoException(
                        GENERIC_INTERNAL_ERROR,
                        "readBucketCount (%s) is greater than the tableBucketCount (%s) which generally points to an issue in plan generation");
            }
        }

        // sort partitions
        partitions = Ordering.natural().onResultOf(HivePartition::getPartitionId).reverse().sortedCopy(partitions);

        Iterable<HivePartitionMetadata> hivePartitions = getPartitionMetadata(
                metastore,
                table,
                tableName,
                partitions,
                bucketHandle,
                session,
                splitSchedulingContext.getWarningCollector(),
                layout.getRequestedColumns(),
                layout.getPredicateColumns(),
                layout.getDomainPredicate().getDomains());

        double ratio = getSplitScanRatio(session, tableName, layout, metadata, partitions);

        HiveSplitLoader hiveSplitLoader = new BackgroundHiveSplitLoader(
                table,
                hivePartitions,
                getInfoColumnConstraints(layout.getDomainPredicate(), layout.getPredicateColumns()),
                createBucketSplitInfo(bucketHandle, bucketFilter),
                session,
                hdfsEnvironment,
                namenodeStats,
                directoryLister,
                executor,
                min(splitLoaderConcurrency, partitions.size()), // Avoid over-committing split loader concurrency
                recursiveDfsWalkerEnabled,
                splitSchedulingContext.schedulerUsesHostAddresses(),
                layout.isPartialAggregationsPushedDown());

        HiveSplitSource splitSource = computeSplitSource(splitSchedulingContext, table, session, hiveSplitLoader, ratio);
        hiveSplitLoader.start(splitSource);

        return splitSource;
    }
@Test
    public void testEncryptionInformation()
            throws Exception
    {
        StorageFormat storageFormat = fromHiveStorageFormat(DWRF);
        String testEncryptionAlgorithm = "test_encryption_algo";
        String testEncryptionProvider = "test_provider";
        Table testTable = createTestTable(storageFormat,
                ImmutableMap.of(
                        ENCRYPT_COLUMNS_KEY, "foo1:col_bigint,col_struct.b.b1;foo2:col_map;foo3:col_struct.a",
                        DWRF_ENCRYPTION_ALGORITHM_KEY, testEncryptionAlgorithm,
                        DWRF_ENCRYPTION_PROVIDER_KEY, testEncryptionProvider));
        PartitionWithStatistics partitionWithStatistics = new PartitionWithStatistics(
                new Partition(
                        "test_db",
                        "test_table",
                        ImmutableList.of(PARTITION_VALUE),
                        new Storage(
                                storageFormat,
                                "location",
                                Optional.empty(),
                                true,
                                ImmutableMap.of(),
                                ImmutableMap.of()),
                        COLUMNS,
                        ImmutableMap.of(),
                        Optional.empty(),
                        false,
                        true,
                        0,
                        0,
                        Optional.empty()),
                PARTITION_NAME,
                PartitionStatistics.empty());

        HiveClientConfig hiveClientConfig = new HiveClientConfig().setPartitionStatisticsBasedOptimizationEnabled(true);
        HdfsEnvironment hdfsEnvironment = new HdfsEnvironment(
                new HiveHdfsConfiguration(new HdfsConfigurationInitializer(hiveClientConfig, new MetastoreClientConfig()), ImmutableSet.of(), hiveClientConfig),
                new MetastoreClientConfig(),
                new NoHdfsAuthentication());
        HiveEncryptionInformationProvider encryptionInformationProvider = new HiveEncryptionInformationProvider(ImmutableList.of(new TestDwrfEncryptionInformationSource()));

        HiveMetadataFactory metadataFactory = new HiveMetadataFactory(
                new TestingExtendedHiveMetastore(testTable, partitionWithStatistics),
                hdfsEnvironment,
                new HivePartitionManager(FUNCTION_AND_TYPE_MANAGER, hiveClientConfig),
                DateTimeZone.forOffsetHours(1),
                true,
                false,
                false,
                true,
                true,
                hiveClientConfig.getMaxPartitionBatchSize(),
                hiveClientConfig.getMaxPartitionsPerScan(),
                false,
                10_000,
                FUNCTION_AND_TYPE_MANAGER,
                new HiveLocationService(hdfsEnvironment),
                FUNCTION_RESOLUTION,
                ROW_EXPRESSION_SERVICE,
                FILTER_STATS_CALCULATOR_SERVICE,
                new TableParameterCodec(),
                HiveTestUtils.PARTITION_UPDATE_CODEC,
                HiveTestUtils.PARTITION_UPDATE_SMILE_CODEC,
                executor,
                new HiveTypeTranslator(),
                new HiveStagingFileCommitter(hdfsEnvironment, executor),
                new HiveZeroRowFileCreator(hdfsEnvironment, new OutputStreamDataSinkFactory(), executor),
                TEST_SERVER_VERSION,
                new HivePartitionObjectBuilder(),
                encryptionInformationProvider,
                new HivePartitionStats(),
                new HiveFileRenamer(),
                HiveColumnConverterProvider.DEFAULT_COLUMN_CONVERTER_PROVIDER,
                new QuickStatsProvider(HDFS_ENVIRONMENT, DO_NOTHING_DIRECTORY_LISTER, new HiveClientConfig(), new NamenodeStats(), ImmutableList.of()),
                new HiveTableWritabilityChecker(false));

        HiveSplitManager splitManager = new HiveSplitManager(
                new TestingHiveTransactionManager(metadataFactory),
                new NamenodeStats(),
                hdfsEnvironment,
                new TestingDirectoryLister(),
                directExecutor(),
                new HiveCoercionPolicy(FUNCTION_AND_TYPE_MANAGER),
                new CounterStat(),
                100,
                hiveClientConfig.getMaxOutstandingSplitsSize(),
                hiveClientConfig.getMinPartitionBatchSize(),
                hiveClientConfig.getMaxPartitionBatchSize(),
                hiveClientConfig.getSplitLoaderConcurrency(),
                false,
                new ConfigBasedCacheQuotaRequirementProvider(new CacheConfig()),
                encryptionInformationProvider,
                new HivePartitionSkippabilityChecker());

        HiveColumnHandle partitionColumn = new HiveColumnHandle(
                "ds",
                HIVE_STRING,
                parseTypeSignature(VARCHAR),
                MAX_PARTITION_KEY_COLUMN_INDEX,
                PARTITION_KEY,
                Optional.empty(),
                Optional.empty());
        List<HivePartition> partitions = ImmutableList.of(
                new HivePartition(
                        new SchemaTableName("test_schema", "test_table"),
                        PARTITION_NAME_WITH_VERSION,
                        ImmutableMap.of(partitionColumn, NullableValue.of(createUnboundedVarcharType(), utf8Slice(PARTITION_VALUE)))));

        SchemaTableName schemaTableName = new SchemaTableName("test_schema", "test_table");
        HiveTableHandle hiveTableHandle = new HiveTableHandle(schemaTableName.getSchemaName(), schemaTableName.getTableName());
        HiveTableLayoutHandle layoutHandle = new HiveTableLayoutHandle.Builder()
                .setSchemaTableName(schemaTableName)
                .setTablePath("test_path")
                .setPartitionColumns(ImmutableList.of(partitionColumn))
                .setDataColumns(COLUMNS)
                .setTableParameters(ImmutableMap.of())
                .setDomainPredicate(TupleDomain.all())
                .setRemainingPredicate(TRUE_CONSTANT)
                .setPredicateColumns(ImmutableMap.of())
                .setPartitionColumnPredicate(TupleDomain.all())
                .setPartitions(partitions)
                .setBucketHandle(Optional.empty())
                .setBucketFilter(Optional.empty())
                .setPushdownFilterEnabled(false)
                .setLayoutString("layout")
                .setRequestedColumns(Optional.empty())
                .setPartialAggregationsPushedDown(false)
                .setAppendRowNumberEnabled(false)
                .setHiveTableHandle(hiveTableHandle)
                .build();

        ConnectorSplitSource splitSource = splitManager.getSplits(
                new HiveTransactionHandle(),
                new TestingConnectorSession(getAllSessionProperties(hiveClientConfig, new HiveCommonClientConfig())),
                layoutHandle,
                SPLIT_SCHEDULING_CONTEXT);
        Optional<EncryptionInformation> encryptionInformation = splitSource.getNextBatch(NOT_PARTITIONED, 100).get().getSplits()
                .stream()
                .map(HiveSplit.class::cast)
                .map(HiveSplit::getEncryptionInformation)
                .findFirst()
                .get();
        assertTrue(encryptionInformation.isPresent());
        Optional<DwrfEncryptionMetadata> dwrfEncryptionMetadata = encryptionInformation.get().getDwrfEncryptionMetadata();
        assertTrue(dwrfEncryptionMetadata.isPresent());
        assertEquals(dwrfEncryptionMetadata.get().getEncryptionAlgorithm(), testEncryptionAlgorithm);
        assertEquals(dwrfEncryptionMetadata.get().getEncryptionProvider(), testEncryptionProvider);
        assertEquals(dwrfEncryptionMetadata.get().getFieldToKeyData().size(), 4);
    }