UPDATE While this approach worked, it was terribly slow and so I have modified this to create files as buffers and this works very well. Here is the update ...
private long getMaxBufferedLines(File file) throws Throwable {
long freeBytes = Runtime.getRuntime().freeMemory() / 2;
return (freeBytes / (file.length() / getLineCount(file)));
}
private void join(File left, File right, File output, JoinType joinType) throws Throwable {
BufferedReader leftFile = new BufferedReader(new FileReader(left));
BufferedReader rightFile = new BufferedReader(new FileReader(right));
BufferedWriter outputFile = new BufferedWriter(new FileWriter(output));
long maxBufferedLines = getMaxBufferedLines(right);
Record leftRecord;
Record rightRecord;
leftRecord = read(leftFile);
rightRecord = read(rightFile);
while( leftRecord != null && rightRecord != null ) {
if( leftRecord.getKey().compareTo(rightRecord.getKey()) < 0) {
if( joinType == JoinType.LeftOuterJoin
|| joinType == JoinType.LeftExclusiveJoin
|| joinType == JoinType.FullExclusiveJoin
|| joinType == JoinType.FullOuterJoin ) {
write(outputFile, leftRecord, null);
}
leftRecord = read(leftFile);
}
else if( leftRecord.getKey().compareTo(rightRecord.getKey()) > 0 ) {
if( joinType == JoinType.RightOuterJoin
|| joinType == JoinType.RightExclusiveJoin
|| joinType == JoinType.FullExclusiveJoin
|| joinType == JoinType.FullOuterJoin ) {
write(outputFile, null, rightRecord);
}
rightRecord = read(rightFile);
}
else if( leftRecord.getKey().compareTo(rightRecord.getKey()) == 0 ) {
String key = leftRecord.getKey();
List<File> rightRecordFileList = new ArrayList<File>();
List<Record> rightRecordList = new ArrayList<Record>();
rightRecordList.add(rightRecord);
rightRecord = consume(key, rightFile, rightRecordList, rightRecordFileList, maxBufferedLines);
while( leftRecord != null && leftRecord.getKey().compareTo(key) == 0 ) {
processRightRecords(outputFile, leftRecord, rightRecordFileList, rightRecordList, joinType);
leftRecord = read(leftFile);
}
// need a dispose for deleting files in list
}
else {
throw new Exception("DATA IS NOT SORTED");
}
}
if( leftRecord != null ) {
if( joinType == JoinType.LeftOuterJoin
|| joinType == JoinType.LeftExclusiveJoin
|| joinType == JoinType.FullExclusiveJoin
|| joinType == JoinType.FullOuterJoin ) {
write(outputFile, leftRecord, null);
}
leftRecord = read(leftFile);
while(leftRecord != null) {
if( joinType == JoinType.LeftOuterJoin
|| joinType == JoinType.LeftExclusiveJoin
|| joinType == JoinType.FullExclusiveJoin
|| joinType == JoinType.FullOuterJoin ) {
write(outputFile, leftRecord, null);
}
leftRecord = read(leftFile);
}
}
else {
if( rightRecord != null ) {
if( joinType == JoinType.RightOuterJoin
|| joinType == JoinType.RightExclusiveJoin
|| joinType == JoinType.FullExclusiveJoin
|| joinType == JoinType.FullOuterJoin ) {
write(outputFile, null, rightRecord);
}
rightRecord = read(rightFile);
while(rightRecord != null) {
if( joinType == JoinType.RightOuterJoin
|| joinType == JoinType.RightExclusiveJoin
|| joinType == JoinType.FullExclusiveJoin
|| joinType == JoinType.FullOuterJoin ) {
write(outputFile, null, rightRecord);
}
rightRecord = read(rightFile);
}
}
}
leftFile.close();
rightFile.close();
outputFile.flush();
outputFile.close();
}
public void processRightRecords(BufferedWriter outputFile, Record leftRecord, List<File> rightFiles, List<Record> rightRecords, JoinType joinType) throws Throwable {
for(File rightFile : rightFiles) {
BufferedReader rightReader = new BufferedReader(new FileReader(rightFile));
Record rightRecord = read(rightReader);
while(rightRecord != null){
if( joinType == JoinType.LeftOuterJoin
|| joinType == JoinType.RightOuterJoin
|| joinType == JoinType.FullOuterJoin
|| joinType == JoinType.InnerJoin ) {
write(outputFile, leftRecord, rightRecord);
}
rightRecord = read(rightReader);
}
rightReader.close();
}
for(Record rightRecord : rightRecords) {
if( joinType == JoinType.LeftOuterJoin
|| joinType == JoinType.RightOuterJoin
|| joinType == JoinType.FullOuterJoin
|| joinType == JoinType.InnerJoin ) {
write(outputFile, leftRecord, rightRecord);
}
}
}
/**
* consume all records having key (either to a single list or multiple files) each file will
* store a buffer full of data. The right record returned represents the outside flow (key is
* already positioned to next one or null) so we can't use this record in below while loop or
* within this block in general when comparing current key. The trick is to keep consuming
* from a List. When it becomes empty, re-fill it from the next file until all files have
* been consumed (and the last node in the list is read). The next outside iteration will be
* ready to be processed (either it will be null or it points to the next biggest key
* @throws Throwable
*
*/
private Record consume(String key, BufferedReader reader, List<Record> records, List<File> files, long bufferMaxRecordLines ) throws Throwable {
boolean processComplete = false;
Record record = records.get(records.size() - 1);
while(!processComplete){
long recordCount = records.size();
if( record.getKey().compareTo(key) == 0 ){
record = read(reader);
while( record != null && record.getKey().compareTo(key) == 0 && recordCount < bufferMaxRecordLines ) {
records.add(record);
recordCount++;
record = read(reader);
}
}
processComplete = true;
// if record is null, we are done
if( record != null ) {
// if the key has changed, we are done
if( record.getKey().compareTo(key) == 0 ) {
// Same key means we have exhausted the buffer.
// Dump entire buffer into a file. The list of file
// pointers will keep track of the files ...
processComplete = false;
dumpBufferToFile(records, files);
records.clear();
records.add(record);
}
}
}
return record;
}
/**
* Dump all records in List of Record objects to a file. Then, add that
* file to List of File objects
*
* NEED TO PLACE A LIMIT ON NUMBER OF FILE POINTERS (check size of file list)
*
* @param records
* @param files
* @throws Throwable
*/
private void dumpBufferToFile(List<Record> records, List<File> files) throws Throwable {
String prefix = "joiner_" + files.size() + 1;
String suffix = ".dat";
File file = File.createTempFile(prefix, suffix, new File("cache"));
BufferedWriter writer = new BufferedWriter(new FileWriter(file));
for( Record record : records ) {
writer.write( record.dump() );
}
files.add(file);
writer.flush();
writer.close();
}