## Repartitioning Operation

@param partitioner
@return
191521
public Stream partition(CustomStreamGrouping partitioner) {
        return partition(Grouping.custom_serialized(Utils.javaSerialize(partitioner)));
    }
920257
static IFunction<Data, Entry<Integer, Data>> toPartition(final IPartitionService partitionService) {
        return new DataToEntry(partitionService);
    }
113485
public static <T> Reducer<PersistentList<T>,T> toPersistentListReversed() {
        return Reducer.fromMonoid(Monoids.concatPersistentCollection(Seq.empty()), a -> Seq.of(a));
    }
778939
public DatabaseURIHelper partition(String partitionKey) {
        if (partitionKey != null) {
            this.path("_partition").path(partitionKey);
        }
        return returnThis();
    }
956557
public int[] partition(double h) {
        for (int i = 0; i < height.length - 1; i++) {
            if (height[i] > height[i + 1]) {
                throw new IllegalStateException("Non-monotonic cluster tree -- the linkage is probably not appropriate!");
            }
        }

        int n = merge.length + 1;
        int k = 2;
        for (; k <= n; k++) {
            if (height[n - k] < h) {
                break;
            }
        }

        if (k <= 2) {
            throw new IllegalArgumentException("The parameter h is too large.");
        }

        return partition(k - 1);
    }
1314547
public int getPartition(byte[] key,
                            byte[] value,
                            int numReduceTasks) {
        try {
            /**
             * {@link partitionId} is the Voldemort primary partition that this
             * record belongs to.
             */
            int partitionId = ByteUtils.readInt(value, ByteUtils.SIZE_OF_INT);

            /**
             * This is the base number we will ultimately mod by {@link numReduceTasks}
             * to determine which reduce task to shuffle to.
             */
            int magicNumber = partitionId;

            if (getSaveKeys() && !buildPrimaryReplicasOnly) {
                /**
                 * When saveKeys is enabled (which also implies we are generating
                 * READ_ONLY_V2 format files), then we are generating files with
                 * a replica type, with one file per replica.
                 *
                 * Each replica is sent to a different reducer, and thus the
                 * {@link magicNumber} is scaled accordingly.
                 *
                 * The downside to this is that it is pretty wasteful. The files
                 * generated for each replicas are identical to one another, so
                 * there's no point in generating them independently in many
                 * reducers.
                 *
                 * This is one of the reasons why buildPrimaryReplicasOnly was
                 * written. In this mode, we only generate the files for the
                 * primary replica, which means the number of reducers is
                 * minimized and {@link magicNumber} does not need to be scaled.
                 */
                int replicaType = (int) ByteUtils.readBytes(value,
                                                            2 * ByteUtils.SIZE_OF_INT,
                                                            ByteUtils.SIZE_OF_BYTE);
                magicNumber = magicNumber * getStoreDef().getReplicationFactor() + replicaType;
            }

            if (!getReducerPerBucket()) {
                /**
                 * Partition files can be split in many chunks in order to limit the
                 * maximum file size downloaded and handled by Voldemort servers.
                 *
                 * {@link chunkId} represents which chunk of partition then current
                 * record belongs to.
                 */
                int chunkId = ReadOnlyUtils.chunk(key, getNumChunks());

                /**
                 * When reducerPerBucket is disabled, all chunks are sent to a
                 * different reducer. This increases parallelism at the expense
                 * of adding more load on Hadoop.
                 *
                 * {@link magicNumber} is thus scaled accordingly, in order to
                 * leverage the extra reducers available to us.
                 */
                magicNumber = magicNumber * getNumChunks() + chunkId;
            }

            /**
             * Finally, we mod {@link magicNumber} by {@link numReduceTasks},
             * since the MapReduce framework expects the return of this function
             * to be bounded by the number of reduce tasks running in the job.
             */
            return magicNumber % numReduceTasks;
        } catch (Exception e) {
            throw new VoldemortException("Caught exception in getPartition()!" +
                                         " key: " + ByteUtils.toHexString(key) +
                                         ", value: " + ByteUtils.toHexString(value) +
                                         ", numReduceTasks: " + numReduceTasks, e);
        }
    }
