@Override
    public int partition(ShuffleKey key, int numPartitions) {
      if (helper != null) {
        helper.open();
      }
      checkNotNull(key);
      RowData row = checkNotNull(key.getRow());

      Integer pkHashCode = null;
      if (distributionHashMode.isSupportPrimaryKey()) {
        pkHashCode = helper.hashKeyValue(row);
      }
      // shuffle by mixed-format partition for partitioned table
      Integer partitionHashCode = null;
      if (distributionHashMode.isSupportPartition()) {
        partitionHashCode = helper.hashPartitionValue(row);
      }
      if (pkHashCode != null && partitionHashCode != null) {
        return Math.abs(Objects.hash(pkHashCode, partitionHashCode)) % numPartitions;
      } else if (pkHashCode != null) {
        return pkHashCode % numPartitions;
      } else if (partitionHashCode != null) {
        return partitionHashCode % numPartitions;
      } else {
        return random.nextInt(numPartitions);
      }
    }
Circular polling feed a streamRecord into a special factor node
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/shuffle/ReadShuffleRulePolicy.java