Allow for export of query name sorted SAM files

adam-core/src/main/scala/org/bdgenomics/adam/rdd/read/AlignmentRecordRDD.scala

@@ -638,23 +638,37 @@ sealed abstract class AlignmentRecordRDD extends AvroRecordGroupGenomicDataset[A
     samStringWriter.toString
   }
 
+  /**
+   * Converts boolean sorted state into SortOrder enum.
+   *
+   * @param isSorted Whether the file is sorted or not.
+   * @return Returns coordinate order if sorted, and unsorted otherwise.
+   */
+  private def isSortedToSortOrder(isSorted: Boolean = false): SAMFileHeader.SortOrder = {
+    if (isSorted) {
+      SAMFileHeader.SortOrder.coordinate
+    } else {
+      SAMFileHeader.SortOrder.unsorted
+    }
+  }
+
   /**
    * Converts an RDD of ADAM read records into SAM records.
    *
    * @return Returns a SAM/BAM formatted RDD of reads, as well as the file header.
    */
   def convertToSam(isSorted: Boolean = false): (RDD[SAMRecordWritable], SAMFileHeader) = ConvertToSAM.time {
+    convertToSam(isSortedToSortOrder(isSorted))
+  }
+
+  def convertToSam(sortOrder: SAMFileHeader.SortOrder): (RDD[SAMRecordWritable], SAMFileHeader) = ConvertToSAM.time {
 
     // create conversion object
     val adamRecordConverter = new AlignmentRecordConverter
 
     // create header and set sort order if needed
     val header = adamRecordConverter.createSAMHeader(sequences, recordGroups)
-    if (isSorted) {
-      header.setSortOrder(SAMFileHeader.SortOrder.coordinate)
-    } else {
-      header.setSortOrder(SAMFileHeader.SortOrder.unsorted)
-    }
+    header.setSortOrder(sortOrder)
 
     // get program records and attach to header
     val pgRecords = processingSteps.map(r => {
@@ -746,12 +760,21 @@ sealed abstract class AlignmentRecordRDD extends AvroRecordGroupGenomicDataset[A
     isSorted: Boolean = false,
     deferMerging: Boolean = false,
     disableFastConcat: Boolean = false): Unit = SAMSave.time {
+    saveAsSam(filePath, asType, asSingleFile, isSortedToSortOrder(isSorted), deferMerging, disableFastConcat)
+  }
 
+  def saveAsSam(
+    filePath: String,
+    asType: Option[SAMFormat],
+    asSingleFile: Boolean,
+    sortOrder: SAMFileHeader.SortOrder,
+    deferMerging: Boolean,
+    disableFastConcat: Boolean): Unit = SAMSave.time {
     val fileType = asType.getOrElse(SAMFormat.inferFromFilePath(filePath))
 
     // convert the records
     val (convertRecords: RDD[SAMRecordWritable], header: SAMFileHeader) =
-      convertToSam(isSorted)
+      convertToSam(sortOrder)
 
     // add keys to our records
     val withKey = convertRecords.keyBy(v => new LongWritable(v.get.getAlignmentStart))
@@ -810,7 +833,7 @@ sealed abstract class AlignmentRecordRDD extends AvroRecordGroupGenomicDataset[A
       // we'll defer the writing to the cram container stream writer, and will
       // do validation here
 
-      require(isSorted, "To save as CRAM, input must be sorted.")
+      require(sortOrder != SAMFileHeader.SortOrder.unsorted, "To save as CRAM, input must be sorted.")
       require(sequences.records.forall(_.md5.isDefined),
         "To save as CRAM, all sequences must have an attached MD5. See %s".format(
           sequences))
@@ -914,6 +937,17 @@ sealed abstract class AlignmentRecordRDD extends AvroRecordGroupGenomicDataset[A
       isSorted = isSorted)
   }
 
+  /**
+   * Sorts our read data by read name.
+   *
+   * @return Returns a new RDD containing sorted reads.
+   */
+  def sortReadsByReadName(): AlignmentRecordRDD = SortReads.time {
+    log.info("Sorting reads by read name")
+
+    transformDataset(_.orderBy("readName", "readInFragment"))
+  }
+
   /**
    * Sorts our read data by reference positions, with contigs ordered by name.
    *

adam-core/src/test/resources/readname_sorted.sam

@@ -0,0 +1,10 @@
+@HD	VN:1.5	SO:queryname
+@SQ	SN:1	LN:1000
+@SQ	SN:chr2	LN:1000
+@SQ	SN:3	LN:1000
+@SQ	SN:4	LN:2000
+A	0	1	1	50	10M	*	0	0	ACACACACAC	**********
+B	0	3	11	40	4M2I4M	*	0	0	ACACACACAC	**********
+C	0	4	1001	25	8M	*	0	0	ACACACAC	********
+D	0	chr2	501	55	10M2S	*	0	0	ACACACACACAC	************
+E	0	chr2	101	45	10M	*	0	0	ACACACACAC	**********

adam-core/src/test/scala/org/bdgenomics/adam/rdd/read/AlignmentRecordRDDSuite.scala

@@ -19,7 +19,7 @@ package org.bdgenomics.adam.rdd.read
 
 import java.io.File
 import java.nio.file.Files
-import htsjdk.samtools.ValidationStringency
+import htsjdk.samtools.{ SAMFileHeader, ValidationStringency }
 import org.apache.spark.api.java.function.Function2
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{ Dataset, SQLContext }
@@ -1775,4 +1775,21 @@ class AlignmentRecordRDDSuite extends ADAMFunSuite {
     assert(alignmentsDs.filterToSamples(Seq("NA12878", "not a sample")).dataset.count() === 565)
     assert(alignmentsDs.filterToSamples(Seq("not a sample")).dataset.count() === 0)
   }
+
+  sparkTest("sort by read name") {
+    val unsortedPath = testFile("unsorted.sam")
+    val ardd = sc.loadBam(unsortedPath)
+    val reads = ardd.rdd
+
+    val actualSortedPath = tmpFile("readname_sorted.sam")
+    ardd.sortReadsByReadName()
+      .saveAsSam(actualSortedPath,
+        asType = None,
+        asSingleFile = true,
+        sortOrder = SAMFileHeader.SortOrder.queryname,
+        deferMerging = false,
+        disableFastConcat = false)
+
+    checkFiles(testFile("readname_sorted.sam"), actualSortedPath)
+  }
 }