Github user ericl commented on a diff in the pull request:
https://github.com/apache/spark/pull/12490#discussion_r60327092
--- Diff:
core/src/main/java/org/apache/spark/util/collection/unsafe/sort/RadixSort.java
---
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection.unsafe.sort;
+
+import org.apache.spark.unsafe.Platform;
+import org.apache.spark.unsafe.array.LongArray;
+
+public class RadixSort {
+
+ /**
+ * Sorts a given array of longs using least-significant-digit radix
sort. This routine assumes
+ * you have extra space at the end of the array at least equal to the
number of records. The
+ * sort is destructive and may relocate the data positioned within the
array.
+ *
+ * @param array array of long elements followed by at least that many
empty slots.
+ * @param numRecords number of data records in the array.
+ * @param startByteIndex the first byte (in range [0, 7]) to sort each
long by, counting from the
+ * least significant byte.
+ * @param endByteIndex the last byte (in range [0, 7]) to sort each long
by, counting from the
+ * least significant byte. Must be greater than
startByteIndex.
+ * @param desc whether this is a descending (binary-order) sort.
+ * @param signed whether this is a signed (two's complement) sort.
+ *
+ * @return The starting index of the sorted data within the given array.
We return this instead
+ * of always copying the data back to position zero for
efficiency.
+ */
+ public static int sort(
+ LongArray array, int numRecords, int startByteIndex, int
endByteIndex,
+ boolean desc, boolean signed) {
+ assert startByteIndex >= 0 : "startByteIndex (" + startByteIndex + ")
should >= 0";
+ assert endByteIndex <= 7 : "endByteIndex (" + endByteIndex + ") should
<= 7";
+ assert endByteIndex > startByteIndex;
+ assert numRecords * 2 <= array.size();
+ int inIndex = 0;
+ int outIndex = numRecords;
+ if (numRecords > 0) {
+ long[][] counts = getCounts(array, numRecords, startByteIndex,
endByteIndex);
+ for (int i = startByteIndex; i <= endByteIndex; i++) {
+ if (counts[i] != null) {
+ sortAtByte(
+ array, numRecords, counts[i], i, inIndex, outIndex,
+ desc, signed && i == endByteIndex);
+ int tmp = inIndex;
+ inIndex = outIndex;
+ outIndex = tmp;
+ }
+ }
+ }
+ return inIndex;
+ }
+
+ /**
+ * Performs a partial sort by copying data into destination offsets for
each byte value at the
+ * specified byte offset.
+ *
+ * @param array array to partially sort.
+ * @param numRecords number of data records in the array.
+ * @param counts counts for each byte value. This routine destructively
modifies this array.
+ * @param byteIdx the byte in a long to sort at, counting from the least
significant byte.
+ * @param inIndex the starting index in the array where input data is
located.
+ * @param outIndex the starting index where sorted output data should be
written.
+ * @param desc whether this is a descending (binary-order) sort.
+ * @param signed whether this is a signed (two's complement) sort (only
applies to last byte).
+ */
+ private static void sortAtByte(
+ LongArray array, int numRecords, long[] counts, int byteIdx, int
inIndex, int outIndex,
+ boolean desc, boolean signed) {
+ assert counts.length == 256;
+ long[] offsets = transformCountsToOffsets(
+ counts, numRecords, array.getBaseOffset() + outIndex * 8, 8, desc,
signed);
+ Object baseObject = array.getBaseObject();
+ long baseOffset = array.getBaseOffset() + inIndex * 8;
+ long maxOffset = baseOffset + numRecords * 8;
+ for (long offset = baseOffset; offset < maxOffset; offset += 8) {
+ long value = Platform.getLong(baseObject, offset);
+ int bucket = (int)((value >>> (byteIdx * 8)) & 0xff);
+ Platform.putLong(baseObject, offsets[bucket], value);
+ offsets[bucket] += 8;
+ }
+ }
+
+ /**
+ * Computes a value histogram for each byte in the given array.
+ *
+ * @param array array to count records in.
+ * @param numRecords number of data records in the array.
+ * @param startByteIndex the first byte to compute counts for (the prior
are skipped).
+ * @param endByteIndex the last byte to compute counts for.
+ *
+ * @return an array of eight 256-byte count arrays, one for each byte
starting from the least
+ * significant byte. If the byte does not need sorting the array
will be null.
+ */
+ private static long[][] getCounts(
+ LongArray array, int numRecords, int startByteIndex, int
endByteIndex) {
+ long[][] counts = new long[8][];
+ // Optimization: do a fast pre-pass to determine which byte indices we
can skip for sorting.
+ // If all the byte values at a particular index are the same we don't
need to count it.
+ long bitwiseMax = 0;
+ long bitwiseMin = ~0L;
--- End diff --
Done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
