[ADAM-1530] Support loading GO:query (S/CR/B)AMs as fragments.

Resolves #1530. Adds support for loading (S/CR/B)AM files that are globally
unsorted yet have group order set to query (read name) as Fragments.
Additionally, adds two unit tests. In the process of running the unit tests,
I found an issue in `SingleReadBucket` that was dropping a read from each
pair after the first pair in a partition due to an odd behavior in the
`Iterator.takeWhile` function, which drops the first item that does not
get taken.

adam-core/src/main/scala/org/bdgenomics/adam/rdd/ADAMContext.scala

@@ -488,10 +488,11 @@ class ADAMContext(@transient val sc: SparkContext) extends Serializable with Log
   }
 
   /**
-   * Checks to see if a set of BAM/CRAM/SAM files are queryname sorted.
+   * Checks to see if a set of BAM/CRAM/SAM files are queryname grouped.
    *
    * If we are loading fragments and the BAM/CRAM/SAM files are sorted by the
-   * read names, this implies that all of the reads in a pair are consecutive in
+   * read names, or the file is unsorted but is query grouped, this implies
+   * that all of the reads in a pair are consecutive in
    * the file. If this is the case, we can configure Hadoop-BAM to keep all of
    * the reads from a fragment in a single split. This allows us to eliminate
    * an expensive groupBy when loading a BAM file as fragments.
@@ -503,7 +504,7 @@ class ADAMContext(@transient val sc: SparkContext) extends Serializable with Log
    * @return Returns true if all files described by the path name are queryname
    *   sorted.
    */
