presto-main/src/main/java/io/prestosql/sql/planner/NodePartitioningManager.java/NodePartitioningManager/getPartitionFunction
class NodePartitioningManager:    public PartitionFunction getPartitionFunction(
            Session session,
            PartitioningScheme partitioningScheme,
            List<Type> partitionChannelTypes)
    {
        Optional<int[]> bucketToPartition = partitioningScheme.getBucketToPartition();
        checkArgument(bucketToPartition.isPresent(), "Bucket to partition must be set before a partition function can be created");

        PartitioningHandle partitioningHandle = partitioningScheme.getPartitioning().getHandle();
        BucketFunction bucketFunction;
        if (partitioningHandle.getConnectorHandle() instanceof SystemPartitioningHandle) {
            checkArgument(partitioningScheme.getBucketToPartition().isPresent(), "Bucket to partition must be set before a partition function can be created");

            return ((SystemPartitioningHandle) partitioningHandle.getConnectorHandle()).getPartitionFunction(
                    partitionChannelTypes,
                    partitioningScheme.getHashColumn().isPresent(),
                    partitioningScheme.getBucketToPartition().get());
        }
        CatalogName catalogName = partitioningHandle.getConnectorId().get();
        ConnectorNodePartitioningProvider partitioningProvider = partitioningProviders.get(catalogName);
        checkArgument(partitioningProvider != null, "No partitioning provider for connector %s", catalogName);

        bucketFunction = partitioningProvider.getBucketFunction(
                partitioningHandle.getTransactionHandle().orElse(null),
                session.toConnectorSession(catalogName),
                partitioningHandle.getConnectorHandle(),
                partitionChannelTypes,
                bucketToPartition.get().length);

        checkArgument(bucketFunction != null, "No function %s", partitioningHandle);
        return new BucketPartitionFunction(bucketFunction, partitioningScheme.getBucketToPartition().get());
    }