public static Optional<ConnectorPageSource> createHivePageSource(
            Set<HiveRecordCursorProvider> cursorProviders,
            Set<HiveBatchPageSourceFactory> pageSourceFactories,
            Configuration configuration,
            ConnectorSession session,
            HiveFileSplit fileSplit,
            OptionalInt tableBucketNumber,
            Storage storage,
            TupleDomain<HiveColumnHandle> effectivePredicate,
            List<HiveColumnHandle> hiveColumns,
            Map<String, HiveColumnHandle> predicateColumns,
            List<HivePartitionKey> partitionKeys,
            DateTimeZone hiveStorageTimeZone,
            TypeManager typeManager,
            SchemaTableName tableName,
            List<HiveColumnHandle> partitionKeyColumnHandles,
            List<Column> tableDataColumns,
            Map<String, String> tableParameters,
            int partitionDataColumnCount,
            TableToPartitionMapping tableToPartitionMapping,
            Optional<BucketConversion> bucketConversion,
            boolean s3SelectPushdownEnabled,
            HiveFileContext hiveFileContext,
            RowExpression remainingPredicate,
            boolean isPushdownFilterEnabled,
            RowExpressionService rowExpressionService,
            Optional<EncryptionInformation> encryptionInformation,
            Optional<byte[]> rowIdPartitionComponent)
    {
        List<HiveColumnHandle> allColumns;

        if (isPushdownFilterEnabled) {
            Set<String> columnNames = hiveColumns.stream().map(HiveColumnHandle::getName).collect(toImmutableSet());
            List<HiveColumnHandle> additionalColumns = predicateColumns.values().stream()
                    .filter(column -> !columnNames.contains(column.getName()))
                    .collect(toImmutableList());

            allColumns = ImmutableList.<HiveColumnHandle>builder()
                    .addAll(hiveColumns)
                    .addAll(additionalColumns)
                    .build();
        }
        else {
            allColumns = hiveColumns;
        }

        List<ColumnMapping> columnMappings = ColumnMapping.buildColumnMappings(
                partitionKeys,
                allColumns,
                bucketConversion.map(BucketConversion::getBucketColumnHandles).orElse(ImmutableList.of()),
                tableToPartitionMapping,
                fileSplit,
                tableBucketNumber);
        Set<Integer> outputIndices = hiveColumns.stream()
                .map(HiveColumnHandle::getHiveColumnIndex)
                .collect(toImmutableSet());

        // Finds the non-synthetic columns.
        List<ColumnMapping> regularAndInterimColumnMappings = ColumnMapping.extractRegularAndInterimColumnMappings(columnMappings);

        Optional<BucketAdaptation> bucketAdaptation = bucketConversion.map(conversion -> toBucketAdaptation(
                conversion,
                regularAndInterimColumnMappings,
                tableBucketNumber,
                ColumnMapping::getIndex,
                isLegacyTimestampBucketing(session)));

        if (isUseRecordPageSourceForCustomSplit(session) && shouldUseRecordReaderFromInputFormat(configuration, storage, fileSplit.getCustomSplitInfo())) {
            return getPageSourceFromCursorProvider(
                    cursorProviders,
                    configuration,
                    session,
                    fileSplit,
                    storage,
                    effectivePredicate,
                    hiveStorageTimeZone,
                    typeManager,
                    tableName,
                    partitionKeyColumnHandles,
                    tableDataColumns,
                    tableParameters,
                    partitionDataColumnCount,
                    tableToPartitionMapping,
                    s3SelectPushdownEnabled,
                    remainingPredicate,
                    isPushdownFilterEnabled,
                    rowExpressionService,
                    allColumns,
                    columnMappings,
                    outputIndices,
                    regularAndInterimColumnMappings,
                    bucketAdaptation);
        }

        for (HiveBatchPageSourceFactory pageSourceFactory : pageSourceFactories) {
            Optional<? extends ConnectorPageSource> pageSource = pageSourceFactory.createPageSource(
                    configuration,
                    session,
                    fileSplit,
                    storage,
                    tableName,
                    tableParameters,
                    toColumnHandles(regularAndInterimColumnMappings, true),
                    effectivePredicate,
                    hiveStorageTimeZone,
                    hiveFileContext,
                    encryptionInformation,
                    rowIdPartitionComponent);
            if (pageSource.isPresent()) {
                HivePageSource hivePageSource = new HivePageSource(
                        columnMappings,
                        bucketAdaptation,
                        hiveStorageTimeZone,
                        typeManager,
                        pageSource.get(),
                        fileSplit.getPath(),
                        rowIdPartitionComponent);

                if (isPushdownFilterEnabled) {
                    return Optional.of(new FilteringPageSource(
                            columnMappings,
                            effectivePredicate,
                            remainingPredicate,
                            typeManager,
                            rowExpressionService,
                            session,
                            outputIndices,
                            hivePageSource));
                }
                return Optional.of(hivePageSource);
            }
        }

        return getPageSourceFromCursorProvider(
                cursorProviders,
                configuration,
                session,
                fileSplit,
                storage,
                effectivePredicate,
                hiveStorageTimeZone,
                typeManager,
                tableName,
                partitionKeyColumnHandles,
                tableDataColumns,
                tableParameters,
                partitionDataColumnCount,
                tableToPartitionMapping,
                s3SelectPushdownEnabled,
                remainingPredicate,
                isPushdownFilterEnabled,
                rowExpressionService,
                allColumns,
                columnMappings,
                outputIndices,
                regularAndInterimColumnMappings,
                bucketAdaptation);
    }