-  private[rdd] def filesAreQuerynameSorted(
+  private[rdd] def filesAreQueryGrouped(
     pathName: String,
     stringency: ValidationStringency = ValidationStringency.STRICT): Boolean = {
 
@@ -521,7 +522,8 @@ class ADAMContext(@transient val sc: SparkContext) extends Serializable with Log
           sc.hadoopConfiguration.set(SAMHeaderReader.VALIDATION_STRINGENCY_PROPERTY, stringency.toString)
           val samHeader = SAMHeaderReader.readSAMHeaderFrom(fp, sc.hadoopConfiguration)
 
-          samHeader.getSortOrder == SAMFileHeader.SortOrder.queryname
+          (samHeader.getSortOrder == SAMFileHeader.SortOrder.queryname ||
+            samHeader.getGroupOrder == SAMFileHeader.GroupOrder.query)
         } catch {
           case e: Throwable => {
             log.error(
@@ -1793,7 +1795,7 @@ class ADAMContext(@transient val sc: SparkContext) extends Serializable with Log
     val trimmedPathName = trimExtensionIfCompressed(pathName)
     if (isBamExt(trimmedPathName)) {
       // check to see if the input files are all queryname sorted
-      if (filesAreQuerynameSorted(pathName)) {
+      if (filesAreQueryGrouped(pathName)) {
         log.info(s"Loading $pathName as queryname sorted BAM/CRAM/SAM and converting to Fragments.")
         loadBam(pathName).transform(RepairPartitions(_))
           .querynameSortedToFragments

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

@@ -26,7 +26,9 @@ import org.bdgenomics.formats.avro.{
   AlignmentRecord,
   Fragment
 }
+import scala.annotation.tailrec
 import scala.collection.JavaConversions._
+import scala.collection.mutable.ListBuffer
 
 private class FragmentIterator(
     reads: Iterator[AlignmentRecord]) extends Iterator[Iterable[AlignmentRecord]] with Serializable {
@@ -42,8 +44,18 @@ private class FragmentIterator(
     // get the read name
     val readName = readIter.head.getReadName
 
-    // take the reads that have this read name
-    readIter.takeWhile(_.getReadName == readName).toIterable
+    @tailrec def getReads(
+      l: ListBuffer[AlignmentRecord]): Iterable[AlignmentRecord] = {
+
+      if (!readIter.hasNext ||
+        readIter.head.getReadName != readName) {
+        l.toIterable
+      } else {
+        getReads(l += readIter.next)
+      }
+    }
+
+    getReads(ListBuffer.empty)
   }
 }
 

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

@@ -0,0 +1,23 @@
+@HD	VN:1.3	SO:queryname
+@SQ	SN:1	LN:249250621
+@SQ	SN:2	LN:243199373
+simread:1:5469106:true	0	1	5469107	60	75M	*	0	0	CTCATTCTCTCTCCTGCTGCACTGTGAAGAGGTGCCTGTTGCCAAGAGTATAAGTTTCCTGAGGCCTCCCAGGCC	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:14397233:false	16	1	14397234	60	75M	*	0	0	TAAAATGCCCCCATCTTCCCAGAGCTGCCAGCCCTCACAATGCCAACAGCTAAATGTACCCAAGTGTTACTGAAC	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:20101800:true	0	1	20101801	35	75M	*	0	0	CTCAGGTGATCCACCCGCCTCGGCCTCCCAAAGTGCTGGGACTACAGGCATGAGGCACCGCGCCTGGCCAGGACT	*	NM:i:0	AS:i:75	XS:i:55
+simread:1:26472783:false	16	1	26472784	60	75M	*	0	0	GTATAAGAGCAGCCTTATTCCTATTTATAATCAGGGTGAAACACCTGTGCCAATGCCAAGACAGGGGTGCCAAGA	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:37577445:false	16	1	37577446	60	75M	*	0	0	CCTAGAGAAGCTCCCACTAGGGCTGCAGTCAATTCCCAGGTCTTAGGTGCTGAGCAGTGGGAGGTGGTGGCCATG	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:50683371:false	16	1	50683372	60	75M	*	0	0	GCTCAGGCCTTGCAAGAATCTCTACTGCCCAACAAGTCCCTACAAGATGGCATTTAAAAGCAGTCCCTCACGCAC	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:89554252:false	16	1	89554253	60	75M	*	0	0	AAATTAAACAGCTCGTTTAACTGATAATCCATACTATATTTGAGTAGGGCTGTCACATGGTTGGAACCTCCGGTT	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:101556378:false	16	1	101556379	60	75M	*	0	0	TTTATTTTTTGAGCATGAAAGTAATATATGCTCAGTGTAAACAATTAGGTCATTATAAATATATTTAACAGGAAT	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:153978724:false	16	1	153978725	60	75M	*	0	0	GCTCACTGCAGCCTCAACCTCCTGGGCCCAAGTGATTTCATCTTATTTTTGGAAAAAAAAACAAACTAAACCAAA	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:163841413:false	16	1	163841414	60	75M	*	0	0	TGTGTAACTAACATAATTGGCACTGTCCCTGTAAATTCAAATTGGATATCCTCCCAAATTTTATTTAAGCAATTG	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:165341382:true	0	1	165341383	60	75M	*	0	0	CTACTCTCATTGACTGTTCAATGCCTATACAAGTAAAACTTTACCAGCACCCAAGTCAAAAAGAAAAAAAAGGGG	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:169801933:true	0	1	169801934	40	75M	*	0	0	AGACTGGGTCTCACTATGTTGCCTAGGCTGGTCTCAAACTCCTGGGCTCAAGTGATCCATCTCTGCCTTCCAAAG	*	NM:i:0	AS:i:75	XS:i:52
+simread:1:186794283:true	0	1	186794284	60	75M	*	0	0	GACAAGATAGTACTTGAGCTAAGCCTTGCAGGTTGAGTAGGATTATTCTAGTGGAATTTAGGGAAACGATGTGCA	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:189606653:true	0	1	189606654	60	75M	*	0	0	TGTATCTTCCTCCCCTGCTGTATGTTTCCTGCCCTCAAACATCACACTCCACGTTCTTCAGCTTTAGGACTTGGA	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:195211965:false	16	1	195211966	60	75M	*	0	0	AAATAAAGTTTGGCTTTCAGTTGTAACTTTGAATATCTTTATCACAGTTATTTAAAGCCTTTAAAAAGCTTTAAT	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:207027738:true	0	1	207027739	60	75M	*	0	0	TTTAATAAATGTTGATTGTCCTATTTAATTATTCTCAACTTTCCGATTTTATTTCCCATGTAACAGTGTTGTTTT	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:231911906:false	16	1	231911907	60	75M	*	0	0	TCATGTAGCATGCATATGGCTAACGGCAAAGTGAGGGAGGAATAATTATAGTAATAATCACAGTGATGACGTGGA	*	NM:i:0	AS:i:75	XS:i:0
+simread:1:237728409:true	0	1	237728410	28	75M	*	0	0	TTTCTTTTTCTTTCTTTCTTTCTTTCTTTCTTTTTCTTTCTTTCTTTCTCTTTCTTTCTTTCTTTCTTTCTTTCT	*	NM:i:0	AS:i:75	XS:i:59
+simread:1:240344442:true	0	1	240344443	24	75M	*	0	0	TACAGGCACCCACCATCATGCCCAGCTAATTTTTGTATTTTTGTAGAAACGGGGTTTCACCATGTTGGCCCAGCT	*	NM:i:0	AS:i:75	XS:i:61
+simread:1:240997787:true	0	1	240997788	60	75M	*	0	0	CTTTATTTTTATTTTTAAGGTTTTTTTTGTTTGTTTGTTTTGAGATGGAGTCTCGCTCCACCGCCCAGACTGGAG	*	NM:i:0	AS:i:75	XS:i:39

adam-core/src/test/resources/sample1.query.sam

@@ -0,0 +1,7 @@
+@HD	VN:1.5	SO:unsorted	GO:query
+H06HDADXX130110:2:2116:3345:91806	77	*	0	0	*	*	0	0	GTTAGGGTTAGGGTTGGGTTAGGGTTAGGGTTAGGGTTAGGGGTAGGGTTAGGGTTAGGGGTAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGGTAGGGCTAGGGTTAAGGGTAGGGTTAGCGAAAGGGCTGGGGTTAGGGGTGCGGGTACGCGTAGCATTAGGGCTAGAAGTAGGATCTGCAGTGCCTGACCGCGTCTGCGCGGCGACTGCCCAAAGCCTGGGGCCGACTCCAGGCTGAAGCTCAT	>=<=???>?>???=??>>8<?><=2=<===1194<?;:?>>?#3==>###########################################################################################################################################################################################################
+H06HDADXX130110:2:2116:3345:91806	141	*	0	0	*	*	0	0	TAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTACCCCTAACCCTAACCCTAACCCTAACCCGTACCCTAAACCCAACCCTAACCACAAAGCAAATCCCAACCTTAACCGGAACCCGAAATCTCGCAGCAAATCTGCAGTAGAGACGCAGACTCAACCATGCGTCTATTAGTACGCATTATCATTGCCTCATGCTTCTTAAGTACAGAGAGATGAC	==;<?>@@@<>>@??<>>???<=>>?>:><@?4=:>7=5=>:<=@;'@A?########################################################################################################################################################################################################
+H06HDADXX130110:1:2103:11970:57672	77	*	0	0	*	*	0	0	GGATAGGGTTAGGGTTAGGGTTAGGGCTAGGGATAGGGGTAGGGTTGGGGTTGGTCATCGGGTGTTTCTTTGTGTTTGAGGTTGATTATTGTGATGGTTAAGGTATCTAGGTATTGTAAAAGTTGGCTTTTAACTTAGAAAATTATGTCATTCTGTTCACAAGTGTTTAGATTGGTAGATAGGTACTATGCGATCACTTCCATTGGCTGAGAGTTCGATTGATTATGAGCCACGCTAGTGGTTGAGATCT	69+26933-:7;;135,53<>7<692(?2=9:**;<=#####################################################################################################################################################################################################################
+H06HDADXX130110:1:2103:11970:57672	141	*	0	0	*	*	0	0	AACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTATCGTCAAACCTTACCTCCTCCCTAGCCTCCACCCTGACCATGACACCAACCATCAGCCTTATAGAAAACCCCAGAGATGCTCTTATCCTATACCACAATTACCCCATAACGAAAGAAAGGACTGAAAACAAATAAGTAAAATTCGTACAAATTATATCTATGAGTATGTCCCTGAGTGTAGGTGTAGGTGCATCC	=>:=>@=?<>>??>;:<?<=;<<?>=;:8;=(5)0-6;1:>?<>##############################################################################################################################################################################################################
+H06JUADXX130110:1:1108:6424:55322	77	*	0	0	*	*	0	0	AACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACTCTAACCCTAACCCTAACCCTAACGGTAACCCTTACCCTTACTGTAACGCTTATCCTAAATCAAATTCTTCCTCTTAAGATCGCTGTTAAAATTAATCCTATTAGAACAGGTCTTCTGGCACCAAGTTATGTCAATATCCCTTACTCTAAACATGCCTTGATCTCTCATGCATCACTTCAGCACAGCTCTTATGGATCTAGGATCCTCAGT	=>;=?=@@=?@?@@9>7@=?=;=?@>29?=?;=>@;4@*0878;40'=@;(3399@9>7@:A############################################################################################################################################################################################
+H06JUADXX130110:1:1108:6424:55322	141	*	0	0	*	*	0	0	AGGGATAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGATAGGGCTAGGGTTAGGGATAGGGATAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTATCGATAGGGATAGGGATAGGGATAGAGTTAGGGCTATGGGTAGGGTTAGAGTCAGGGAAAGAGATAGGGATGGAGATGGGGTTAAAAAGAAGTCAAGGAATTAAGGTAGGGAAACGGTTCGAGATCTGTAAAGGGCAACGA	>>;>*9?:@??@@????@????>@?>>@>@?>?????@@???????=<??8;*;:>?;+A?@?>89?@######################################################################################################################################################################################

adam-core/src/test/resources/sample1.queryname.sam

@@ -0,0 +1,7 @@
+@HD	VN:1.5	SO:queryname
+H06HDADXX130110:1:2103:11970:57672	77	*	0	0	*	*	0	0	GGATAGGGTTAGGGTTAGGGTTAGGGCTAGGGATAGGGGTAGGGTTGGGGTTGGTCATCGGGTGTTTCTTTGTGTTTGAGGTTGATTATTGTGATGGTTAAGGTATCTAGGTATTGTAAAAGTTGGCTTTTAACTTAGAAAATTATGTCATTCTGTTCACAAGTGTTTAGATTGGTAGATAGGTACTATGCGATCACTTCCATTGGCTGAGAGTTCGATTGATTATGAGCCACGCTAGTGGTTGAGATCT	69+26933-:7;;135,53<>7<692(?2=9:**;<=#####################################################################################################################################################################################################################
+H06HDADXX130110:1:2103:11970:57672	141	*	0	0	*	*	0	0	AACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTATCGTCAAACCTTACCTCCTCCCTAGCCTCCACCCTGACCATGACACCAACCATCAGCCTTATAGAAAACCCCAGAGATGCTCTTATCCTATACCACAATTACCCCATAACGAAAGAAAGGACTGAAAACAAATAAGTAAAATTCGTACAAATTATATCTATGAGTATGTCCCTGAGTGTAGGTGTAGGTGCATCC	=>:=>@=?<>>??>;:<?<=;<<?>=;:8;=(5)0-6;1:>?<>##############################################################################################################################################################################################################
+H06HDADXX130110:2:2116:3345:91806	77	*	0	0	*	*	0	0	GTTAGGGTTAGGGTTGGGTTAGGGTTAGGGTTAGGGTTAGGGGTAGGGTTAGGGTTAGGGGTAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGGTAGGGCTAGGGTTAAGGGTAGGGTTAGCGAAAGGGCTGGGGTTAGGGGTGCGGGTACGCGTAGCATTAGGGCTAGAAGTAGGATCTGCAGTGCCTGACCGCGTCTGCGCGGCGACTGCCCAAAGCCTGGGGCCGACTCCAGGCTGAAGCTCAT	>=<=???>?>???=??>>8<?><=2=<===1194<?;:?>>?#3==>###########################################################################################################################################################################################################
+H06HDADXX130110:2:2116:3345:91806	141	*	0	0	*	*	0	0	TAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACCCTACCCCTAACCCTAACCCTAACCCTAACCCGTACCCTAAACCCAACCCTAACCACAAAGCAAATCCCAACCTTAACCGGAACCCGAAATCTCGCAGCAAATCTGCAGTAGAGACGCAGACTCAACCATGCGTCTATTAGTACGCATTATCATTGCCTCATGCTTCTTAAGTACAGAGAGATGAC	==;<?>@@@<>>@??<>>???<=>>?>:><@?4=:>7=5=>:<=@;'@A?########################################################################################################################################################################################################
+H06JUADXX130110:1:1108:6424:55322	77	*	0	0	*	*	0	0	AACCCTAACCCTAACCCTAACCCTAACCCTAACCCTAACTCTAACCCTAACCCTAACCCTAACGGTAACCCTTACCCTTACTGTAACGCTTATCCTAAATCAAATTCTTCCTCTTAAGATCGCTGTTAAAATTAATCCTATTAGAACAGGTCTTCTGGCACCAAGTTATGTCAATATCCCTTACTCTAAACATGCCTTGATCTCTCATGCATCACTTCAGCACAGCTCTTATGGATCTAGGATCCTCAGT	=>;=?=@@=?@?@@9>7@=?=;=?@>29?=?;=>@;4@*0878;40'=@;(3399@9>7@:A############################################################################################################################################################################################
+H06JUADXX130110:1:1108:6424:55322	141	*	0	0	*	*	0	0	AGGGATAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTAGGGATAGGGCTAGGGTTAGGGATAGGGATAGGGTTAGGGTTAGGGTTAGGGTTAGGGTTATCGATAGGGATAGGGATAGGGATAGAGTTAGGGCTATGGGTAGGGTTAGAGTCAGGGAAAGAGATAGGGATGGAGATGGGGTTAAAAAGAAGTCAAGGAATTAAGGTAGGGAAACGGTTCGAGATCTGTAAAGGGCAACGA	>>;>*9?:@??@@????@????>@?>>@>@?>?????@@???????=<??8;*;:>?;+A?@?>89?@######################################################################################################################################################################################

adam-core/src/test/scala/org/bdgenomics/adam/rdd/ADAMContextSuite.scala

@@ -520,4 +520,22 @@ class ADAMContextSuite extends ADAMFunSuite {
     assert(fastqReads2.rdd.count === 488)
     assert(pairedReads.rdd.count === 976)
   }
+
+  sparkTest("load queryname sorted sam as fragments") {
+    val samFile = testFile("sample1.queryname.sam")
+    assert(sc.filesAreQueryGrouped(samFile))
+    val fragments = sc.loadFragments(samFile)
+    assert(fragments.rdd.count === 3)
+    val reads = fragments.toReads
+    assert(reads.rdd.count === 6)
+  }
+
+  sparkTest("load query grouped sam as fragments") {
+    val samFile = testFile("sample1.query.sam")
+    assert(sc.filesAreQueryGrouped(samFile))
+    val fragments = sc.loadFragments(samFile)
+    assert(fragments.rdd.count === 3)
+    val reads = fragments.toReads
+    assert(reads.rdd.count === 6)
+  }
 }