Selects a partition and spills it. The number of the spilled partition is returned.

@return The number of the spilled partition.
198676
@Override
	protected int spillPartition() throws IOException {
		// find the largest partition
		int largestNumBlocks = 0;
		int largestPartNum = -1;

		for (int i = 0; i < partitionsBeingBuilt.size(); i++) {
			BinaryHashPartition p = partitionsBeingBuilt.get(i);
			if (p.isInMemory() && p.getNumOccupiedMemorySegments() > largestNumBlocks) {
				largestNumBlocks = p.getNumOccupiedMemorySegments();
				largestPartNum = i;
			}
		}
		final BinaryHashPartition p = partitionsBeingBuilt.get(largestPartNum);

		// spill the partition
		int numBuffersFreed = p.spillPartition(this.ioManager,
				this.currentEnumerator.next(), this.buildSpillReturnBuffers);
		this.buildSpillRetBufferNumbers += numBuffersFreed;

		LOG.info(String.format("Grace hash join: Ran out memory, choosing partition " +
						"[%d] to spill, %d memory segments being freed",
				largestPartNum, numBuffersFreed));

		// grab as many buffers as are available directly
		MemorySegment currBuff;
		while (this.buildSpillRetBufferNumbers > 0 && (currBuff = this.buildSpillReturnBuffers.poll()) != null) {
			this.availableMemory.add(currBuff);
			this.buildSpillRetBufferNumbers--;
		}
		numSpillFiles++;
		spillInBytes += numBuffersFreed * segmentSize;
		// The bloomFilter is built after the data is spilled, so that we can use enough memory.
		p.buildBloomFilterAndFreeBucket();
		return largestPartNum;
	}
1011389
public InternalPartition[] getPartitionsCopy() {
        NopPartitionListener listener = new NopPartitionListener();
        InternalPartition[] result = new InternalPartition[partitions.length];
        for (int i = 0; i < partitionCount; i++) {
            result[i] = partitions[i].copy(listener);
        }
        return result;
    }
534108
def getStringPartition(self):
        
        res = ''
        for s in self.partitions[self.index].getSlice():
            start = ''
            stop = ''
            if s.start is not None:
                start = int(s.start)
            if s.stop is not None:
                stop = int(s.stop)
            res += '{0}:{1},'.format(start, stop)
        return res
347302
public Integer getSelectedLastRow() {
        final Object result = getStateHelper().eval(PropertyKeys.selectedLastRow);
        if (result == null) {
            return null;
        }
        return Integer.valueOf(result.toString());
    }
327819
public int getNumberAvailable() {
        int result = 0;
        for (int i = 0; i < data.size(); i++) {
            if (data.get(i) != null) {
                result += 1;
            }
        }
        return result;
    }
617396
public Integer getSelectedRow() {
        final Object result = getStateHelper().eval(PropertyKeys.selectedRow);
        if (result == null) {
            return null;
        }
        return Integer.valueOf(result.toString());
    }
105191
@Override
	public int getPartition(DatumWrapper<ITuple> key, NullWritable value, int numPartitions) {
		if(numPartitions == 1) {
			// in this case the schema is not checked if it's valid
			return 0;
		} else {
			ITuple tuple = key.datum();
			String sourceName = tuple.getSchema().getName();
			Integer schemaId = tupleMRConfig.getSchemaIdByName(sourceName);
			if(schemaId == null) {
				throw new RuntimeException("Schema name '" + sourceName
				    + "' is unknown. Known schemas are : "
				    + tupleMRConfig.getIntermediateSchemaNames());
			}
			int[] fieldsToPartition = serInfo.getPartitionFieldsIndexes().get(schemaId);
			if(fieldsToPartition.length == 0) {
				throw new RuntimeException("Fields to partition is 0. Something has been wrongly configured.");
			}
			return (partialHashCode(tuple, fieldsToPartition) & Integer.MAX_VALUE)
			    % numPartitions;
		}
	}
1007859
@Override
    public int getInContainerRowCount(BBContainer tupleData) {
        // according to TableOutputStream.cpp:TupleOutputStream::endRows() the row count is
        // at offset 4 (second integer)
        ByteBuffer bb = tupleData.b().duplicate();
        bb.position(getHeaderSize());
        bb.getInt(); // skip first four (partition id)

        return bb.getInt();
    }
366939
public java.util.List<String> getPartitionSerialList() {
        if (partitionSerialList == null) {
            partitionSerialList = new com.amazonaws.internal.SdkInternalList<String>();
        }
        return partitionSerialList;
    }