@Test
    public void testUseRecordReaderWithInputFormatAnnotationAndCustomSplit()
    {
        StorageFormat storageFormat = StorageFormat.create(ParquetHiveSerDe.class.getName(), HoodieParquetInputFormat.class.getName(), "");
        Storage storage = new Storage(storageFormat, "test", Optional.empty(), true, ImmutableMap.of(), ImmutableMap.of());
        Map<String, String> customSplitInfo = ImmutableMap.of(
                CUSTOM_FILE_SPLIT_CLASS_KEY, HoodieRealtimeFileSplit.class.getName(),
                HUDI_BASEPATH_KEY, "/test/file.parquet",
                HUDI_DELTA_FILEPATHS_KEY, "/test/.file_100.log",
                HUDI_MAX_COMMIT_TIME_KEY, "100");
        HiveRecordCursorProvider recordCursorProvider = new MockHiveRecordCursorProvider();
        HiveBatchPageSourceFactory hiveBatchPageSourceFactory = new MockHiveBatchPageSourceFactory();

        HiveFileSplit fileSplit = new HiveFileSplit(
                "/test/",
                0,
                100,
                200,
                Instant.now().toEpochMilli(),
                Optional.empty(),
                customSplitInfo,
                0);
        Optional<ConnectorPageSource> pageSource = HivePageSourceProvider.createHivePageSource(
                ImmutableSet.of(recordCursorProvider),
                ImmutableSet.of(hiveBatchPageSourceFactory),
                new Configuration(),
                new TestingConnectorSession(new HiveSessionProperties(
                        new HiveClientConfig().setUseRecordPageSourceForCustomSplit(true),
                        new OrcFileWriterConfig(),
                        new ParquetFileWriterConfig(),
                        new CacheConfig()).getSessionProperties()),
                fileSplit,
                OptionalInt.empty(),
                storage,
                TupleDomain.none(),
                ImmutableList.of(),
                ImmutableMap.of(),
                ImmutableList.of(),
                DateTimeZone.UTC,
                new TestingTypeManager(),
                new SchemaTableName("test", "test"),
                ImmutableList.of(),
                ImmutableList.of(),
                ImmutableMap.of(),
                0,
                TableToPartitionMapping.empty(),
                Optional.empty(),
                false,
                null,
                null,
                false,
                null,
                Optional.empty(),
                Optional.empty());
        assertTrue(pageSource.isPresent());
        assertTrue(pageSource.get() instanceof RecordPageSource);
    }