#include "utils/ydhdfs.h"
#include "triple.h"

const char * input_dir = "/dblp_sorted";
const char * output_dir = "/dblp_merged";

struct dirReader
{
	string inDir;
	hdfsFS fs;
	int numFiles;
	hdfsFile in;
	LineReader* reader;
	int cur;

	dirReader(const char * indir)
	{
		inDir = indir;
		fs = getHdfsFS();
		hdfsFileInfo* fileinfo = hdfsListDirectory(fs, indir, &numFiles);
		if (fileinfo == NULL) {
			fprintf(stderr, "Failed to list directory %s!\n", inDir.c_str());
			exit(-1);
		}
		hdfsFreeFileInfo(fileinfo, numFiles);
		cur=0;
		string fname = inDir + "/part_0";
		cout<<"Processing "<<fname<<endl;
		in = getRHandle(fname.c_str(), fs);
		reader = new LineReader(fs, in);
	}

	char* readLine()
	{
		reader->readLine();
		if (!reader->eof()) return reader->getLine();
		else
		{
			hdfsCloseFile(fs, in);
			delete reader;
			cur++;
			if(cur<numFiles)
			{
				char temp[10];
				sprintf(temp, "%d", cur);
				string fname = inDir;
				fname += "/part_";
				fname += temp;
				cout<<"Processing "<<fname<<endl;
				in = getRHandle(fname.c_str(), fs);
				reader = new LineReader(fs, in);
				return readLine();
			}
			else return NULL;
		}
	}
};

int main(int argc, char* argv[]){
	hdfsFS fs = getHdfsFS();
	if(hdfsExists(fs, output_dir) == 0) {
		fprintf(stderr, "Output path \"%s\" already exists!\n", output_dir);
		hdfsDisconnect(fs);
		return -1;
	}
	dirReader reader(input_dir);
	BufferedWriter writer(output_dir, fs, -2);//cannot use -1, which avoids file splitting
	char * line;
	string preKey = "";
	string list = "";
	int vid = 0;
	while((line = reader.readLine()) != NULL)
	{
		statement o;
		o.parseTriple(line);
		string curKey = o.subject;
		if(curKey != preKey)
		{
			if(preKey != "")
			{
				writer.check();
				writer.write(list.c_str());
				writer.write("\n");
			}
			preKey = curKey;
			char temp[10];
			sprintf(temp, "%d", vid);
			vid++;
			list = temp;
			list += '\t' + o.subject;
		}
		list += '\t' + o.predicate + '\t' + o.object;
	}
	//flush
	if(preKey != "")
	{
		writer.check();
		writer.write(list.c_str());
	}
	return 0;
}