615539
public static List<PartitionDefinition> assignPartitions(List<PartitionDefinition> partitions, int currentTask, int totalTasks) {
        int esPartitions = partitions.size();
        if (totalTasks >= esPartitions) {
            return (currentTask >= esPartitions ? Collections.<PartitionDefinition>emptyList() : Collections.singletonList(partitions.get(currentTask)));
        } else {
            int partitionsPerTask = esPartitions / totalTasks;
            int remainder = esPartitions % totalTasks;

            int partitionsPerCurrentTask = partitionsPerTask;

            // spread the reminder against the tasks
            if (currentTask < remainder) {
                partitionsPerCurrentTask++;
            }

            // find the offset inside the collection
            int offset = partitionsPerTask * currentTask;
            if (currentTask != 0) {
                offset += (remainder > currentTask ? 1 : remainder);
            }

            // common case
            if (partitionsPerCurrentTask == 1) {
                return Collections.singletonList(partitions.get(offset));
            }

            List<PartitionDefinition> pa = new ArrayList<PartitionDefinition>(partitionsPerCurrentTask);
            for (int index = offset; index < offset + partitionsPerCurrentTask; index++) {
                pa.add(partitions.get(index));
            }
            return pa;
        }
    }
411004
@PublicEvolving
	public PartitionOperator<T> withOrders(Order... orders) {
		Preconditions.checkState(pMethod == PartitionMethod.RANGE, "Orders cannot be applied for %s partition " +
				"method", pMethod);
		Preconditions.checkArgument(pKeys.getOriginalKeyFieldTypes().length == orders.length, "The number of key " +
				"fields and orders should be the same.");
		this.orders = orders;

		return this;
	}
423624
@SuppressWarnings("unchecked")
  public Class<? extends Partitioner<?,?>> getPartitionerClass() 
     throws ClassNotFoundException {
    return (Class<? extends Partitioner<?,?>>) 
      conf.getClass(PARTITIONER_CLASS_ATTR, HashPartitioner.class);
  }
1040433
public PartitionMapper<Partition<T>> getOrCreateMapper()
   {
      Node node = childNode.getOrCreate("mapper");
      PartitionMapper<Partition<T>> mapper = new PartitionMapperImpl<Partition<T>>(this, "mapper", childNode, node);
      return mapper;
   }
1272928
@Override
  public Streamlet<R> repartition(int numPartitions) {
    return this.map((a) -> a).setNumPartitions(numPartitions);
  }
1201176
private String getMismatchedPartitioner(Class<? extends IPartitioner> expectedPartitioner) {
        String partitioner = null;
        try {
            partitioner = _astyanaxKeyspace.describePartitioner();
            boolean matches = CassandraPartitioner.fromClass(partitioner).matches(expectedPartitioner.getName());
            if (matches) {
                return null;
            } else {
                return partitioner;
            }
        } catch (ConnectionException e) {
            throw Throwables.propagate(e);
        } catch (IllegalArgumentException e) {
            // Only thrown if the partitioner doesn't match any compatible partitioner, so by definition it is mismatched.
            return partitioner;
        }

    }
