static PartitionFunction createPartitionFunction(
            PartitioningProviderManager partitioningProviderManager,
            Session session,
            PartitioningHandle partitioning,
            int partitionCount,
            List<Type> partitioningChannelTypes,
            boolean isHashPrecomputed)
    {
        if (partitioning.getConnectorHandle() instanceof SystemPartitioningHandle) {
            HashGenerator hashGenerator;
            if (isHashPrecomputed) {
                hashGenerator = new PrecomputedHashGenerator(0);
            }
            else {
                hashGenerator = InterpretedHashGenerator.createPositionalWithTypes(partitioningChannelTypes);
            }
            return new LocalPartitionGenerator(hashGenerator, partitionCount);
        }

        ConnectorNodePartitioningProvider partitioningProvider = partitioningProviderManager.getPartitioningProvider(partitioning.getConnectorId().get());
        int bucketCount = partitioningProvider.getBucketCount(
                partitioning.getTransactionHandle().orElse(null),
                session.toConnectorSession(partitioning.getConnectorId().get()),
                partitioning.getConnectorHandle());
        int[] bucketToPartition = new int[bucketCount];
        for (int bucket = 0; bucket < bucketCount; bucket++) {
            bucketToPartition[bucket] = bucket % partitionCount;
        }

        BucketFunction bucketFunction = partitioningProvider.getBucketFunction(
                partitioning.getTransactionHandle().orElse(null),
                session.toConnectorSession(partitioning.getConnectorId().get()),
                partitioning.getConnectorHandle(),
                partitioningChannelTypes,
                bucketCount);

        checkArgument(bucketFunction != null, "No bucket function for partitioning: %s", partitioning);
        return new BucketPartitionFunction(bucketFunction, bucketToPartition);
    }
@Test
    public void testCreatePartitionFunction()
    {
        int partitionCount = 10;
        PartitioningProviderManager partitioningProviderManager = new PartitioningProviderManager();
        partitioningProviderManager.addPartitioningProvider(
                new ConnectorId("prism"),
                new ConnectorNodePartitioningProvider() {
                    @Override
                    public ConnectorBucketNodeMap getBucketNodeMap(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorPartitioningHandle partitioningHandle, List<Node> sortedNodes)
                    {
                        return createBucketNodeMap(Stream.generate(() -> sortedNodes).flatMap(List::stream).limit(10).collect(toImmutableList()), SOFT_AFFINITY);
                    }

                    @Override
                    public ToIntFunction<ConnectorSplit> getSplitBucketFunction(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorPartitioningHandle partitioningHandle)
                    {
                        return null;
                    }

                    @Override
                    public BucketFunction getBucketFunction(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorPartitioningHandle partitioningHandle, List<Type> partitionChannelTypes, int bucketCount)
                    {
                        return (Page page, int position) -> partitionCount;
                    }

                    @Override
                    public int getBucketCount(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorPartitioningHandle partitioningHandle)
                    {
                        return 10;
                    }
                });
        PartitioningHandle partitioningHandle = new PartitioningHandle(
                Optional.of(new ConnectorId("prism")),
                Optional.of(new ConnectorTransactionHandle() {
                    @Override
                    public int hashCode()
                    {
                        return super.hashCode();
                    }

                    @Override
                    public boolean equals(Object obj)
                    {
                        return super.equals(obj);
                    }
                }),
                new ConnectorPartitioningHandle() {
                });
        PartitionFunction partitionFunction = createPartitionFunction(partitioningProviderManager, session, partitioningHandle, 600, ImmutableList.of(), false);

        assertEquals(partitionFunction.getPartitionCount(), partitionCount);
    }