presto-iceberg/src/main/java/io/prestosql/plugin/iceberg/IcebergPageSourceProvider.java/IcebergPageSourceProvider/createPageSource
class IcebergPageSourceProvider:    @Override
    public ConnectorPageSource createPageSource(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit connectorSplit, ConnectorTableHandle connectorTable, List<ColumnHandle> columns)
    {
        IcebergSplit split = (IcebergSplit) connectorSplit;
        IcebergTableHandle table = (IcebergTableHandle) connectorTable;

        List<IcebergColumnHandle> icebergColumns = columns.stream()
                .map(IcebergColumnHandle.class::cast)
                .collect(toImmutableList());

        Map<Integer, String> partitionKeys = split.getPartitionKeys();

        List<IcebergColumnHandle> regularColumns = columns.stream()
                .map(IcebergColumnHandle.class::cast)
                .filter(column -> !partitionKeys.containsKey(column.getId()))
                .collect(toImmutableList());

        Path path = new Path(split.getPath());
        long start = split.getStart();
        long length = split.getLength();
        HdfsContext hdfsContext = new HdfsContext(session, table.getSchemaName(), table.getTableName());
        ConnectorPageSource parquetPageSource = createParquetPageSource(
                hdfsEnvironment,
                session.getUser(),
                hdfsEnvironment.getConfiguration(hdfsContext, path),
                path,
                start,
                length,
                regularColumns,
                options
                        .withFailOnCorruptedStatistics(isFailOnCorruptedParquetStatistics(session))
                        .withMaxReadBlockSize(getParquetMaxReadBlockSize(session)),
                split.getPredicate(),
                fileFormatDataSourceStats);

        return new IcebergPageSource(icebergColumns, partitionKeys, parquetPageSource, session.getTimeZoneKey());
    }