702545
List<PartitionInstance> getAbsent() {
    List<PartitionInstance> absentPartitions = new ArrayList<>();
    for (Map.Entry<Integer, List<PartitionInstance>> e : nodeIndexToHostedPartitions.entrySet()) {
      if (e.getKey() < 0) {
        absentPartitions.addAll(e.getValue());
      }
    }
    return absentPartitions;
  }
135897
private boolean reachEndRange() {
        // current row key
        ByteBuffer rowKey;

        if (keyValidator instanceof CompositeType) {
            ByteBuffer[] keys = new ByteBuffer[partitionBoundColumns.size()];
            for (int i = 0; i < partitionBoundColumns.size(); i++) {
                keys[i] = partitionBoundColumns.get(i).value.duplicate();
            }

            rowKey = CompositeType.build(keys);
        } else {
            rowKey = partitionBoundColumns.get(0).value;
        }

        String endToken = String.valueOf(split.getEndToken());
        String currentToken = partitioner.getToken(rowKey).toString();

        return endToken.equals(currentToken);
    }
343868
int getNumOccupiedMemorySegments() {
		// either the number of memory segments, or one for spilling
		final int numPartitionBuffers = this.partitionBuffers != null ?
				this.partitionBuffers.length
				: this.buildSideWriteBuffer.getNumOccupiedMemorySegments();
		return numPartitionBuffers + bucketArea.buckets.length + bucketArea.numOverflowSegments;
	}
80648
private long[] fetchPerPartitionTxnId() {
        ZooKeeper zk = VoltDB.instance().getHostMessenger().getZK();
        byte partitionTxnIdsBytes[] = null;
        try {
            partitionTxnIdsBytes = zk.getData(VoltZK.perPartitionTxnIds, false, null);
        } catch (KeeperException.NoNodeException e){return null;}//Can be no node if the cluster was never restored
        catch (Exception e) {
            VoltDB.crashLocalVoltDB("Error retrieving per partition txn ids", true, e);
        }
        ByteBuffer buf = ByteBuffer.wrap(partitionTxnIdsBytes);

        int count = buf.getInt();
        Long partitionTxnId = null;
        long partitionTxnIds[] = new long[count];
        for (int ii = 0; ii < count; ii++) {
            long txnId = buf.getLong();
            partitionTxnIds[ii] = txnId;
            int partitionId = TxnEgo.getPartitionId(txnId);
            if (partitionId == m_partitionId) {
                partitionTxnId = txnId;
                continue;
            }
        }
        if (partitionTxnId != null) {
            return partitionTxnIds;
        }
        return null;
    }
367970
public int first() {
    assertNonEmpty();
    short firstKey = keys[0];
    Container container = values[0];
    return firstKey << 16 | container.first();
  }
418569
private double getPartitionFunction() {
    if (partitionFunction != -1.0) {
      return partitionFunction;
    }

    partitionFunction = 0.0;
    Iterator<Outcome> outcomeIterator = outcomeIterator();
    while (outcomeIterator.hasNext()) {
      partitionFunction += outcomeIterator.next().getProbability();
    }
    return partitionFunction;
  }
1249162
public int getTotalCreatedConnections(){
		int total=0;
		for (int i=0; i < this.partitionCount && this.partitions[i] != null; i++){
			total+=this.partitions[i].getCreatedConnections();
		}
		return total;
	}
304807
def _spill(self):
        
        global MemoryBytesSpilled, DiskBytesSpilled
        path = self._get_spill_dir(self.spills)
        if not os.path.exists(path):
            os.makedirs(path)

        used_memory = get_used_memory()
        if not self.pdata:
            # The data has not been partitioned, it will iterator the
            # dataset once, write them into different files, has no
            # additional memory. It only called when the memory goes
            # above limit at the first time.

            # open all the files for writing
            streams = [open(os.path.join(path, str(i)), 'wb')
                       for i in range(self.partitions)]

            for k, v in self.data.items():
                h = self._partition(k)
                # put one item in batch, make it compatible with load_stream
                # it will increase the memory if dump them in batch
                self.serializer.dump_stream([(k, v)], streams[h])

            for s in streams:
                DiskBytesSpilled += s.tell()
                s.close()

            self.data.clear()
            self.pdata.extend([{} for i in range(self.partitions)])

        else:
            for i in range(self.partitions):
                p = os.path.join(path, str(i))
                with open(p, "wb") as f:
                    # dump items in batch
                    self.serializer.dump_stream(iter(self.pdata[i].items()), f)
                self.pdata[i].clear()
                DiskBytesSpilled += os.path.getsize(p)

        self.spills += 1
        gc.collect()  # release the memory as much as possible
        MemoryBytesSpilled += max(used_memory - get_used_memory(), 0) << 20