You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/05/27 17:01:23 UTC
[5/7] flink git commit: [tests] Consolidate manually triggered tests
into a dedicated package
[tests] Consolidate manually triggered tests into a dedicated package
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3061eb12
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3061eb12
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3061eb12
Branch: refs/heads/master
Commit: 3061eb126a69319a4ad45995f66f98c58dc23fdc
Parents: e96f279
Author: Stephan Ewen <se...@apache.org>
Authored: Wed May 27 11:27:56 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed May 27 16:36:02 2015 +0200
----------------------------------------------------------------------
.../sort/MassiveStringSortingITCase.java | 388 ------------------
.../sort/MassiveStringValueSortingITCase.java | 402 ------------------
.../flink/test/manual/MassiveStringSorting.java | 391 ++++++++++++++++++
.../test/manual/MassiveStringValueSorting.java | 403 +++++++++++++++++++
.../test/manual/NotSoMiniClusterIterations.java | 117 ++++++
.../apache/flink/test/manual/package-info.java | 24 ++
.../java/org/apache/flink/test/misc/CCTest.java | 115 ------
7 files changed, 935 insertions(+), 905 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java
deleted file mode 100644
index b96fdf2..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringSortingITCase.java
+++ /dev/null
@@ -1,388 +0,0 @@
-/*
- * 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.flink.runtime.operators.sort;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileReader;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.util.Random;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.base.StringComparator;
-import org.apache.flink.api.common.typeutils.base.StringSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
-import org.apache.flink.api.java.typeutils.TypeInfoParser;
-import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.operators.testutils.DummyInvokable;
-import org.apache.flink.util.MutableObjectIterator;
-import org.junit.Assert;
-
-public class MassiveStringSortingITCase {
-
- private static final long SEED = 347569784659278346L;
-
-
- public void testStringSorting() {
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<String> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TypeSerializer<String> serializer = StringSerializer.INSTANCE;
- TypeComparator<String> comparator = new StringComparator(true);
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<String> inputIterator = new StringReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<String>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<String>(serializer, String.class), comparator, 1.0, 4, 0.8f);
-
- MutableObjectIterator<String> sortedData = sorter.getIterator();
-
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- String next;
-
- while ((next = verifyReader.readLine()) != null) {
- String nextFromStratoSort = sortedData.next("");
-
- Assert.assertNotNull(nextFromStratoSort);
- Assert.assertEquals(next, nextFromStratoSort);
- }
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- @SuppressWarnings("unchecked")
- public void testStringTuplesSorting() {
- final int NUM_STRINGS = 300000;
-
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<Tuple2<String, String[]>> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TupleTypeInfo<Tuple2<String, String[]>> typeInfo = (TupleTypeInfo<Tuple2<String, String[]>>) (TupleTypeInfo<?>) TypeInfoParser.parse("Tuple2<String, String[]>");
-
- TypeSerializer<Tuple2<String, String[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
- TypeComparator<Tuple2<String, String[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<Tuple2<String, String[]>> inputIterator = new StringTupleReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<Tuple2<String, String[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<Tuple2<String, String[]>>(serializer, (Class<Tuple2<String, String[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
-
-
-
- // use this part to verify that all if good when sorting in memory
-
-// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
-// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
-//
-// {
-// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
-// while ((wi = inputIterator.next(wi)) != null) {
-// Assert.assertTrue(nks.write(wi));
-// }
-//
-// new QuickSort().sort(nks);
-// }
-//
-// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
-
- MutableObjectIterator<Tuple2<String, String[]>> sortedData = sorter.getIterator();
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- MutableObjectIterator<Tuple2<String, String[]>> verifyIterator = new StringTupleReaderMutableObjectIterator(verifyReader);
-
- Tuple2<String, String[]> next = new Tuple2<String, String[]>("", new String[0]);
- Tuple2<String, String[]> nextFromStratoSort = new Tuple2<String, String[]>("", new String[0]);
-
- int num = 0;
-
- while ((next = verifyIterator.next(next)) != null) {
- num++;
-
- nextFromStratoSort = sortedData.next(nextFromStratoSort);
- Assert.assertNotNull(nextFromStratoSort);
-
- Assert.assertEquals(next.f0, nextFromStratoSort.f0);
- Assert.assertArrayEquals(next.f1, nextFromStratoSort.f1);
- }
-
- Assert.assertNull(sortedData.next(nextFromStratoSort));
- Assert.assertEquals(NUM_STRINGS, num);
-
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private static final class StringReaderMutableObjectIterator implements MutableObjectIterator<String> {
-
- private final BufferedReader reader;
-
- public StringReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public String next(String reuse) throws IOException {
- return reader.readLine();
- }
-
- @Override
- public String next() throws IOException {
- return reader.readLine();
- }
- }
-
- private static final class StringTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<String, String[]>> {
-
- private final BufferedReader reader;
-
- public StringTupleReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public Tuple2<String, String[]> next(Tuple2<String, String[]> reuse) throws IOException {
- String line = reader.readLine();
- if (line == null) {
- return null;
- }
-
- String[] parts = line.split(" ");
- reuse.f0 = parts[0];
- reuse.f1 = parts;
- return reuse;
- }
-
- @Override
- public Tuple2<String, String[]> next() throws IOException {
- return next(new Tuple2<String, String[]>());
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
- final int resetValue = prefix.length();
-
- bld.append(prefix);
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(resetValue);
-
- int len = rnd.nextInt(20) + 300;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
-
- String str = bld.toString();
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(0);
-
- int numComps = rnd.nextInt(5) + 1;
-
- for (int z = 0; z < numComps; z++) {
- if (z > 0) {
- bld.append(' ');
- }
- bld.append(prefix);
-
- int len = rnd.nextInt(20) + 10;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
- }
-
- String str = bld.toString();
-
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- // --------------------------------------------------------------------------------------------
-
- public static void main(String[] args) {
- new MassiveStringSortingITCase().testStringSorting();
- new MassiveStringSortingITCase().testStringTuplesSorting();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java
deleted file mode 100644
index 5265cdb..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MassiveStringValueSortingITCase.java
+++ /dev/null
@@ -1,402 +0,0 @@
-/*
- * 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.flink.runtime.operators.sort;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileReader;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.util.Random;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
-import org.apache.flink.api.java.typeutils.TypeInfoParser;
-import org.apache.flink.api.java.typeutils.runtime.CopyableValueComparator;
-import org.apache.flink.api.java.typeutils.runtime.CopyableValueSerializer;
-import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.operators.testutils.DummyInvokable;
-import org.apache.flink.types.StringValue;
-import org.apache.flink.util.MutableObjectIterator;
-import org.junit.Assert;
-
-public class MassiveStringValueSortingITCase {
-
- private static final long SEED = 347569784659278346L;
-
- public void testStringValueSorting() {
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<StringValue> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TypeSerializer<StringValue> serializer = new CopyableValueSerializer<StringValue>(StringValue.class);
- TypeComparator<StringValue> comparator = new CopyableValueComparator<StringValue>(true, StringValue.class);
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<StringValue> inputIterator = new StringValueReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<StringValue>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<StringValue>(serializer, StringValue.class), comparator, 1.0, 4, 0.8f);
-
- MutableObjectIterator<StringValue> sortedData = sorter.getIterator();
-
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- String nextVerify;
- StringValue nextFromFlinkSort = new StringValue();
-
- while ((nextVerify = verifyReader.readLine()) != null) {
- nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
-
- Assert.assertNotNull(nextFromFlinkSort);
- Assert.assertEquals(nextVerify, nextFromFlinkSort.getValue());
- }
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- @SuppressWarnings("unchecked")
- public void testStringValueTuplesSorting() {
- final int NUM_STRINGS = 300000;
-
- File input = null;
- File sorted = null;
-
- try {
- // the source file
- input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
-
- // the sorted file
- sorted = File.createTempFile("sorted_strings", "txt");
-
- String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
-
- Process p = null;
- try {
- p = Runtime.getRuntime().exec(command);
- int retCode = p.waitFor();
- if (retCode != 0) {
- throw new Exception("Command failed with return code " + retCode);
- }
- p = null;
- } finally {
- if (p != null) {
- p.destroy();
- }
- }
-
- // sort the data
- UnilateralSortMerger<Tuple2<StringValue, StringValue[]>> sorter = null;
- BufferedReader reader = null;
- BufferedReader verifyReader = null;
-
- try {
- MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
- IOManager ioMan = new IOManagerAsync();
-
- TupleTypeInfo<Tuple2<StringValue, StringValue[]>> typeInfo = (TupleTypeInfo<Tuple2<StringValue, StringValue[]>>) (TupleTypeInfo<?>)
- TypeInfoParser.parse("Tuple2<org.apache.flink.types.StringValue, org.apache.flink.types.StringValue[]>");
-
- TypeSerializer<Tuple2<StringValue, StringValue[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
- TypeComparator<Tuple2<StringValue, StringValue[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
-
- reader = new BufferedReader(new FileReader(input));
- MutableObjectIterator<Tuple2<StringValue, StringValue[]>> inputIterator = new StringValueTupleReaderMutableObjectIterator(reader);
-
- sorter = new UnilateralSortMerger<Tuple2<StringValue, StringValue[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
- new RuntimeSerializerFactory<Tuple2<StringValue, StringValue[]>>(serializer, (Class<Tuple2<StringValue, StringValue[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
-
-
-
- // use this part to verify that all if good when sorting in memory
-
-// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
-// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
-//
-// {
-// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
-// while ((wi = inputIterator.next(wi)) != null) {
-// Assert.assertTrue(nks.write(wi));
-// }
-//
-// new QuickSort().sort(nks);
-// }
-//
-// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
-
- MutableObjectIterator<Tuple2<StringValue, StringValue[]>> sortedData = sorter.getIterator();
- reader.close();
-
- // verify
- verifyReader = new BufferedReader(new FileReader(sorted));
- MutableObjectIterator<Tuple2<StringValue, StringValue[]>> verifyIterator = new StringValueTupleReaderMutableObjectIterator(verifyReader);
-
- Tuple2<StringValue, StringValue[]> nextVerify = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
- Tuple2<StringValue, StringValue[]> nextFromFlinkSort = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
-
- int num = 0;
-
- while ((nextVerify = verifyIterator.next(nextVerify)) != null) {
- num++;
-
- nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
- Assert.assertNotNull(nextFromFlinkSort);
-
- Assert.assertEquals(nextVerify.f0, nextFromFlinkSort.f0);
- Assert.assertArrayEquals(nextVerify.f1, nextFromFlinkSort.f1);
- }
-
- Assert.assertNull(sortedData.next(nextFromFlinkSort));
- Assert.assertEquals(NUM_STRINGS, num);
-
- }
- finally {
- if (reader != null) {
- reader.close();
- }
- if (verifyReader != null) {
- verifyReader.close();
- }
- if (sorter != null) {
- sorter.close();
- }
- }
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail(e.getMessage());
- }
- finally {
- if (input != null) {
- input.delete();
- }
- if (sorted != null) {
- sorted.delete();
- }
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private static final class StringValueReaderMutableObjectIterator implements MutableObjectIterator<StringValue> {
-
- private final BufferedReader reader;
-
- public StringValueReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public StringValue next(StringValue reuse) throws IOException {
- String line = reader.readLine();
-
- if (line == null) {
- return null;
- }
-
- reuse.setValue(line);
- return reuse;
- }
-
- @Override
- public StringValue next() throws IOException {
- return next(new StringValue());
- }
- }
-
- private static final class StringValueTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<StringValue, StringValue[]>> {
-
- private final BufferedReader reader;
-
- public StringValueTupleReaderMutableObjectIterator(BufferedReader reader) {
- this.reader = reader;
- }
-
- @Override
- public Tuple2<StringValue, StringValue[]> next(Tuple2<StringValue, StringValue[]> reuse) throws IOException {
- String line = reader.readLine();
- if (line == null) {
- return null;
- }
-
- String[] parts = line.split(" ");
- reuse.f0.setValue(parts[0]);
- reuse.f1 = new StringValue[parts.length];
-
- for (int i = 0; i < parts.length; i++) {
- reuse.f1[i] = new StringValue(parts[i]);
- }
-
- return reuse;
- }
-
- @Override
- public Tuple2<StringValue, StringValue[]> next() throws IOException {
- return next(new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]));
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
- final int resetValue = prefix.length();
-
- bld.append(prefix);
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(resetValue);
-
- int len = rnd.nextInt(20) + 300;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
-
- String str = bld.toString();
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
- final Random rnd = new Random(SEED);
-
- final StringBuilder bld = new StringBuilder();
-
- File f = File.createTempFile("strings", "txt");
- BufferedWriter wrt = null;
- try {
- wrt = new BufferedWriter(new FileWriter(f));
-
- for (int i = 0 ; i < numStrings; i++) {
- bld.setLength(0);
-
- int numComps = rnd.nextInt(5) + 1;
-
- for (int z = 0; z < numComps; z++) {
- if (z > 0) {
- bld.append(' ');
- }
- bld.append(prefix);
-
- int len = rnd.nextInt(20) + 10;
- for (int k = 0; k < len; k++) {
- char c = (char) (rnd.nextInt(80) + 40);
- bld.append(c);
- }
- }
-
- String str = bld.toString();
-
- wrt.write(str);
- wrt.newLine();
- }
- } finally {
- wrt.close();
- }
-
- return f;
- }
-
- // --------------------------------------------------------------------------------------------
-
- public static void main(String[] args) {
- new MassiveStringValueSortingITCase().testStringValueSorting();
- new MassiveStringValueSortingITCase().testStringValueTuplesSorting();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
new file mode 100644
index 0000000..2da29e8
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
@@ -0,0 +1,391 @@
+/*
+ * 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.flink.test.manual;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Random;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.StringComparator;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.util.MutableObjectIterator;
+import org.junit.Assert;
+
+public class MassiveStringSorting {
+
+ private static final long SEED = 347569784659278346L;
+
+
+ public void testStringSorting() {
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<String> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TypeSerializer<String> serializer = StringSerializer.INSTANCE;
+ TypeComparator<String> comparator = new StringComparator(true);
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<String> inputIterator = new StringReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<String>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<String>(serializer, String.class), comparator, 1.0, 4, 0.8f);
+
+ MutableObjectIterator<String> sortedData = sorter.getIterator();
+
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ String next;
+
+ while ((next = verifyReader.readLine()) != null) {
+ String nextFromStratoSort = sortedData.next("");
+
+ Assert.assertNotNull(nextFromStratoSort);
+ Assert.assertEquals(next, nextFromStratoSort);
+ }
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testStringTuplesSorting() {
+ final int NUM_STRINGS = 300000;
+
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<Tuple2<String, String[]>> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TupleTypeInfo<Tuple2<String, String[]>> typeInfo = (TupleTypeInfo<Tuple2<String, String[]>>)
+ TypeInfoParser.<Tuple2<String, String[]>>parse("Tuple2<String, String[]>");
+
+ TypeSerializer<Tuple2<String, String[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
+ TypeComparator<Tuple2<String, String[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<Tuple2<String, String[]>> inputIterator = new StringTupleReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<Tuple2<String, String[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<Tuple2<String, String[]>>(serializer, (Class<Tuple2<String, String[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
+
+
+
+ // use this part to verify that all if good when sorting in memory
+
+// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
+// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
+//
+// {
+// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
+// while ((wi = inputIterator.next(wi)) != null) {
+// Assert.assertTrue(nks.write(wi));
+// }
+//
+// new QuickSort().sort(nks);
+// }
+//
+// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
+
+ MutableObjectIterator<Tuple2<String, String[]>> sortedData = sorter.getIterator();
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ MutableObjectIterator<Tuple2<String, String[]>> verifyIterator = new StringTupleReaderMutableObjectIterator(verifyReader);
+
+ Tuple2<String, String[]> next = new Tuple2<String, String[]>("", new String[0]);
+ Tuple2<String, String[]> nextFromStratoSort = new Tuple2<String, String[]>("", new String[0]);
+
+ int num = 0;
+
+ while ((next = verifyIterator.next(next)) != null) {
+ num++;
+
+ nextFromStratoSort = sortedData.next(nextFromStratoSort);
+ Assert.assertNotNull(nextFromStratoSort);
+
+ Assert.assertEquals(next.f0, nextFromStratoSort.f0);
+ Assert.assertArrayEquals(next.f1, nextFromStratoSort.f1);
+ }
+
+ Assert.assertNull(sortedData.next(nextFromStratoSort));
+ Assert.assertEquals(NUM_STRINGS, num);
+
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static final class StringReaderMutableObjectIterator implements MutableObjectIterator<String> {
+
+ private final BufferedReader reader;
+
+ public StringReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public String next(String reuse) throws IOException {
+ return reader.readLine();
+ }
+
+ @Override
+ public String next() throws IOException {
+ return reader.readLine();
+ }
+ }
+
+ private static final class StringTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<String, String[]>> {
+
+ private final BufferedReader reader;
+
+ public StringTupleReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public Tuple2<String, String[]> next(Tuple2<String, String[]> reuse) throws IOException {
+ String line = reader.readLine();
+ if (line == null) {
+ return null;
+ }
+
+ String[] parts = line.split(" ");
+ reuse.f0 = parts[0];
+ reuse.f1 = parts;
+ return reuse;
+ }
+
+ @Override
+ public Tuple2<String, String[]> next() throws IOException {
+ return next(new Tuple2<String, String[]>());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+ final int resetValue = prefix.length();
+
+ bld.append(prefix);
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(resetValue);
+
+ int len = rnd.nextInt(20) + 300;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+
+ String str = bld.toString();
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(0);
+
+ int numComps = rnd.nextInt(5) + 1;
+
+ for (int z = 0; z < numComps; z++) {
+ if (z > 0) {
+ bld.append(' ');
+ }
+ bld.append(prefix);
+
+ int len = rnd.nextInt(20) + 10;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+ }
+
+ String str = bld.toString();
+
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static void main(String[] args) {
+ new MassiveStringSorting().testStringSorting();
+ new MassiveStringSorting().testStringTuplesSorting();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
new file mode 100644
index 0000000..bc65ab9
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
@@ -0,0 +1,403 @@
+/*
+ * 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.flink.test.manual;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Random;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.api.java.typeutils.runtime.CopyableValueComparator;
+import org.apache.flink.api.java.typeutils.runtime.CopyableValueSerializer;
+import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.types.StringValue;
+import org.apache.flink.util.MutableObjectIterator;
+import org.junit.Assert;
+
+public class MassiveStringValueSorting {
+
+ private static final long SEED = 347569784659278346L;
+
+ public void testStringValueSorting() {
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStrings(300000, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<StringValue> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TypeSerializer<StringValue> serializer = new CopyableValueSerializer<StringValue>(StringValue.class);
+ TypeComparator<StringValue> comparator = new CopyableValueComparator<StringValue>(true, StringValue.class);
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<StringValue> inputIterator = new StringValueReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<StringValue>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<StringValue>(serializer, StringValue.class), comparator, 1.0, 4, 0.8f);
+
+ MutableObjectIterator<StringValue> sortedData = sorter.getIterator();
+
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ String nextVerify;
+ StringValue nextFromFlinkSort = new StringValue();
+
+ while ((nextVerify = verifyReader.readLine()) != null) {
+ nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
+
+ Assert.assertNotNull(nextFromFlinkSort);
+ Assert.assertEquals(nextVerify, nextFromFlinkSort.getValue());
+ }
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testStringValueTuplesSorting() {
+ final int NUM_STRINGS = 300000;
+
+ File input = null;
+ File sorted = null;
+
+ try {
+ // the source file
+ input = generateFileWithStringTuples(NUM_STRINGS, "http://some-uri.com/that/is/a/common/prefix/to/all");
+
+ // the sorted file
+ sorted = File.createTempFile("sorted_strings", "txt");
+
+ String[] command = {"/bin/bash","-c","export LC_ALL=\"C\" && cat \"" + input.getAbsolutePath() + "\" | sort > \"" + sorted.getAbsolutePath() + "\""};
+
+ Process p = null;
+ try {
+ p = Runtime.getRuntime().exec(command);
+ int retCode = p.waitFor();
+ if (retCode != 0) {
+ throw new Exception("Command failed with return code " + retCode);
+ }
+ p = null;
+ } finally {
+ if (p != null) {
+ p.destroy();
+ }
+ }
+
+ // sort the data
+ UnilateralSortMerger<Tuple2<StringValue, StringValue[]>> sorter = null;
+ BufferedReader reader = null;
+ BufferedReader verifyReader = null;
+
+ try {
+ MemoryManager mm = new DefaultMemoryManager(1024 * 1024, 1);
+ IOManager ioMan = new IOManagerAsync();
+
+ TupleTypeInfo<Tuple2<StringValue, StringValue[]>> typeInfo = (TupleTypeInfo<Tuple2<StringValue, StringValue[]>>)
+ TypeInfoParser.<Tuple2<StringValue, StringValue[]>>parse("Tuple2<org.apache.flink.types.StringValue, org.apache.flink.types.StringValue[]>");
+
+ TypeSerializer<Tuple2<StringValue, StringValue[]>> serializer = typeInfo.createSerializer(new ExecutionConfig());
+ TypeComparator<Tuple2<StringValue, StringValue[]>> comparator = typeInfo.createComparator(new int[] { 0 }, new boolean[] { true }, 0, new ExecutionConfig());
+
+ reader = new BufferedReader(new FileReader(input));
+ MutableObjectIterator<Tuple2<StringValue, StringValue[]>> inputIterator = new StringValueTupleReaderMutableObjectIterator(reader);
+
+ sorter = new UnilateralSortMerger<Tuple2<StringValue, StringValue[]>>(mm, ioMan, inputIterator, new DummyInvokable(),
+ new RuntimeSerializerFactory<Tuple2<StringValue, StringValue[]>>(serializer, (Class<Tuple2<StringValue, StringValue[]>>) (Class<?>) Tuple2.class), comparator, 1.0, 4, 0.8f);
+
+
+
+ // use this part to verify that all if good when sorting in memory
+
+// List<MemorySegment> memory = mm.allocatePages(new DummyInvokable(), mm.computeNumberOfPages(1024*1024*1024));
+// NormalizedKeySorter<Tuple2<String, String[]>> nks = new NormalizedKeySorter<Tuple2<String,String[]>>(serializer, comparator, memory);
+//
+// {
+// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new String[0]);
+// while ((wi = inputIterator.next(wi)) != null) {
+// Assert.assertTrue(nks.write(wi));
+// }
+//
+// new QuickSort().sort(nks);
+// }
+//
+// MutableObjectIterator<Tuple2<String, String[]>> sortedData = nks.getIterator();
+
+ MutableObjectIterator<Tuple2<StringValue, StringValue[]>> sortedData = sorter.getIterator();
+ reader.close();
+
+ // verify
+ verifyReader = new BufferedReader(new FileReader(sorted));
+ MutableObjectIterator<Tuple2<StringValue, StringValue[]>> verifyIterator = new StringValueTupleReaderMutableObjectIterator(verifyReader);
+
+ Tuple2<StringValue, StringValue[]> nextVerify = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
+ Tuple2<StringValue, StringValue[]> nextFromFlinkSort = new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]);
+
+ int num = 0;
+
+ while ((nextVerify = verifyIterator.next(nextVerify)) != null) {
+ num++;
+
+ nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
+ Assert.assertNotNull(nextFromFlinkSort);
+
+ Assert.assertEquals(nextVerify.f0, nextFromFlinkSort.f0);
+ Assert.assertArrayEquals(nextVerify.f1, nextFromFlinkSort.f1);
+ }
+
+ Assert.assertNull(sortedData.next(nextFromFlinkSort));
+ Assert.assertEquals(NUM_STRINGS, num);
+
+ }
+ finally {
+ if (reader != null) {
+ reader.close();
+ }
+ if (verifyReader != null) {
+ verifyReader.close();
+ }
+ if (sorter != null) {
+ sorter.close();
+ }
+ }
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ finally {
+ if (input != null) {
+ input.delete();
+ }
+ if (sorted != null) {
+ sorted.delete();
+ }
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static final class StringValueReaderMutableObjectIterator implements MutableObjectIterator<StringValue> {
+
+ private final BufferedReader reader;
+
+ public StringValueReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public StringValue next(StringValue reuse) throws IOException {
+ String line = reader.readLine();
+
+ if (line == null) {
+ return null;
+ }
+
+ reuse.setValue(line);
+ return reuse;
+ }
+
+ @Override
+ public StringValue next() throws IOException {
+ return next(new StringValue());
+ }
+ }
+
+ private static final class StringValueTupleReaderMutableObjectIterator implements MutableObjectIterator<Tuple2<StringValue, StringValue[]>> {
+
+ private final BufferedReader reader;
+
+ public StringValueTupleReaderMutableObjectIterator(BufferedReader reader) {
+ this.reader = reader;
+ }
+
+ @Override
+ public Tuple2<StringValue, StringValue[]> next(Tuple2<StringValue, StringValue[]> reuse) throws IOException {
+ String line = reader.readLine();
+ if (line == null) {
+ return null;
+ }
+
+ String[] parts = line.split(" ");
+ reuse.f0.setValue(parts[0]);
+ reuse.f1 = new StringValue[parts.length];
+
+ for (int i = 0; i < parts.length; i++) {
+ reuse.f1[i] = new StringValue(parts[i]);
+ }
+
+ return reuse;
+ }
+
+ @Override
+ public Tuple2<StringValue, StringValue[]> next() throws IOException {
+ return next(new Tuple2<StringValue, StringValue[]>(new StringValue(), new StringValue[0]));
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private File generateFileWithStrings(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+ final int resetValue = prefix.length();
+
+ bld.append(prefix);
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(resetValue);
+
+ int len = rnd.nextInt(20) + 300;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+
+ String str = bld.toString();
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ private File generateFileWithStringTuples(int numStrings, String prefix) throws IOException {
+ final Random rnd = new Random(SEED);
+
+ final StringBuilder bld = new StringBuilder();
+
+ File f = File.createTempFile("strings", "txt");
+ BufferedWriter wrt = null;
+ try {
+ wrt = new BufferedWriter(new FileWriter(f));
+
+ for (int i = 0 ; i < numStrings; i++) {
+ bld.setLength(0);
+
+ int numComps = rnd.nextInt(5) + 1;
+
+ for (int z = 0; z < numComps; z++) {
+ if (z > 0) {
+ bld.append(' ');
+ }
+ bld.append(prefix);
+
+ int len = rnd.nextInt(20) + 10;
+ for (int k = 0; k < len; k++) {
+ char c = (char) (rnd.nextInt(80) + 40);
+ bld.append(c);
+ }
+ }
+
+ String str = bld.toString();
+
+ wrt.write(str);
+ wrt.newLine();
+ }
+ } finally {
+ wrt.close();
+ }
+
+ return f;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static void main(String[] args) {
+ new MassiveStringValueSorting().testStringValueSorting();
+ new MassiveStringValueSorting().testStringValueTuplesSorting();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java b/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java
new file mode 100644
index 0000000..d340f57
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java
@@ -0,0 +1,117 @@
+/*
+ * 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.flink.test.manual;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.aggregation.Aggregations;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.examples.java.graph.ConnectedComponents;
+import org.apache.flink.examples.java.graph.util.ConnectedComponentsData;
+import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+
+import static org.junit.Assert.fail;
+
+/**
+ * This test starts a mini cluster with 100 task managers and runs connected components
+ * with a parallelism of 100.
+ */
+public class NotSoMiniClusterIterations {
+
+ private static final int PARALLELISM = 100;
+
+ public static void main(String[] args) {
+ if ((Runtime.getRuntime().maxMemory() >>> 20) < 5000) {
+ throw new RuntimeException("This test program needs to run with at least 5GB of heap space.");
+ }
+
+ LocalFlinkMiniCluster cluster = null;
+
+ try {
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, PARALLELISM);
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 8);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 1000);
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SEGMENT_SIZE_KEY, 8 * 1024);
+
+ config.setInteger("taskmanager.net.server.numThreads", 1);
+ config.setInteger("taskmanager.net.client.numThreads", 1);
+
+ cluster = new LocalFlinkMiniCluster(config, false);
+
+ runConnectedComponents(cluster.getJobManagerRPCPort());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ finally {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+ }
+
+ private static void runConnectedComponents(int jmPort) throws Exception {
+
+ ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("localhost", jmPort);
+ env.setParallelism(PARALLELISM);
+ env.getConfig().disableSysoutLogging();
+
+ // read vertex and edge data
+ DataSet<Long> vertices = ConnectedComponentsData.getDefaultVertexDataSet(env)
+ .rebalance();
+
+ DataSet<Tuple2<Long, Long>> edges = ConnectedComponentsData.getDefaultEdgeDataSet(env)
+ .rebalance()
+ .flatMap(new ConnectedComponents.UndirectEdge());
+
+ // assign the initial components (equal to the vertex id)
+ DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices
+ .map(new ConnectedComponents.DuplicateValue<Long>());
+
+ // open a delta iteration
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
+ verticesWithInitialId.iterateDelta(verticesWithInitialId, 100, 0);
+
+ // apply the step logic: join with the edges, select the minimum neighbor,
+ // update if the component of the candidate is smaller
+ DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset().join(edges)
+ .where(0).equalTo(0)
+ .with(new ConnectedComponents.NeighborWithComponentIDJoin())
+
+ .groupBy(0).aggregate(Aggregations.MIN, 1)
+
+ .join(iteration.getSolutionSet())
+ .where(0).equalTo(0)
+ .with(new ConnectedComponents.ComponentIdFilter());
+
+ // close the delta iteration (delta and new workset are identical)
+ DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
+
+ result.output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
+
+ env.execute();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java b/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java
new file mode 100644
index 0000000..893f3cc
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/package-info.java
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+/**
+ * This package contains various tests that are not automatically executed, but
+ * need to be manually invoked, because they are extremely heavy of require larger-than-usual
+ * JVMs.
+ */
+package org.apache.flink.test.manual;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/3061eb12/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java b/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
deleted file mode 100644
index cca6461..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/CCTest.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * 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.flink.test.misc;
-
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.aggregation.Aggregations;
-import org.apache.flink.api.java.io.DiscardingOutputFormat;
-import org.apache.flink.api.java.operators.DeltaIteration;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.examples.java.graph.ConnectedComponents;
-import org.apache.flink.examples.java.graph.util.ConnectedComponentsData;
-import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
-
-import static org.junit.Assert.fail;
-
-public class CCTest {
-
- private static final int PARALLELISM = 100;
-
- public static void main(String[] args) {
- if ((Runtime.getRuntime().maxMemory() >>> 20) < 5000) {
- throw new RuntimeException("This est program needs to run with at least 5GB of heap space.");
- }
-
-
- LocalFlinkMiniCluster cluster = null;
-
- try {
- Configuration config = new Configuration();
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, PARALLELISM);
- config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 8);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1);
- config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 1000);
- config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_BUFFER_SIZE_KEY, 8 * 1024);
-
-
- config.setInteger("taskmanager.net.server.numThreads", 1);
- config.setInteger("taskmanager.net.client.numThreads", 1);
-
- cluster = new LocalFlinkMiniCluster(config, false);
-
- runConnectedComponents(cluster.getJobManagerRPCPort());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- finally {
- if (cluster != null) {
- cluster.shutdown();
- }
- }
- }
-
- private static void runConnectedComponents(int jmPort) throws Exception {
-
- ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("localhost", jmPort);
- env.setParallelism(PARALLELISM);
- env.getConfig().disableSysoutLogging();
-
- // read vertex and edge data
- DataSet<Long> vertices = ConnectedComponentsData.getDefaultVertexDataSet(env)
- .rebalance();
-
- DataSet<Tuple2<Long, Long>> edges = ConnectedComponentsData.getDefaultEdgeDataSet(env)
- .rebalance()
- .flatMap(new ConnectedComponents.UndirectEdge());
-
- // assign the initial components (equal to the vertex id)
- DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices
- .map(new ConnectedComponents.DuplicateValue<Long>());
-
- // open a delta iteration
- DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
- verticesWithInitialId.iterateDelta(verticesWithInitialId, 100, 0);
-
- // apply the step logic: join with the edges, select the minimum neighbor,
- // update if the component of the candidate is smaller
- DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset().join(edges)
- .where(0).equalTo(0)
- .with(new ConnectedComponents.NeighborWithComponentIDJoin())
-
- .groupBy(0).aggregate(Aggregations.MIN, 1)
-
- .join(iteration.getSolutionSet())
- .where(0).equalTo(0)
- .with(new ConnectedComponents.ComponentIdFilter());
-
- // close the delta iteration (delta and new workset are identical)
- DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
-
- result.output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
-
- env.execute();
- }
-}