[#827] add indexed to readLogicalGraph (#828)

fixes #827

gradoop-examples/src/main/java/org/gradoop/examples/AbstractRunner.java

@@ -23,6 +23,7 @@
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.gradoop.flink.io.impl.csv.CSVDataSink;
 import org.gradoop.flink.io.impl.csv.CSVDataSource;
+import org.gradoop.flink.io.impl.csv.indexed.IndexedCSVDataSource;
 import org.gradoop.flink.io.impl.json.JSONDataSink;
 import org.gradoop.flink.io.impl.json.JSONDataSource;
 import org.gradoop.flink.model.api.epgm.GraphCollection;
@@ -70,7 +71,7 @@ protected static CommandLine parseArguments(String[] args, String className)
    * @return EPGM logical graph
    */
   @SuppressWarnings("unchecked")
-  protected static LogicalGraph readLogicalGraph(String directory) {
+  protected static LogicalGraph readLogicalGraph(String directory) throws IOException {
     return readLogicalGraph(directory, "json");
   }
 
@@ -82,7 +83,8 @@ protected static LogicalGraph readLogicalGraph(String directory) {
    * @return EPGM logical graph
    */
   @SuppressWarnings("unchecked")
-  protected static LogicalGraph readLogicalGraph(String directory, String format) {
+  protected static LogicalGraph readLogicalGraph(String directory, String format)
+    throws IOException {
     directory = appendSeparator(directory);
 
     GradoopFlinkConfig config = GradoopFlinkConfig.createConfig(getExecutionEnvironment());
@@ -92,6 +94,8 @@ protected static LogicalGraph readLogicalGraph(String directory, String format)
       return new JSONDataSource(directory, config).getLogicalGraph();
     } else if (format.equals("csv")) {
       return new CSVDataSource(directory, config).getLogicalGraph();
+    } else if (format.equals("indexed")) {
+      return new IndexedCSVDataSource(directory, config).getLogicalGraph();
     } else {
       throw new IllegalArgumentException("Unsupported format: " + format);
     }