1349702
private void compactPartition(final int partitionNumber) throws IOException {
		// do nothing if table was closed, parameter is invalid or no garbage exists
		if (this.closed || partitionNumber >= this.partitions.size() || this.partitions.get(partitionNumber).isCompacted()) {
			return;
		}
		// release all segments owned by compaction partition
		this.compactionMemory.clearAllMemory(availableMemory);
		this.compactionMemory.allocateSegments(1);
		this.compactionMemory.pushDownPages();
		T tempHolder = this.buildSideSerializer.createInstance();
		final int numPartitions = this.partitions.size();
		InMemoryPartition<T> partition = this.partitions.remove(partitionNumber);
		MemorySegment[] overflowSegments = partition.overflowSegments;
		long pointer;
		int pointerOffset;
		int bucketOffset;
		final int bucketsPerSegment = this.bucketsPerSegmentMask + 1;
		for (int i = 0, bucket = partitionNumber; i < this.buckets.length && bucket < this.numBuckets; i++) {
			MemorySegment segment = this.buckets[i];
			// go over all buckets in the segment belonging to the partition
			for (int k = bucket % bucketsPerSegment; k < bucketsPerSegment && bucket < this.numBuckets; k += numPartitions, bucket += numPartitions) {
				bucketOffset = k * HASH_BUCKET_SIZE;
				if((int)segment.get(bucketOffset + HEADER_PARTITION_OFFSET) != partitionNumber) {
					throw new IOException("Accessed wrong bucket! wanted: " + partitionNumber + " got: " + segment.get(bucketOffset + HEADER_PARTITION_OFFSET));
				}
				// loop over all segments that are involved in the bucket (original bucket plus overflow buckets)
				int countInSegment = segment.getInt(bucketOffset + HEADER_COUNT_OFFSET);
				int numInSegment = 0;
				pointerOffset = bucketOffset + BUCKET_POINTER_START_OFFSET;
				while (true) {
					while (numInSegment < countInSegment) {
						pointer = segment.getLong(pointerOffset);
						tempHolder = partition.readRecordAt(pointer, tempHolder);
						pointer = this.compactionMemory.appendRecord(tempHolder);
						segment.putLong(pointerOffset, pointer);
						pointerOffset += POINTER_LEN;
						numInSegment++;
					}
					// this segment is done. check if there is another chained bucket
					final long forwardPointer = segment.getLong(bucketOffset + HEADER_FORWARD_OFFSET);
					if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {
						break;
					}
					final int overflowSegNum = (int) (forwardPointer >>> 32);
					segment = overflowSegments[overflowSegNum];
					bucketOffset = (int) forwardPointer;
					countInSegment = segment.getInt(bucketOffset + HEADER_COUNT_OFFSET);
					pointerOffset = bucketOffset + BUCKET_POINTER_START_OFFSET;
					numInSegment = 0;
				}
				segment = this.buckets[i];
			}
		}
		// swap partition with compaction partition
		this.compactionMemory.setPartitionNumber(partitionNumber);
		this.partitions.add(partitionNumber, compactionMemory);
		this.partitions.get(partitionNumber).overflowSegments = partition.overflowSegments;
		this.partitions.get(partitionNumber).numOverflowSegments = partition.numOverflowSegments;
		this.partitions.get(partitionNumber).nextOverflowBucket = partition.nextOverflowBucket;
		this.partitions.get(partitionNumber).setIsCompacted(true);
		//this.partitions.get(partitionNumber).pushDownPages();
		this.compactionMemory = partition;
		this.compactionMemory.resetRecordCounter();
		this.compactionMemory.setPartitionNumber(-1);
		this.compactionMemory.overflowSegments = null;
		this.compactionMemory.numOverflowSegments = 0;
		this.compactionMemory.nextOverflowBucket = 0;
		// try to allocate maximum segment count
		this.compactionMemory.clearAllMemory(this.availableMemory);
		int maxSegmentNumber = this.getMaxPartition();
		this.compactionMemory.allocateSegments(maxSegmentNumber);
		this.compactionMemory.resetRWViews();
		this.compactionMemory.pushDownPages();
	}
262505
PartitionStrategy getSubpartitionStrategy(int startIndex) {
    if (startIndex == 0) {
      return this;
    }
    if (startIndex >= fieldPartitioners.size()) {
      return null;
    }
    return new PartitionStrategy(fieldPartitioners.subList(startIndex,
        fieldPartitioners.size()));
  }
1105588
@JsonCreator
    public static Partitioning jsonCreate(
            @JsonProperty("handle") PartitioningHandle handle,
            @JsonProperty("arguments") List<ArgumentBinding> arguments)
    {
        return new Partitioning(handle, arguments);
    }
286103
private static List<Element> getLastPartition(List<List<Element>> partitions) {
		if (partitions.isEmpty()) {
			List<Element> newPartition = new LinkedList<Element>();
			partitions.add(newPartition);
			return newPartition;
		} else {
			return partitions.get(partitions.size() - 1);
		}
	}
351254
public PartitionInput withParameters(java.util.Map<String, String> parameters) {
        setParameters(parameters);
        return this;
    }