passed map kmer to readid operator
diff --git a/genomix/genomix-data/src/main/java/edu/uci/ics/genomix/type/PositionListWritable.java b/genomix/genomix-data/src/main/java/edu/uci/ics/genomix/type/PositionListWritable.java
index a578543..40817e8 100644
--- a/genomix/genomix-data/src/main/java/edu/uci/ics/genomix/type/PositionListWritable.java
+++ b/genomix/genomix-data/src/main/java/edu/uci/ics/genomix/type/PositionListWritable.java
@@ -10,7 +10,7 @@
import edu.uci.ics.genomix.data.Marshal;
-public class PositionListWritable implements Writable, Iterable<PositionWritable>, Serializable{
+public class PositionListWritable implements Writable, Iterable<PositionWritable>, Serializable {
/**
*
*/
@@ -121,6 +121,13 @@
valueCount += 1;
}
+ public static int getCountByDataLength(int length) {
+ if (length % PositionWritable.LENGTH != 0) {
+ throw new IllegalArgumentException("Length of positionlist is invalid");
+ }
+ return length / PositionWritable.LENGTH;
+ }
+
public int getCountOfPosition() {
return valueCount;
}
@@ -151,14 +158,18 @@
}
@Override
- public String toString(){
+ public String toString() {
StringBuilder sbuilder = new StringBuilder();
sbuilder.append('[');
- for(PositionWritable pos : this){
+ for (PositionWritable pos : this) {
sbuilder.append(pos.toString());
sbuilder.append(',');
}
- sbuilder.setCharAt(sbuilder.length()-1, ']');
+ if (valueCount > 0) {
+ sbuilder.setCharAt(sbuilder.length() - 1, ']');
+ } else {
+ sbuilder.append(']');
+ }
return sbuilder.toString();
}
}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDNormarlizedComputeFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDNormarlizedComputeFactory.java
index 4d00731..1f3efbf 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDNormarlizedComputeFactory.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDNormarlizedComputeFactory.java
@@ -1,8 +1,8 @@
package edu.uci.ics.genomix.hyracks.data.accessors;
+import edu.uci.ics.genomix.data.Marshal;
import edu.uci.ics.hyracks.api.dataflow.value.INormalizedKeyComputer;
import edu.uci.ics.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
public class ReadIDNormarlizedComputeFactory implements INormalizedKeyComputerFactory{
@@ -17,7 +17,7 @@
@Override
public int normalize(byte[] bytes, int start, int length) {
- return IntegerSerializerDeserializer.getInt(bytes, start);
+ return Marshal.getInt(bytes, start);
}
};
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDPartitionComputerFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDPartitionComputerFactory.java
index d328bd1..7229364 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDPartitionComputerFactory.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/data/accessors/ReadIDPartitionComputerFactory.java
@@ -2,10 +2,10 @@
import java.nio.ByteBuffer;
+import edu.uci.ics.genomix.data.Marshal;
import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
import edu.uci.ics.hyracks.api.dataflow.value.ITuplePartitionComputer;
import edu.uci.ics.hyracks.api.dataflow.value.ITuplePartitionComputerFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
public class ReadIDPartitionComputerFactory implements ITuplePartitionComputerFactory {
@@ -26,7 +26,7 @@
ByteBuffer buf = accessor.getBuffer();
- int hash = IntegerSerializerDeserializer.getInt(buf.array(), startOffset + fieldOffset + slotLength);
+ int hash = Marshal.getInt(buf.array(), startOffset + fieldOffset + slotLength);
if (hash < 0) {
hash = -(hash + 1);
}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/MapKmerPositionToReadOperator.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/MapKmerPositionToReadOperator.java
index 09c0d2e..84334c6 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/MapKmerPositionToReadOperator.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/MapKmerPositionToReadOperator.java
@@ -18,186 +18,160 @@
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
-public class MapKmerPositionToReadOperator extends
- AbstractSingleActivityOperatorDescriptor {
+public class MapKmerPositionToReadOperator extends AbstractSingleActivityOperatorDescriptor {
- public MapKmerPositionToReadOperator(IOperatorDescriptorRegistry spec,
- RecordDescriptor recDesc) {
- super(spec, 1, 1);
- recordDescriptors[0] = recDesc;
- }
+ public MapKmerPositionToReadOperator(IOperatorDescriptorRegistry spec, RecordDescriptor recDesc) {
+ super(spec, 1, 1);
+ recordDescriptors[0] = recDesc;
+ }
- private static final long serialVersionUID = 1L;
- public static final int InputKmerField = 0;
- public static final int InputPosListField = 1;
+ private static final long serialVersionUID = 1L;
+ public static final int InputKmerField = 0;
+ public static final int InputPosListField = 1;
- public static final int OutputReadIDField = 0;
- public static final int OutputPosInReadField = 1;
- public static final int OutputOtherReadIDListField = 2;
- public static final int OutputKmerField = 3; // may not needed
+ public static final int OutputReadIDField = 0;
+ public static final int OutputPosInReadField = 1;
+ public static final int OutputOtherReadIDListField = 2;
+ public static final int OutputKmerField = 3; // may not needed
- public static final RecordDescriptor readIDOutputRec = new RecordDescriptor(
- new ISerializerDeserializer[] { null, null, null, null });
+ public static final RecordDescriptor readIDOutputRec = new RecordDescriptor(new ISerializerDeserializer[] { null,
+ null, null, null });
- /**
- * Map (Kmer, {(ReadID,PosInRead),...}) into
- * (ReadID,PosInRead,{OtherReadID,...},*Kmer*) OtherReadID appears only when
- * otherReadID.otherPos==0
- */
- public class MapKmerPositionToReadNodePushable extends
- AbstractUnaryInputUnaryOutputOperatorNodePushable {
- private final IHyracksTaskContext ctx;
- private final RecordDescriptor inputRecDesc;
- private final RecordDescriptor outputRecDesc;
+ /**
+ * Map (Kmer, {(ReadID,PosInRead),...}) into
+ * (ReadID,PosInRead,{OtherReadID,...},*Kmer*) OtherReadID appears only when
+ * otherReadID.otherPos==0
+ */
+ public class MapKmerPositionToReadNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
+ private final IHyracksTaskContext ctx;
+ private final RecordDescriptor inputRecDesc;
+ private final RecordDescriptor outputRecDesc;
- private FrameTupleAccessor accessor;
- private ByteBuffer writeBuffer;
- private ArrayTupleBuilder builder;
- private FrameTupleAppender appender;
+ private FrameTupleAccessor accessor;
+ private ByteBuffer writeBuffer;
+ private ArrayTupleBuilder builder;
+ private FrameTupleAppender appender;
- private PositionReference positionEntry;
- private ArrayBackedValueStorage posListEntry;
- private ArrayBackedValueStorage zeroPositionCollection;
- private ArrayBackedValueStorage noneZeroPositionCollection;
+ private PositionReference positionEntry;
+ private ArrayBackedValueStorage posListEntry;
+ private ArrayBackedValueStorage zeroPositionCollection;
+ private ArrayBackedValueStorage noneZeroPositionCollection;
- public MapKmerPositionToReadNodePushable(IHyracksTaskContext ctx,
- RecordDescriptor inputRecDesc, RecordDescriptor outputRecDesc) {
- this.ctx = ctx;
- this.inputRecDesc = inputRecDesc;
- this.outputRecDesc = outputRecDesc;
- this.positionEntry = new PositionReference();
- this.posListEntry = new ArrayBackedValueStorage();
- this.zeroPositionCollection = new ArrayBackedValueStorage();
- this.noneZeroPositionCollection = new ArrayBackedValueStorage();
- }
+ public MapKmerPositionToReadNodePushable(IHyracksTaskContext ctx, RecordDescriptor inputRecDesc,
+ RecordDescriptor outputRecDesc) {
+ this.ctx = ctx;
+ this.inputRecDesc = inputRecDesc;
+ this.outputRecDesc = outputRecDesc;
+ this.positionEntry = new PositionReference();
+ this.posListEntry = new ArrayBackedValueStorage();
+ this.zeroPositionCollection = new ArrayBackedValueStorage();
+ this.noneZeroPositionCollection = new ArrayBackedValueStorage();
+ }
- @Override
- public void open() throws HyracksDataException {
- accessor = new FrameTupleAccessor(ctx.getFrameSize(), inputRecDesc);
- writeBuffer = ctx.allocateFrame();
- builder = new ArrayTupleBuilder(outputRecDesc.getFieldCount());
- appender = new FrameTupleAppender(ctx.getFrameSize());
- appender.reset(writeBuffer, true);
- writer.open();
- posListEntry.reset();
- }
+ @Override
+ public void open() throws HyracksDataException {
+ accessor = new FrameTupleAccessor(ctx.getFrameSize(), inputRecDesc);
+ writeBuffer = ctx.allocateFrame();
+ builder = new ArrayTupleBuilder(outputRecDesc.getFieldCount());
+ appender = new FrameTupleAppender(ctx.getFrameSize());
+ appender.reset(writeBuffer, true);
+ writer.open();
+ posListEntry.reset();
+ }
- @Override
- public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- accessor.reset(buffer);
- int tupleCount = accessor.getTupleCount();
- for (int i = 0; i < tupleCount; i++) {
- scanPosition(i, zeroPositionCollection,
- noneZeroPositionCollection);
- scanAgainToOutputTuple(i, zeroPositionCollection,
- noneZeroPositionCollection, builder);
- }
- }
+ @Override
+ public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+ accessor.reset(buffer);
+ int tupleCount = accessor.getTupleCount();
+ for (int i = 0; i < tupleCount; i++) {
+ scanPosition(i, zeroPositionCollection, noneZeroPositionCollection);
+ scanAgainToOutputTuple(i, zeroPositionCollection, noneZeroPositionCollection, builder);
+ }
+ }
- private void scanPosition(int tIndex,
- ArrayBackedValueStorage zeroPositionCollection2,
- ArrayBackedValueStorage noneZeroPositionCollection2) {
- zeroPositionCollection2.reset();
- noneZeroPositionCollection2.reset();
- byte[] data = accessor.getBuffer().array();
- int offsetPoslist = accessor.getTupleStartOffset(tIndex)
- + accessor.getFieldSlotsLength()
- + accessor.getFieldStartOffset(tIndex, InputPosListField);
- for (int i = 0; i < accessor.getFieldLength(tIndex,
- InputPosListField); i += PositionReference.LENGTH) {
- positionEntry.setNewReference(data, offsetPoslist + i);
- if (positionEntry.getPosInRead() == 0) {
- zeroPositionCollection2.append(positionEntry);
- } else {
- noneZeroPositionCollection2.append(positionEntry);
- }
- }
+ private void scanPosition(int tIndex, ArrayBackedValueStorage zeroPositionCollection2,
+ ArrayBackedValueStorage noneZeroPositionCollection2) {
+ zeroPositionCollection2.reset();
+ noneZeroPositionCollection2.reset();
+ byte[] data = accessor.getBuffer().array();
+ int offsetPoslist = accessor.getTupleStartOffset(tIndex) + accessor.getFieldSlotsLength()
+ + accessor.getFieldStartOffset(tIndex, InputPosListField);
+ for (int i = 0; i < accessor.getFieldLength(tIndex, InputPosListField); i += PositionReference.LENGTH) {
+ positionEntry.setNewReference(data, offsetPoslist + i);
+ if (positionEntry.getPosInRead() == 0) {
+ zeroPositionCollection2.append(positionEntry);
+ } else {
+ noneZeroPositionCollection2.append(positionEntry);
+ }
+ }
- }
+ }
- private void scanAgainToOutputTuple(int tIndex,
- ArrayBackedValueStorage zeroPositionCollection,
- ArrayBackedValueStorage noneZeroPositionCollection,
- ArrayTupleBuilder builder2) {
- byte[] data = accessor.getBuffer().array();
- int offsetPoslist = accessor.getTupleStartOffset(tIndex)
- + accessor.getFieldSlotsLength()
- + accessor.getFieldStartOffset(tIndex, InputPosListField);
- for (int i = 0; i < accessor.getFieldLength(tIndex,
- InputPosListField); i += PositionReference.LENGTH) {
- positionEntry.setNewReference(data, offsetPoslist + i);
- if (positionEntry.getPosInRead() != 0) {
- appendNodeToBuilder(tIndex, positionEntry,
- zeroPositionCollection, builder2);
- } else {
- appendNodeToBuilder(tIndex, positionEntry,
- noneZeroPositionCollection, builder2);
- }
- }
- }
+ private void scanAgainToOutputTuple(int tIndex, ArrayBackedValueStorage zeroPositionCollection,
+ ArrayBackedValueStorage noneZeroPositionCollection, ArrayTupleBuilder builder2) {
+ byte[] data = accessor.getBuffer().array();
+ int offsetPoslist = accessor.getTupleStartOffset(tIndex) + accessor.getFieldSlotsLength()
+ + accessor.getFieldStartOffset(tIndex, InputPosListField);
+ for (int i = 0; i < accessor.getFieldLength(tIndex, InputPosListField); i += PositionReference.LENGTH) {
+ positionEntry.setNewReference(data, offsetPoslist + i);
+ if (positionEntry.getPosInRead() != 0) {
+ appendNodeToBuilder(tIndex, positionEntry, zeroPositionCollection, builder2);
+ } else {
+ appendNodeToBuilder(tIndex, positionEntry, noneZeroPositionCollection, builder2);
+ }
+ }
+ }
- private void appendNodeToBuilder(int tIndex, PositionReference pos,
- ArrayBackedValueStorage posList2, ArrayTupleBuilder builder2) {
- try {
- builder2.addField(pos.getByteArray(), 0,
- PositionReference.INTBYTES);
- builder2.addField(pos.getByteArray(),
- PositionReference.INTBYTES, 1);
- // ? ask Yingyi, if support empty bytes[]
- if (posList2 == null) {
- builder2.addFieldEndOffset();
- } else {
- builder2.addField(posList2.getByteArray(),
- posList2.getStartOffset(), posList2.getLength());
- }
- // set kmer, may not useful
- byte[] data = accessor.getBuffer().array();
- int offsetKmer = accessor.getTupleStartOffset(tIndex)
- + accessor.getFieldSlotsLength()
- + accessor.getFieldStartOffset(tIndex, InputKmerField);
- builder2.addField(data, offsetKmer,
- accessor.getFieldLength(tIndex, InputKmerField));
+ private void appendNodeToBuilder(int tIndex, PositionReference pos, ArrayBackedValueStorage posList2,
+ ArrayTupleBuilder builder2) {
+ try {
+ builder2.addField(pos.getByteArray(), pos.getStartOffset(), PositionReference.INTBYTES);
+ builder2.addField(pos.getByteArray(), pos.getStartOffset() + PositionReference.INTBYTES, 1);
+ if (posList2 == null) {
+ builder2.addFieldEndOffset();
+ } else {
+ builder2.addField(posList2.getByteArray(), posList2.getStartOffset(), posList2.getLength());
+ }
+ // set kmer, may not useful
+ byte[] data = accessor.getBuffer().array();
+ int offsetKmer = accessor.getTupleStartOffset(tIndex) + accessor.getFieldSlotsLength()
+ + accessor.getFieldStartOffset(tIndex, InputKmerField);
+ builder2.addField(data, offsetKmer, accessor.getFieldLength(tIndex, InputKmerField));
- if (!appender.append(builder2.getFieldEndOffsets(),
- builder2.getByteArray(), 0, builder2.getSize())) {
- FrameUtils.flushFrame(writeBuffer, writer);
- appender.reset(writeBuffer, true);
- if (!appender.append(builder2.getFieldEndOffsets(),
- builder2.getByteArray(), 0, builder2.getSize())) {
- throw new IllegalStateException();
- }
- }
- builder2.reset();
- } catch (HyracksDataException e) {
- throw new IllegalStateException(
- "Failed to Add a field to the tuple by copying the data bytes from a byte array.");
- }
- }
+ if (!appender.append(builder2.getFieldEndOffsets(), builder2.getByteArray(), 0, builder2.getSize())) {
+ FrameUtils.flushFrame(writeBuffer, writer);
+ appender.reset(writeBuffer, true);
+ if (!appender.append(builder2.getFieldEndOffsets(), builder2.getByteArray(), 0, builder2.getSize())) {
+ throw new IllegalStateException();
+ }
+ }
+ builder2.reset();
+ } catch (HyracksDataException e) {
+ throw new IllegalStateException(
+ "Failed to Add a field to the tuple by copying the data bytes from a byte array.");
+ }
+ }
- @Override
- public void fail() throws HyracksDataException {
- writer.fail();
- }
+ @Override
+ public void fail() throws HyracksDataException {
+ writer.fail();
+ }
- @Override
- public void close() throws HyracksDataException {
- if (appender.getTupleCount() > 0) {
- FrameUtils.flushFrame(writeBuffer, writer);
- }
- writer.close();
- }
+ @Override
+ public void close() throws HyracksDataException {
+ if (appender.getTupleCount() > 0) {
+ FrameUtils.flushFrame(writeBuffer, writer);
+ }
+ writer.close();
+ }
- }
+ }
- @Override
- public AbstractOperatorNodePushable createPushRuntime(
- IHyracksTaskContext ctx,
- IRecordDescriptorProvider recordDescProvider, int partition,
- int nPartitions) {
- return new MapKmerPositionToReadNodePushable(
- ctx,
- recordDescProvider.getInputRecordDescriptor(getActivityId(), 0),
- recordDescriptors[0]);
- }
+ @Override
+ public AbstractOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
+ IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
+ return new MapKmerPositionToReadNodePushable(ctx, recordDescProvider.getInputRecordDescriptor(getActivityId(),
+ 0), recordDescriptors[0]);
+ }
}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/ReadsKeyValueParserFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/ReadsKeyValueParserFactory.java
index b36b82d..37d5289 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/ReadsKeyValueParserFactory.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/ReadsKeyValueParserFactory.java
@@ -47,14 +47,16 @@
private KmerBytesWritable kmer;
private PositionReference pos;
private boolean bReversed;
+ private final int readLength;
public static final RecordDescriptor readKmerOutputRec = new RecordDescriptor(new ISerializerDeserializer[] { null,
null });
- public ReadsKeyValueParserFactory(int k, boolean bGenerateReversed) {
+ public ReadsKeyValueParserFactory(int readlength, int k, boolean bGenerateReversed) {
bReversed = bGenerateReversed;
kmer = new KmerBytesWritable(k);
pos = new PositionReference();
+ this.readLength = readlength;
}
@Override
@@ -76,7 +78,7 @@
try {
readID = Integer.parseInt(geneLine[0]);
} catch (NumberFormatException e) {
- LOG.warn("Invalid data");
+ LOG.warn("Invalid data " );
return;
}
@@ -84,6 +86,10 @@
Matcher geneMatcher = genePattern.matcher(geneLine[1]);
boolean isValid = geneMatcher.matches();
if (isValid) {
+ if (geneLine[1].length() != readLength){
+ LOG.warn("Invalid readlength at: " + readID );
+ return;
+ }
SplitReads(readID, geneLine[1].getBytes(), writer);
}
}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateKmerAggregateFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateKmerAggregateFactory.java
index 4006fb4..6ee7946 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateKmerAggregateFactory.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateKmerAggregateFactory.java
@@ -47,7 +47,6 @@
@Override
public AggregateState createAggregateStates() {
- System.out.println("CreateState");
return new AggregateState(new ArrayBackedValueStorage());
}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateReadIDAggregateFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateReadIDAggregateFactory.java
index c7552ca..fe9dab5 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateReadIDAggregateFactory.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/AggregateReadIDAggregateFactory.java
@@ -7,6 +7,7 @@
import edu.uci.ics.genomix.hyracks.dataflow.MapKmerPositionToReadOperator;
import edu.uci.ics.hyracks.api.comm.IFrameTupleAccessor;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.data.std.util.ArrayBackedValueStorage;
@@ -21,6 +22,7 @@
*
*/
private static final long serialVersionUID = 1L;
+
public static final int InputReadIDField = MapKmerPositionToReadOperator.OutputReadIDField;
public static final int InputPosInReadField = MapKmerPositionToReadOperator.OutputPosInReadField;
public static final int InputPositionListField = MapKmerPositionToReadOperator.OutputOtherReadIDListField;
@@ -29,6 +31,9 @@
public static final int OutputReadIDField = 0;
public static final int OutputPositionListField = 1;
+ public static final RecordDescriptor readIDAggregateRec = new RecordDescriptor(new ISerializerDeserializer[] {
+ null, null });
+
public AggregateReadIDAggregateFactory() {
}
@@ -65,7 +70,7 @@
ArrayBackedValueStorage storage = (ArrayBackedValueStorage) state.state;
storage.reset();
DataOutput out = storage.getDataOutput();
- byte posInRead = readByteField(accessor, tIndex, InputPositionListField);
+ byte posInRead = readByteField(accessor, tIndex, InputPosInReadField);
try {
out.writeByte(posInRead);
@@ -74,7 +79,8 @@
} catch (IOException e) {
throw new HyracksDataException("Failed to write into temporary storage");
}
-
+ // make fake feild
+ tupleBuilder.addFieldEndOffset();
}
private void writeBytesToStorage(DataOutput out, IFrameTupleAccessor accessor, int tIndex, int idField)
@@ -95,7 +101,7 @@
int stateTupleIndex, AggregateState state) throws HyracksDataException {
ArrayBackedValueStorage storage = (ArrayBackedValueStorage) state.state;
DataOutput out = storage.getDataOutput();
- byte posInRead = readByteField(accessor, tIndex, InputPositionListField);
+ byte posInRead = readByteField(accessor, tIndex, InputPosInReadField);
try {
out.writeByte(posInRead);
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/MergeReadIDAggregateFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/MergeReadIDAggregateFactory.java
index d3971fb..fb36898 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/MergeReadIDAggregateFactory.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/aggregators/MergeReadIDAggregateFactory.java
@@ -14,57 +14,55 @@
import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
-public class MergeReadIDAggregateFactory implements
- IAggregatorDescriptorFactory {
+public class MergeReadIDAggregateFactory implements IAggregatorDescriptorFactory {
- /**
+ /**
*
*/
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = 1L;
- private final int ValidPosCount;
+ private final int ValidPosCount;
- public MergeReadIDAggregateFactory(int readLength, int kmerLength) {
- ValidPosCount = getPositionCount(readLength, kmerLength);
- }
+ public MergeReadIDAggregateFactory(int readLength, int kmerLength) {
+ ValidPosCount = getPositionCount(readLength, kmerLength);
+ }
- public static int getPositionCount(int readLength, int kmerLength) {
- return readLength - kmerLength + 1;
- }
+ public static int getPositionCount(int readLength, int kmerLength) {
+ return readLength - kmerLength + 1;
+ }
- public static final int InputReadIDField = AggregateReadIDAggregateFactory.OutputReadIDField;
- public static final int InputPositionListField = AggregateReadIDAggregateFactory.OutputPositionListField;
+ public static final int InputReadIDField = AggregateReadIDAggregateFactory.OutputReadIDField;
+ public static final int InputPositionListField = AggregateReadIDAggregateFactory.OutputPositionListField;
- public static final int BYTE_SIZE = 1;
- public static final int INTEGER_SIZE = 4;
+ public static final int BYTE_SIZE = 1;
+ public static final int INTEGER_SIZE = 4;
- /**
- * (ReadID, {(PosInRead,{OtherPositoin..},Kmer) ...} to Aggregate as
- * (ReadID, Storage[posInRead]={PositionList,Kmer})
- *
- */
- @Override
+ /**
+ * (ReadID, {(PosInRead,{OtherPositoin..},Kmer) ...} to Aggregate as
+ * (ReadID, Storage[posInRead]={PositionList,Kmer})
+ */
+ @Override
public IAggregatorDescriptor createAggregator(IHyracksTaskContext ctx, RecordDescriptor inRecordDescriptor,
RecordDescriptor outRecordDescriptor, int[] keyFields, int[] keyFieldsInPartialResults)
throws HyracksDataException {
return new IAggregatorDescriptor() {
-
- class PositionArray{
- public ArrayBackedValueStorage[] storages;
- public int count;
-
- public PositionArray(ArrayBackedValueStorage[] storages2, int i){
- storages = storages2;
- count = i;
- }
-
- public void reset(){
- for (ArrayBackedValueStorage each : storages) {
+
+ class PositionArray {
+ public ArrayBackedValueStorage[] storages;
+ public int count;
+
+ public PositionArray(ArrayBackedValueStorage[] storages2, int i) {
+ storages = storages2;
+ count = i;
+ }
+
+ public void reset() {
+ for (ArrayBackedValueStorage each : storages) {
each.reset();
}
count = 0;
- }
- }
+ }
+ }
@Override
public AggregateState createAggregateStates() {
@@ -72,7 +70,7 @@
for (int i = 0; i < storages.length; i++) {
storages[i] = new ArrayBackedValueStorage();
}
- return new AggregateState(new PositionArray(storages,0));
+ return new AggregateState(new PositionArray(storages, 0));
}
@Override
@@ -98,6 +96,10 @@
fieldOffset += writeBytesToStorage(storages[posInRead], fieldBuffer, fieldOffset);
positionArray.count++;
}
+ // make fake fields
+ for (int i = 0; i < ValidPosCount; i++) {
+ tupleBuilder.addFieldEndOffset();
+ }
}
private int writeBytesToStorage(ArrayBackedValueStorage storage, ByteBuffer fieldBuffer, int fieldOffset)
@@ -152,7 +154,7 @@
@Override
public void outputFinalResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor, int tIndex,
AggregateState state) throws HyracksDataException {
- PositionArray positionArray = (PositionArray) state.state;
+ PositionArray positionArray = (PositionArray) state.state;
ArrayBackedValueStorage[] storages = positionArray.storages;
if (positionArray.count != storages.length) {
throw new IllegalStateException("Final aggregate position number is invalid");
@@ -160,7 +162,8 @@
DataOutput fieldOutput = tupleBuilder.getDataOutput();
try {
for (int i = 0; i < storages.length; i++) {
- fieldOutput.write(storages[i].getByteArray(), storages[i].getStartOffset(), storages[i].getLength());
+ fieldOutput.write(storages[i].getByteArray(), storages[i].getStartOffset(),
+ storages[i].getLength());
tupleBuilder.addFieldEndOffset();
}
} catch (IOException e) {
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/io/MapperKmerToReadIDTextWriterFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/io/MapperKmerToReadIDTextWriterFactory.java
deleted file mode 100644
index 32e7b4d..0000000
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/io/MapperKmerToReadIDTextWriterFactory.java
+++ /dev/null
@@ -1,25 +0,0 @@
-package edu.uci.ics.genomix.hyracks.dataflow.io;
-
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.hdfs.api.ITupleWriter;
-import edu.uci.ics.hyracks.hdfs.api.ITupleWriterFactory;
-
-public class MapperKmerToReadIDTextWriterFactory implements ITupleWriterFactory {
-
- /**
- *
- */
- private static final long serialVersionUID = 1L;
-
- public MapperKmerToReadIDTextWriterFactory(int kmerSize) {
- // TODO Auto-generated constructor stub
- }
-
- @Override
- public ITupleWriter getTupleWriter(IHyracksTaskContext ctx) throws HyracksDataException {
- // TODO Auto-generated method stub
- return null;
- }
-
-}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/io/ReadIDAggregationTextWriterFactory.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/io/ReadIDAggregationTextWriterFactory.java
deleted file mode 100644
index 28b8484..0000000
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/dataflow/io/ReadIDAggregationTextWriterFactory.java
+++ /dev/null
@@ -1,25 +0,0 @@
-package edu.uci.ics.genomix.hyracks.dataflow.io;
-
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.hdfs.api.ITupleWriter;
-import edu.uci.ics.hyracks.hdfs.api.ITupleWriterFactory;
-
-public class ReadIDAggregationTextWriterFactory implements ITupleWriterFactory {
-
- /**
- *
- */
- private static final long serialVersionUID = 1L;
-
- public ReadIDAggregationTextWriterFactory(int kmerSize) {
- // TODO Auto-generated constructor stub
- }
-
- @Override
- public ITupleWriter getTupleWriter(IHyracksTaskContext ctx) throws HyracksDataException {
- // TODO Auto-generated method stub
- return null;
- }
-
-}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenBrujinGraph.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenBrujinGraph.java
index aa0c05d..8e45884 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenBrujinGraph.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenBrujinGraph.java
@@ -173,7 +173,7 @@
.getSplits(hadoopJobConfFactory.getConf(), ncNodeNames.length);
return new HDFSReadOperatorDescriptor(jobSpec, ReadsKeyValueParserFactory.readKmerOutputRec,
- hadoopJobConfFactory.getConf(), splits, readSchedule, new ReadsKeyValueParserFactory(kmerSize,
+ hadoopJobConfFactory.getConf(), splits, readSchedule, new ReadsKeyValueParserFactory(readLength, kmerSize,
bGenerateReversedKmer));
} catch (Exception e) {
throw new HyracksDataException(e);
@@ -232,16 +232,16 @@
int[] keyFields = new int[] { 0 }; // the id of grouped key
// (ReadID, {(PosInRead,{OtherPositoin..},Kmer) ...}
ExternalSortOperatorDescriptor sorter = new ExternalSortOperatorDescriptor(jobSpec, frameLimits, keyFields,
- new IBinaryComparatorFactory[] { PointableBinaryComparatorFactory.of(KmerPointable.FACTORY) },
- ReadsKeyValueParserFactory.readKmerOutputRec);
+ new IBinaryComparatorFactory[] { PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY) },
+ MapKmerPositionToReadOperator.readIDOutputRec);
connectOperators(jobSpec, mapKmerToRead, ncNodeNames, sorter, ncNodeNames, new OneToOneConnectorDescriptor(
jobSpec));
- RecordDescriptor readIDCombineRec = new RecordDescriptor(new ISerializerDeserializer[] { null, null });
+
RecordDescriptor readIDFinalRec = new RecordDescriptor(
new ISerializerDeserializer[MergeReadIDAggregateFactory.getPositionCount(readLength, kmerSize)]);
Object[] objs = generateAggeragateDescriptorbyType(jobSpec, keyFields, new AggregateReadIDAggregateFactory(),
- new MergeReadIDAggregateFactory(readLength, kmerSize), new ReadIDPartitionComputerFactory(),
- new ReadIDNormarlizedComputeFactory(), IntegerPointable.FACTORY, readIDCombineRec, readIDFinalRec);
+ new MergeReadIDAggregateFactory(readLength, kmerSize), new ReadIDPartitionComputerFactory(), null,
+ IntegerPointable.FACTORY, AggregateReadIDAggregateFactory.readIDAggregateRec, readIDFinalRec);
AbstractOperatorDescriptor readLocalAggregator = (AbstractOperatorDescriptor) objs[0];
connectOperators(jobSpec, sorter, ncNodeNames, readLocalAggregator, ncNodeNames,
new OneToOneConnectorDescriptor(jobSpec));
@@ -327,9 +327,6 @@
logDebug("Group by Read Operator");
lastOperator = generateGroupbyReadJob(jobSpec, lastOperator);
- logDebug("Map ReadInfo to Node");
- lastOperator = generateMapperFromReadToNode(jobSpec, lastOperator);
-
logDebug("Write node to result");
lastOperator = generateNodeWriterOpertator(jobSpec, lastOperator);
@@ -352,9 +349,9 @@
bGenerateReversedKmer = conf.getBoolean(GenomixJobConf.REVERSED_KMER, GenomixJobConf.DEFAULT_REVERSED);
String type = conf.get(GenomixJobConf.GROUPBY_TYPE, GenomixJobConf.GROUPBY_TYPE_PRECLUSTER);
- if (type.equalsIgnoreCase("external")) {
+ if (type.equalsIgnoreCase(GenomixJobConf.GROUPBY_TYPE_EXTERNAL)) {
groupbyType = GroupbyType.EXTERNAL;
- } else if (type.equalsIgnoreCase("precluster")) {
+ } else if (type.equalsIgnoreCase(GenomixJobConf.GROUPBY_TYPE_PRECLUSTER)) {
groupbyType = GroupbyType.PRECLUSTER;
} else {
groupbyType = GroupbyType.HYBRIDHASH;
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenGroupbyReadID.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenGroupbyReadID.java
index 8a11379..d5e52fa 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenGroupbyReadID.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenGroupbyReadID.java
@@ -1,14 +1,17 @@
package edu.uci.ics.genomix.hyracks.job;
+import java.io.DataOutput;
import java.util.Map;
-import edu.uci.ics.genomix.hyracks.dataflow.io.ReadIDAggregationTextWriterFactory;
import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.exceptions.HyracksException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.connectors.OneToOneConnectorDescriptor;
+import edu.uci.ics.hyracks.hdfs.api.ITupleWriter;
import edu.uci.ics.hyracks.hdfs.api.ITupleWriterFactory;
import edu.uci.ics.hyracks.hdfs.dataflow.HDFSReadOperatorDescriptor;
import edu.uci.ics.hyracks.hdfs.dataflow.HDFSWriteOperatorDescriptor;
@@ -46,22 +49,52 @@
logDebug("Group by Read Operator");
lastOperator = generateGroupbyReadJob(jobSpec, lastOperator);
- logDebug("Map ReadInfo to Node");
- lastOperator = generateMapperFromReadToNode(jobSpec, lastOperator);
-
logDebug("Write node to result");
- generateRootByWriteReadIDAggregationResult(jobSpec, lastOperator);
-
+ lastOperator = generateRootByWriteReadIDAggregationResult(jobSpec, lastOperator);
+ jobSpec.addRoot(lastOperator);
return jobSpec;
}
public AbstractOperatorDescriptor generateRootByWriteReadIDAggregationResult(JobSpecification jobSpec,
AbstractOperatorDescriptor readCrossAggregator) throws HyracksException {
- ITupleWriterFactory readWriter = new ReadIDAggregationTextWriterFactory(kmerSize);
- HDFSWriteOperatorDescriptor writeKmerOperator = new HDFSWriteOperatorDescriptor(jobSpec, hadoopJobConfFactory.getConf(), readWriter);
+ HDFSWriteOperatorDescriptor writeKmerOperator = new HDFSWriteOperatorDescriptor(jobSpec,
+ hadoopJobConfFactory.getConf(), new ITupleWriterFactory() {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public ITupleWriter getTupleWriter(IHyracksTaskContext ctx) throws HyracksDataException {
+ // TODO Auto-generated method stub
+ return new ITupleWriter() {
+
+ @Override
+ public void open(DataOutput output) throws HyracksDataException {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void write(DataOutput output, ITupleReference tuple) throws HyracksDataException {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void close(DataOutput output) throws HyracksDataException {
+ // TODO Auto-generated method stub
+
+ }
+
+ };
+ }
+
+ });
connectOperators(jobSpec, readCrossAggregator, ncNodeNames, writeKmerOperator, ncNodeNames,
new OneToOneConnectorDescriptor(jobSpec));
- jobSpec.addRoot(writeKmerOperator);
+
return writeKmerOperator;
}
diff --git a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenMapKmerToRead.java b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenMapKmerToRead.java
index 4fa554f..f2cad73 100644
--- a/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenMapKmerToRead.java
+++ b/genomix/genomix-hyracks/src/main/java/edu/uci/ics/genomix/hyracks/job/JobGenMapKmerToRead.java
@@ -1,14 +1,23 @@
package edu.uci.ics.genomix.hyracks.job;
+import java.io.DataOutput;
+import java.io.IOException;
import java.util.Map;
-import edu.uci.ics.genomix.hyracks.dataflow.io.MapperKmerToReadIDTextWriterFactory;
+import edu.uci.ics.genomix.data.Marshal;
+import edu.uci.ics.genomix.hyracks.dataflow.MapKmerPositionToReadOperator;
+import edu.uci.ics.genomix.hyracks.dataflow.ReadsKeyValueParserFactory;
+import edu.uci.ics.genomix.type.KmerBytesWritable;
+import edu.uci.ics.genomix.type.PositionListWritable;
import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.exceptions.HyracksException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.connectors.OneToOneConnectorDescriptor;
+import edu.uci.ics.hyracks.hdfs.api.ITupleWriter;
import edu.uci.ics.hyracks.hdfs.api.ITupleWriterFactory;
import edu.uci.ics.hyracks.hdfs.dataflow.HDFSReadOperatorDescriptor;
import edu.uci.ics.hyracks.hdfs.dataflow.HDFSWriteOperatorDescriptor;
@@ -26,13 +35,79 @@
super(job, scheduler, ncMap, numPartitionPerMachine);
// TODO Auto-generated constructor stub
}
-
- public AbstractOperatorDescriptor generateRootByWriteMapperFromKmerToReadID(JobSpecification jobSpec, AbstractOperatorDescriptor mapper) throws HyracksException{
+
+ public AbstractOperatorDescriptor generateRootByWriteMapperFromKmerToReadID(JobSpecification jobSpec,
+ AbstractOperatorDescriptor mapper) throws HyracksException {
// Output Kmer
- ITupleWriterFactory kmerWriter = new MapperKmerToReadIDTextWriterFactory(kmerSize);
- HDFSWriteOperatorDescriptor writeKmerOperator = new HDFSWriteOperatorDescriptor(jobSpec, hadoopJobConfFactory.getConf(), kmerWriter);
- connectOperators(jobSpec, mapper, ncNodeNames, writeKmerOperator, ncNodeNames,
- new OneToOneConnectorDescriptor(jobSpec));
+ HDFSWriteOperatorDescriptor writeKmerOperator = new HDFSWriteOperatorDescriptor(jobSpec,
+ hadoopJobConfFactory.getConf(), new ITupleWriterFactory() {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public ITupleWriter getTupleWriter(IHyracksTaskContext ctx) throws HyracksDataException {
+ return new ITupleWriter() {
+
+ private KmerBytesWritable kmer = new KmerBytesWritable(kmerSize);
+ private PositionListWritable plist = new PositionListWritable();
+
+ @Override
+ public void open(DataOutput output) throws HyracksDataException {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void write(DataOutput output, ITupleReference tuple) throws HyracksDataException {
+ try {
+ int readID = Marshal.getInt(
+ tuple.getFieldData(MapKmerPositionToReadOperator.OutputReadIDField),
+ tuple.getFieldStart(MapKmerPositionToReadOperator.OutputReadIDField));
+ byte posInRead = tuple
+ .getFieldData(MapKmerPositionToReadOperator.OutputPosInReadField)[tuple
+ .getFieldStart(MapKmerPositionToReadOperator.OutputPosInReadField)];
+ int posCount = PositionListWritable.getCountByDataLength(tuple
+ .getFieldLength(MapKmerPositionToReadOperator.OutputOtherReadIDListField));
+ plist.setNewReference(
+ posCount,
+ tuple.getFieldData(MapKmerPositionToReadOperator.OutputOtherReadIDListField),
+ tuple.getFieldStart(MapKmerPositionToReadOperator.OutputOtherReadIDListField));
+
+ if (kmer.getLength() > tuple
+ .getFieldLength(ReadsKeyValueParserFactory.OutputKmerField)) {
+ throw new IllegalArgumentException("Not enough kmer bytes");
+ }
+ kmer.setNewReference(
+ tuple.getFieldData(MapKmerPositionToReadOperator.OutputKmerField),
+ tuple.getFieldStart(MapKmerPositionToReadOperator.OutputKmerField));
+
+ String out = readID + "\t" + (int) posInRead + "\t";
+ output.write(out.getBytes());
+ output.write(plist.toString().getBytes());
+ output.writeByte('\t');
+ output.write(kmer.toString().getBytes());
+ output.writeByte('\n');
+ } catch (IOException e) {
+ throw new HyracksDataException(e);
+ }
+
+ }
+
+ @Override
+ public void close(DataOutput output) throws HyracksDataException {
+ // TODO Auto-generated method stub
+
+ }
+
+ };
+ }
+
+ });
+ connectOperators(jobSpec, mapper, ncNodeNames, writeKmerOperator, ncNodeNames, new OneToOneConnectorDescriptor(
+ jobSpec));
jobSpec.addRoot(writeKmerOperator);
return writeKmerOperator;
}
@@ -51,7 +126,7 @@
lastOperator = generateMapperFromKmerToRead(jobSpec, lastOperator);
generateRootByWriteMapperFromKmerToReadID(jobSpec, lastOperator);
-
+
return jobSpec;
}
}
diff --git a/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/JobRunStepByStepTest.java b/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/JobRunStepByStepTest.java
index 6cb2491..631fc5b 100644
--- a/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/JobRunStepByStepTest.java
+++ b/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/JobRunStepByStepTest.java
@@ -31,6 +31,7 @@
@SuppressWarnings("deprecation")
public class JobRunStepByStepTest {
private static final int KmerSize = 5;
+ private static final int ReadLength = 8;
private static final String ACTUAL_RESULT_DIR = "actual";
private static final String PATH_TO_HADOOP_CONF = "src/test/resources/hadoop/conf";
@@ -41,6 +42,8 @@
private static final String EXPECTED_DIR = "src/test/resources/expected/";
private static final String EXPECTED_READER_RESULT = EXPECTED_DIR + "result_after_initial_read";
private static final String EXPECTED_OUPUT_KMER = EXPECTED_DIR + "result_after_kmerAggregate";
+ private static final String EXPECTED_KMER_TO_READID = EXPECTED_DIR + "result_after_kmer2readId";
+ private static final String EXPECTED_GROUPBYREADID = EXPECTED_DIR + "result_after_readIDAggreage";
private static final String DUMPED_RESULT = ACTUAL_RESULT_DIR + HDFS_OUTPUT_PATH + "/merged.txt";
private static final String CONVERT_RESULT = DUMPED_RESULT + ".txt";
@@ -56,36 +59,40 @@
@Test
public void TestAll() throws Exception {
//TestReader();
- TestGroupbyKmer();
- // TestMapKmerToRead();
- // TestGroupByReadID();
+ //TestGroupbyKmer();
+ //TestMapKmerToRead();
+ TestGroupByReadID();
// TestEndToEnd();
}
public void TestReader() throws Exception {
+ cleanUpReEntry();
conf.set(GenomixJobConf.OUTPUT_FORMAT, GenomixJobConf.OUTPUT_FORMAT_TEXT);
driver.runJob(new GenomixJobConf(conf), Plan.CHECK_KMERREADER, true);
- Assert.assertEquals(true, checkResults(EXPECTED_READER_RESULT, false));
+ Assert.assertEquals(true, checkResults(EXPECTED_READER_RESULT, -1));
}
public void TestGroupbyKmer() throws Exception {
+ cleanUpReEntry();
conf.set(GenomixJobConf.OUTPUT_FORMAT, GenomixJobConf.OUTPUT_FORMAT_TEXT);
conf.set(GenomixJobConf.GROUPBY_TYPE, GenomixJobConf.GROUPBY_TYPE_PRECLUSTER);
driver.runJob(new GenomixJobConf(conf), Plan.OUTPUT_KMERHASHTABLE, true);
- Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER, true));
+ Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER, 1));
}
public void TestMapKmerToRead() throws Exception {
+ cleanUpReEntry();
conf.set(GenomixJobConf.OUTPUT_FORMAT, GenomixJobConf.OUTPUT_FORMAT_TEXT);
driver.runJob(new GenomixJobConf(conf), Plan.OUTPUT_MAP_KMER_TO_READ, true);
- Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER,false));
+ Assert.assertEquals(true, checkResults(EXPECTED_KMER_TO_READID, 2));
}
public void TestGroupByReadID() throws Exception {
+ cleanUpReEntry();
conf.set(GenomixJobConf.OUTPUT_FORMAT, GenomixJobConf.OUTPUT_FORMAT_TEXT);
- conf.set(GenomixJobConf.GROUPBY_TYPE, GenomixJobConf.GROUPBY_TYPE_EXTERNAL);
+ conf.set(GenomixJobConf.GROUPBY_TYPE, GenomixJobConf.GROUPBY_TYPE_PRECLUSTER);
driver.runJob(new GenomixJobConf(conf), Plan.OUTPUT_GROUPBY_READID, true);
- Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER,false));
+ Assert.assertEquals(true, checkResults(EXPECTED_GROUPBYREADID, -1));
}
public void TestEndToEnd() throws Exception {
@@ -93,11 +100,11 @@
cleanUpReEntry();
conf.set(GenomixJobConf.GROUPBY_TYPE, GenomixJobConf.GROUPBY_TYPE_EXTERNAL);
driver.runJob(new GenomixJobConf(conf), Plan.BUILD_DEBRUJIN_GRAPH, true);
- Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER,false));
+ Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER, -1));
cleanUpReEntry();
conf.set(GenomixJobConf.GROUPBY_TYPE, GenomixJobConf.GROUPBY_TYPE_PRECLUSTER);
driver.runJob(new GenomixJobConf(conf), Plan.BUILD_DEBRUJIN_GRAPH, true);
- Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER,false));
+ Assert.assertEquals(true, checkResults(EXPECTED_OUPUT_KMER, -1));
}
@Before
@@ -112,6 +119,7 @@
FileOutputFormat.setOutputPath(conf, new Path(HDFS_OUTPUT_PATH));
conf.setInt(GenomixJobConf.KMER_LENGTH, KmerSize);
+ conf.setInt(GenomixJobConf.READ_LENGTH, ReadLength);
driver = new Driver(edu.uci.ics.hyracks.hdfs.utils.HyracksUtils.CC_HOST,
edu.uci.ics.hyracks.hdfs.utils.HyracksUtils.TEST_HYRACKS_CC_CLIENT_PORT, numPartitionPerMachine);
}
@@ -156,7 +164,7 @@
}
}
- private boolean checkResults(String expectedPath, boolean checkPos) throws Exception {
+ private boolean checkResults(String expectedPath, int poslistField) throws Exception {
File dumped = null;
String format = conf.get(GenomixJobConf.OUTPUT_FORMAT);
if (GenomixJobConf.OUTPUT_FORMAT_TEXT.equalsIgnoreCase(format)) {
@@ -202,8 +210,8 @@
dumped = new File(CONVERT_RESULT);
}
- if (checkPos) {
- TestUtils.compareWithUnSortedPosition(new File(expectedPath), dumped);
+ if (poslistField > 0) {
+ TestUtils.compareWithUnSortedPosition(new File(expectedPath), dumped, poslistField);
} else {
TestUtils.compareWithSortedResult(new File(expectedPath), dumped);
}
diff --git a/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/TestUtils.java b/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/TestUtils.java
index e97df1b..49e21ba 100644
--- a/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/TestUtils.java
+++ b/genomix/genomix-hyracks/src/test/java/edu/uci/ics/genomix/hyracks/test/TestUtils.java
@@ -61,7 +61,8 @@
}
}
- public static void compareWithUnSortedPosition(File expectedFile, File actualFile) throws Exception {
+ public static void compareWithUnSortedPosition(File expectedFile, File actualFile, int poslistField)
+ throws Exception {
BufferedReader readerActual = new BufferedReader(new FileReader(actualFile));
BufferedReader readerExpected = new BufferedReader(new FileReader(expectedFile));
ArrayList<String> actualLines = new ArrayList<String>();
@@ -77,7 +78,7 @@
if (lineExpected == null) {
throw new Exception("Actual result changed at line " + num + ":\n< " + actualLine + "\n> ");
}
- if (!containStrings(lineExpected, actualLine)) {
+ if (!containStrings(lineExpected, actualLine, poslistField)) {
throw new Exception("Result for changed at line " + num + ":\n< " + lineExpected + "\n> "
+ actualLine);
}
@@ -162,24 +163,32 @@
return true;
}
- private static boolean containStrings(String lineExpected, String actualLine) {
- String keyExp = lineExpected.split("\\\t")[0];
- String keyAct = actualLine.split("\\\t")[0];
- if (!keyAct.equals(keyExp)) {
+ private static boolean containStrings(String lineExpected, String actualLine, int poslistField) {
+ String[] fieldsExp = lineExpected.split("\\\t");
+ String[] fieldsAct = actualLine.split("\\\t");
+ if (fieldsAct.length != fieldsExp.length) {
return false;
}
+ for (int i = 0; i < fieldsAct.length; i++) {
+ if (i == poslistField) {
+ continue;
+ }
+ if (!fieldsAct[i].equals(fieldsExp[i])) {
+ return false;
+ }
+ }
ArrayList<String> posExp = new ArrayList<String>();
ArrayList<String> posAct = new ArrayList<String>();
- String valueExp = lineExpected.split("\\\t")[1];
+ String valueExp = lineExpected.split("\\\t")[poslistField];
String[] valuesExp = valueExp.substring(1, valueExp.length() - 1).split(",");
for (String str : valuesExp) {
posExp.add(str);
}
- String valueAct = actualLine.split("\\\t")[1];
+ String valueAct = actualLine.split("\\\t")[poslistField];
String[] valuesAct = valueAct.substring(1, valueAct.length() - 1).split(",");
for (String str : valuesAct) {
diff --git a/genomix/genomix-hyracks/src/test/resources/data/webmap/text.txt b/genomix/genomix-hyracks/src/test/resources/data/webmap/text.txt
index 08f0f95..13190dd 100755
--- a/genomix/genomix-hyracks/src/test/resources/data/webmap/text.txt
+++ b/genomix/genomix-hyracks/src/test/resources/data/webmap/text.txt
@@ -2,4 +2,5 @@
2 AATAGAAG
3 AATAGAAG
4 AATAGAAG
-5 AATAGAAG
\ No newline at end of file
+5 AATAGAAG
+6 AGAAGAAG
diff --git a/genomix/genomix-hyracks/src/test/resources/expected/result_after_initial_read b/genomix/genomix-hyracks/src/test/resources/expected/result_after_initial_read
index 728c093..1091d2e 100644
--- a/genomix/genomix-hyracks/src/test/resources/expected/result_after_initial_read
+++ b/genomix/genomix-hyracks/src/test/resources/expected/result_after_initial_read
@@ -1,3 +1,4 @@
+AAGAA (6,2)
AATAG (1,0)
AATAG (2,0)
AATAG (3,0)
@@ -8,13 +9,16 @@
AGAAG (3,3)
AGAAG (4,3)
AGAAG (5,3)
+AGAAG (6,0)
+AGAAG (6,3)
ATAGA (1,1)
ATAGA (2,1)
ATAGA (3,1)
ATAGA (4,1)
ATAGA (5,1)
+GAAGA (6,1)
TAGAA (1,2)
TAGAA (2,2)
TAGAA (3,2)
TAGAA (4,2)
-TAGAA (5,2)
\ No newline at end of file
+TAGAA (5,2)
diff --git a/genomix/genomix-hyracks/src/test/resources/expected/result_after_kmer2readId b/genomix/genomix-hyracks/src/test/resources/expected/result_after_kmer2readId
new file mode 100644
index 0000000..0ca9de6
--- /dev/null
+++ b/genomix/genomix-hyracks/src/test/resources/expected/result_after_kmer2readId
@@ -0,0 +1,24 @@
+1 0 [] AATAG
+1 1 [] ATAGA
+1 2 [] TAGAA
+1 3 [(6,0)] AGAAG
+2 0 [] AATAG
+2 1 [] ATAGA
+2 2 [] TAGAA
+2 3 [(6,0)] AGAAG
+3 0 [] AATAG
+3 1 [] ATAGA
+3 2 [] TAGAA
+3 3 [(6,0)] AGAAG
+4 0 [] AATAG
+4 1 [] ATAGA
+4 2 [] TAGAA
+4 3 [(6,0)] AGAAG
+5 0 [] AATAG
+5 1 [] ATAGA
+5 2 [] TAGAA
+5 3 [(6,0)] AGAAG
+6 0 [(1,3),(2,3),(3,3),(4,3),(5,3),(6,3)] AGAAG
+6 1 [] GAAGA
+6 2 [] AAGAA
+6 3 [(6,0)] AGAAG
\ No newline at end of file
diff --git a/genomix/genomix-hyracks/src/test/resources/expected/result_after_kmerAggregate b/genomix/genomix-hyracks/src/test/resources/expected/result_after_kmerAggregate
index d5624d7..499200a 100644
--- a/genomix/genomix-hyracks/src/test/resources/expected/result_after_kmerAggregate
+++ b/genomix/genomix-hyracks/src/test/resources/expected/result_after_kmerAggregate
@@ -1,4 +1,6 @@
+AAGAA [(6,2)]
AATAG [(1,0),(2,0),(3,0),(4,0),(5,0)]
-AGAAG [(1,3),(2,3),(3,3),(4,3),(5,3)]
+AGAAG [(1,3),(2,3),(3,3),(4,3),(5,3),(6,0),(6,3)]
ATAGA [(1,1),(2,1),(3,1),(4,1),(5,1)]
-TAGAA [(1,2),(2,2),(3,2),(4,2),(5,2)]
+GAAGA [(6,1)]
+TAGAA [(1,2),(2,2),(3,2),(4,2),(5,2)]
\ No newline at end of file
diff --git a/genomix/genomix-hyracks/src/test/resources/expected/result_after_readIDAggreage b/genomix/genomix-hyracks/src/test/resources/expected/result_after_readIDAggreage
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-hyracks/src/test/resources/expected/result_after_readIDAggreage