/*
 * 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 com.intel.hibench.streambench;

import java.io.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.*;

public class FileDataGenNew {
    Configuration fsConf = new Configuration();

    FileDataGenNew (String HDFSMaster){
        fsConf.set("fs.default.name", HDFSMaster);
    }

	public BufferedReader loadDataFromFile(String filepath, long offset){
		try {
            Path pt = new Path(filepath);
            FileSystem fs = FileSystem.get(fsConf);
            FSDataInputStream fileHandler = fs.open(pt);
            if (offset>0) fileHandler.seek(offset);
			BufferedReader reader = new BufferedReader(new InputStreamReader(fileHandler));
            if (offset>0) reader.readLine();
			return reader;
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}
}

@@ -849,24 +849,80 @@
  java.io.*;%0A
+import java.util.Enumeration;%0Aimport java.util.Vector;%0A%0A
 import org.a
@@ -1324,16 +1324,238 @@
 sConf);%0A
+            InputStreamReader isr;%0A            if (fs.isDirectory(pt)) %7B // multiple parts%0A                isr = new InputStreamReader(OpenMultiplePartsWithOffset(fs, pt, offset));%0A            %7D else %7B  // single file%0A    
         
@@ -1607,32 +1607,36 @@
 t);%0A            
+    
 if (offset%3E0) fi
@@ -1659,16 +1659,89 @@
 ffset);%0A
+                isr = new InputStreamReader(fileHandler);%0A            %7D%0A%0A
 %09%09%09Buffe
@@ -1782,42 +1782,11 @@
 der(
-new InputStreamReader(fileHandler)
+isr
 );%0A 
@@ -1828,16 +1828,51 @@
 dLine();
+ // skip first line in case of seek
 %0A%09%09%09retu
@@ -2004,24 +2004,1379 @@
 %09%09%7D%0A
-%09%09return null;%0A%09
+        assert false: %22Should not reach here!%22;%0A        return null;%0A%09%7D%0A%0A    private InputStream OpenMultiplePartsWithOffset(FileSystem fs, Path pt, long offset) throws IOException %7B%0A        RemoteIterator%3CLocatedFileStatus%3E rit = fs.listFiles(pt, false);%0A        Vector%3CFSDataInputStream%3E fileHandleList= new Vector%3CFSDataInputStream%3E();%0A        while (rit.hasNext()) %7B%0A            Path path = rit.next().getPath();%0A            String filename = path.toString().substring(path.getParent().toString().length(), path.toString().length());%0A%0A            if (filename.startsWith(%22/part-%22)) %7B%0A                long filesize = fs.getFileStatus(path).getLen();%0A                if (offset %3C filesize) %7B%0A                    FSDataInputStream handle = fs.open(path);%0A                    if (offset%3E0) %7B%0A                        handle.seek(offset);%0A                    %7D%0A                    fileHandleList.add(handle);%0A                %7D%0A                offset -= filesize;%0A            %7D%0A        %7D%0A        if (fileHandleList.size()==1) return fileHandleList.get(0);%0A        else if (fileHandleList.size()%3E1)%7B%0A            Enumeration%3CFSDataInputStream%3E enu = fileHandleList.elements();%0A            return new SequenceInputStream(enu);%0A        %7D else %7B%0A            System.err.println(%22Error, no source file loaded. run genSeedDataset.sh fisrt!%22);%0A            return null;%0A        %7D%0A    
 %7D%0A%7D%0A

improve datagen