Author: chetanm
Date: Fri Sep 15 10:35:08 2017
New Revision: 1808443
URL: http://svn.apache.org/viewvc?rev=1808443&view=rev
Log:
OAK-6671 - Enable support for custom types in ExternalSort
Modified:
jackrabbit/oak/trunk/oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/sort/ExternalSort.java
jackrabbit/oak/trunk/oak-commons/src/test/java/org/apache/jackrabbit/oak/commons/sort/ExternalSortTest.java
Modified:
jackrabbit/oak/trunk/oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/sort/ExternalSort.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/sort/ExternalSort.java?rev=1808443&r1=1808442&r2=1808443&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/sort/ExternalSort.java
(original)
+++
jackrabbit/oak/trunk/oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/sort/ExternalSort.java
Fri Sep 15 10:35:08 2017
@@ -34,9 +34,13 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;
+import java.util.function.Function;
import java.util.zip.Deflater;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
+
+import static java.util.function.Function.identity;
+
/**
* Source copied from a publicly available library.
* @see <a
href="https://code.google.com/p/externalsortinginjava/";>https://code.google.com/p/externalsortinginjava</a>
@@ -164,7 +168,7 @@ public class ExternalSort {
* This will simply load the file by blocks of lines, then sort them
in-memory, and write the
* result to temporary files that have to be merged later. You can specify
a bound on the number
* of temporary files that will be created.
- *
+ *
* @param file
* some flat file
* @param cmp
@@ -183,8 +187,44 @@ public class ExternalSort {
* @return a list of temporary flat files
*/
public static List<File> sortInBatch(File file, Comparator<String> cmp,
- int maxtmpfiles, long maxMemory, Charset cs, File tmpdirectory,
- boolean distinct, int numHeader, boolean usegzip)
+ int maxtmpfiles, long maxMemory,
Charset cs, File tmpdirectory,
+ boolean distinct, int numHeader,
boolean usegzip)
+ throws IOException {
+ return sortInBatch(file, cmp, maxtmpfiles, maxMemory, cs,
tmpdirectory, distinct,
+ numHeader, usegzip, identity(), identity());
+ }
+
+ /**
+ * This will simply load the file by blocks of lines, then sort them
in-memory, and write the
+ * result to temporary files that have to be merged later. You can specify
a bound on the number
+ * of temporary files that will be created.
+ *
+ * @param file
+ * some flat file
+ * @param cmp
+ * string comparator
+ * @param maxtmpfiles
+ * maximal number of temporary files
+ * @param cs
+ * character set to use (can use Charset.defaultCharset())
+ * @param tmpdirectory
+ * location of the temporary files (set to null for default
location)
+ * @param distinct
+ * Pass <code>true</code> if duplicate lines should be
discarded.
+ * @param numHeader
+ * number of lines to preclude before sorting starts
+ * @param usegzip use gzip compression for the temporary files
+ * @param typeToString
+ * function to map string to custom type. User for coverting
line to custom type for the
+ * purpose of sorting
+ * @param stringToType
+ * function to map custom type to string. Used for storing sorted
content back to file
+ * @return a list of temporary flat files
+ */
+ public static <T> List<File> sortInBatch(File file, Comparator<T> cmp,
+ int maxtmpfiles, long maxMemory,
Charset cs, File tmpdirectory,
+ boolean distinct, int numHeader,
boolean usegzip,
+ Function<T, String> typeToString,
Function<String, T> stringToType)
throws IOException {
List<File> files = new ArrayList<File>();
BufferedReader fbr = new BufferedReader(new InputStreamReader(
@@ -193,7 +233,7 @@ public class ExternalSort {
long blocksize = estimateBestSizeOfBlocks(file, maxtmpfiles,
maxMemory);
try {
- List<String> tmplist = new ArrayList<String>();
+ List<T> tmplist = new ArrayList<>();
String line = "";
try {
int counter = 0;
@@ -207,7 +247,7 @@ public class ExternalSort {
counter++;
continue;
}
- tmplist.add(line);
+ tmplist.add(stringToType.apply(line));
// ram usage estimation, not
// very accurate, still more
// realistic that the simple 2 *
@@ -216,13 +256,13 @@ public class ExternalSort {
.estimatedSizeOf(line);
}
files.add(sortAndSave(tmplist, cmp, cs,
- tmpdirectory, distinct, usegzip));
+ tmpdirectory, distinct, usegzip, typeToString));
tmplist.clear();
}
} catch (EOFException oef) {
if (tmplist.size() > 0) {
files.add(sortAndSave(tmplist, cmp, cs,
- tmpdirectory, distinct, usegzip));
+ tmpdirectory, distinct, usegzip, typeToString));
tmplist.clear();
}
}
@@ -260,7 +300,7 @@ public class ExternalSort {
/**
* Sort a list and save it to a temporary file
- *
+ *
* @return the file containing the sorted data
* @param tmplist
* data to be sorted
@@ -274,8 +314,31 @@ public class ExternalSort {
* Pass <code>true</code> if duplicate lines should be
discarded.
*/
public static File sortAndSave(List<String> tmplist,
- Comparator<String> cmp, Charset cs, File tmpdirectory,
- boolean distinct, boolean usegzip) throws IOException {
+ Comparator<String> cmp, Charset cs, File
tmpdirectory,
+ boolean distinct, boolean usegzip) throws
IOException {
+ return sortAndSave(tmplist, cmp, cs, tmpdirectory, distinct, usegzip,
identity());
+ }
+
+ /**
+ * Sort a list and save it to a temporary file
+ *
+ * @return the file containing the sorted data
+ * @param tmplist
+ * data to be sorted
+ * @param cmp
+ * string comparator
+ * @param cs
+* charset to use for output (can use Charset.defaultCharset())
+ * @param tmpdirectory
+* location of the temporary files (set to null for default location)
+ * @param distinct
+ * @param typeToString
+ * function to map string to custom type. User for coverting line
to custom type for the
+ * purpose of sorting
+ */
+ public static <T> File sortAndSave(List<T> tmplist,
+ Comparator<T> cmp, Charset cs, File
tmpdirectory,
+ boolean distinct, boolean usegzip,
Function<T, String> typeToString) throws IOException {
Collections.sort(tmplist, cmp);
File newtmpfile = File.createTempFile("sortInBatch",
"flatfile", tmpdirectory);
@@ -291,12 +354,12 @@ public class ExternalSort {
}
BufferedWriter fbw = new BufferedWriter(new OutputStreamWriter(
out, cs));
- String lastLine = null;
+ T lastLine = null;
try {
- for (String r : tmplist) {
+ for (T r : tmplist) {
// Skip duplicate lines
if (!distinct || (lastLine == null || (lastLine != null &&
cmp.compare(r, lastLine) != 0))) {
- fbw.write(r);
+ fbw.write(typeToString.apply(r));
fbw.newLine();
lastLine = r;
}
@@ -370,7 +433,7 @@ public class ExternalSort {
/**
* This merges a bunch of temporary flat files
- *
+ *
* @param files
* The {@link List} of sorted {@link File}s to be merged.
* @param distinct
@@ -389,10 +452,42 @@ public class ExternalSort {
* @return The number of lines sorted. (P. Beaudoin)
* @since v0.1.4
*/
- public static int mergeSortedFiles(List<File> files, File outputfile,
- final Comparator<String> cmp, Charset cs, boolean distinct,
- boolean append, boolean usegzip) throws IOException {
- ArrayList<BinaryFileBuffer> bfbs = new ArrayList<BinaryFileBuffer>();
+ public static <T> int mergeSortedFiles(List<File> files, File outputfile,
+ final Comparator<String> cmp,
Charset cs, boolean distinct,
+ boolean append, boolean usegzip)
throws IOException {
+ return mergeSortedFiles(files, outputfile, cmp, cs, distinct, append,
usegzip, Function.identity(), Function.identity());
+ }
+
+ /**
+ * This merges a bunch of temporary flat files
+ *
+ * @param files
+ * The {@link List} of sorted {@link File}s to be merged.
+ * @param outputfile
+ * The output {@link File} to merge the results to.
+ * @param cmp
+ * The {@link Comparator} to use to compare {@link String}s.
+ * @param cs
+ * The {@link Charset} to be used for the byte to character
conversion.
+ * @param distinct
+ * Pass <code>true</code> if duplicate lines should be
discarded. ([email protected])
+ * @param append
+ * Pass <code>true</code> if result should append to {@link
File} instead of
+ * overwrite. Default to be false for overloading methods.
+ * @param usegzip
+ * assumes we used gzip compression for temporary files
+ * @param typeToString
+ * function to map string to custom type. User for coverting
line to custom type for the
+ * purpose of sorting
+ * @param stringToType
+ * function to map custom type to string. Used for storing sorted
content back to file
+ * @since v0.1.4
+ */
+ public static <T> int mergeSortedFiles(List<File> files, File outputfile,
+ final Comparator<T> cmp, Charset
cs, boolean distinct,
+ boolean append, boolean usegzip,
Function<T, String> typeToString,
+ Function<String, T> stringToType)
throws IOException {
+ ArrayList<BinaryFileBuffer<T>> bfbs = new ArrayList<>();
for (File f : files) {
final int bufferSize = 2048;
InputStream in = new FileInputStream(f);
@@ -405,12 +500,12 @@ public class ExternalSort {
cs));
}
- BinaryFileBuffer bfb = new BinaryFileBuffer(br);
+ BinaryFileBuffer<T> bfb = new BinaryFileBuffer<>(br, stringToType);
bfbs.add(bfb);
}
BufferedWriter fbw = new BufferedWriter(new OutputStreamWriter(
new FileOutputStream(outputfile, append), cs));
- int rowcounter = merge(fbw, cmp, distinct, bfbs);
+ int rowcounter = merge(fbw, cmp, distinct, bfbs, typeToString);
for (File f : files) {
f.delete();
}
@@ -428,16 +523,19 @@ public class ExternalSort {
* Pass <code>true</code> if duplicate lines should be
discarded. ([email protected])
* @param buffers
* Where the data should be read.
- * @return The number of lines sorted. (P. Beaudoin)
- *
+ * @param typeToString
+ * function to map string to custom type. User for coverting
line to custom type for the
+ * purpose of sorting
+ *
*/
- public static int merge(BufferedWriter fbw, final Comparator<String> cmp,
boolean distinct,
- List<BinaryFileBuffer> buffers) throws IOException {
- PriorityQueue<BinaryFileBuffer> pq = new
PriorityQueue<BinaryFileBuffer>(
- 11, new Comparator<BinaryFileBuffer>() {
+ public static <T> int merge(BufferedWriter fbw, final Comparator<T> cmp,
boolean distinct,
+ List<BinaryFileBuffer<T>> buffers, Function<T,
String> typeToString)
+ throws IOException {
+ PriorityQueue<BinaryFileBuffer<T>> pq = new PriorityQueue<>(
+ 11, new Comparator<BinaryFileBuffer<T>>() {
@Override
- public int compare(BinaryFileBuffer i,
- BinaryFileBuffer j) {
+ public int compare(BinaryFileBuffer<T> i,
+ BinaryFileBuffer<T> j) {
return cmp.compare(i.peek(), j.peek());
}
});
@@ -447,14 +545,14 @@ public class ExternalSort {
}
}
int rowcounter = 0;
- String lastLine = null;
+ T lastLine = null;
try {
while (pq.size() > 0) {
- BinaryFileBuffer bfb = pq.poll();
- String r = bfb.pop();
+ BinaryFileBuffer<T> bfb = pq.poll();
+ T r = bfb.pop();
// Skip duplicate lines
if (!distinct || (lastLine == null || (lastLine != null &&
cmp.compare(r, lastLine) != 0))) {
- fbw.write(r);
+ fbw.write(typeToString.apply(r));
fbw.newLine();
lastLine = r;
}
@@ -630,14 +728,16 @@ public class ExternalSort {
};
}
-class BinaryFileBuffer {
- public BufferedReader fbr;
- private String cache;
+class BinaryFileBuffer<T> {
+ public final BufferedReader fbr;
+ private final Function<String, T> stringToType;
+ private T cache;
private boolean empty;
- public BinaryFileBuffer(BufferedReader r)
+ public BinaryFileBuffer(BufferedReader r, Function<String, T> stringToType)
throws IOException {
this.fbr = r;
+ this.stringToType = stringToType;
reload();
}
@@ -647,7 +747,7 @@ class BinaryFileBuffer {
private void reload() throws IOException {
try {
- if ((this.cache = fbr.readLine()) == null) {
+ if ((this.cache = stringToType.apply(fbr.readLine())) == null) {
this.empty = true;
this.cache = null;
} else {
@@ -663,15 +763,15 @@ class BinaryFileBuffer {
this.fbr.close();
}
- public String peek() {
+ public T peek() {
if (empty()) {
return null;
}
return this.cache;
}
- public String pop() throws IOException {
- String answer = peek();
+ public T pop() throws IOException {
+ T answer = peek();
reload();
return answer;
}
Modified:
jackrabbit/oak/trunk/oak-commons/src/test/java/org/apache/jackrabbit/oak/commons/sort/ExternalSortTest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-commons/src/test/java/org/apache/jackrabbit/oak/commons/sort/ExternalSortTest.java?rev=1808443&r1=1808442&r2=1808443&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-commons/src/test/java/org/apache/jackrabbit/oak/commons/sort/ExternalSortTest.java
(original)
+++
jackrabbit/oak/trunk/oak-commons/src/test/java/org/apache/jackrabbit/oak/commons/sort/ExternalSortTest.java
Fri Sep 15 10:35:08 2017
@@ -16,6 +16,7 @@
*/
package org.apache.jackrabbit.oak.commons.sort;
+import static com.google.common.collect.Iterables.transform;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@@ -31,9 +32,17 @@ import java.nio.channels.FileChannel;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Scanner;
+import java.util.function.Function;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.collect.Iterables;
+import com.google.common.io.Files;
+import com.google.common.primitives.Ints;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -443,6 +452,90 @@ public class ExternalSortTest {
}
+ @Test
+ public void customType() throws Exception{
+ File out = folder.newFile();
+ int testCount = 1000;
+
+ List<TestLine> testLines = new ArrayList<>(1000);
+ for (int i = 0; i < testCount; i++) {
+ testLines.add(new TestLine(i + ":" + "foo-"+i));
+ }
+
+ Collections.shuffle(testLines);
+
+ Comparator<TestLine> cmp = Comparator.naturalOrder();
+ Charset charset = Charsets.UTF_8;
+
+ Function<String, TestLine> stringToType = line -> line != null ? new
TestLine(line) : null;
+ Function<TestLine, String> typeToString = tl -> tl != null ? tl.line :
null;
+
+ String testData = Joiner.on('\n').join(transform(testLines, tl ->
tl.line));
+ File testFile = folder.newFile();
+ Files.write(testData, testFile, charset);
+
+ List<File> listOfFiles = ExternalSort.sortInBatch(testFile, cmp,
+ ExternalSort.DEFAULTMAXTEMPFILES,
+ 100,
+ charset,
+ folder.newFolder(),
+ false,
+ 0,
+ false,
+ typeToString,
+ stringToType);
+
+ ExternalSort.mergeSortedFiles(listOfFiles, out, cmp,
+ charset, false, true, false, typeToString, stringToType);
+
+
+ Collections.sort(testLines);
+
+ List<TestLine> linesFromSortedFile = new ArrayList<>();
+ Files.readLines(out, charset).forEach(line ->
linesFromSortedFile.add(new TestLine(line)));
+
+ assertEquals(testLines, linesFromSortedFile);
+
+ }
+
+ private static class TestLine implements Comparable<TestLine> {
+ final String line;
+ final int value;
+
+ public TestLine(String line) {
+ this.line = line;
+ this.value = Integer.valueOf(line.substring(0, line.indexOf(':')));
+ }
+
+ @Override
+ public int compareTo(TestLine o) {
+ return Ints.compare(value, o.value);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TestLine testLine = (TestLine) o;
+
+ if (value != testLine.value) return false;
+ return line.equals(testLine.line);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = line.hashCode();
+ result = 31 * result + value;
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return line;
+ }
+ }
+
public static ArrayList<String> readLines(File f) throws IOException {
ArrayList<String> answer = new ArrayList<String>();
BufferedReader r = new BufferedReader(new FileReader(f));
