[ADAM-625] Enable globbing for BAM.

adam-cli/src/main/scala/org/bdgenomics/adam/cli/Transform.scala

@@ -85,6 +85,14 @@ class TransformArgs extends Args4jBase with ADAMSaveAnyArgs with ParquetArgs {
   var coalesce: Int = -1
   @Args4jOption(required = false, name = "-sort_fastq_output", usage = "Sets whether to sort the FASTQ output, if saving as FASTQ. False by default. Ignored if not saving as FASTQ.")
   var sortFastqOutput: Boolean = false
+  @Args4jOption(required = false, name = "-force_load_bam", usage = "Forces Transform to load from BAM/SAM.")
+  var forceLoadBam: Boolean = false
+  @Args4jOption(required = false, name = "-force_load_fastq", usage = "Forces Transform to load from unpaired FASTQ.")
+  var forceLoadFastq: Boolean = false
+  @Args4jOption(required = false, name = "-force_load_ifastq", usage = "Forces Transform to load from interleaved FASTQ.")
+  var forceLoadIFastq: Boolean = false
+  @Args4jOption(required = false, name = "-force_load_parquet", usage = "Forces Transform to load from Parquet.")
+  var forceLoadParquet: Boolean = false
 }
 
 class Transform(protected val args: TransformArgs) extends ADAMSparkCommand[TransformArgs] with Logging {
@@ -155,7 +163,19 @@ class Transform(protected val args: TransformArgs) extends ADAMSparkCommand[Tran
   }
 
   def run(sc: SparkContext, job: Job) {
-    this.apply(sc.loadAlignments(args.inputPath)).adamSave(args)
+    this.apply({
+      if (args.forceLoadBam) {
+        sc.loadBam(args.inputPath)
+      } else if (args.forceLoadFastq) {
+        sc.loadUnpairedFastq(args.inputPath)
+      } else if (args.forceLoadIFastq) {
+        sc.loadInterleavedFastq(args.inputPath)
+      } else if (args.forceLoadParquet) {
+        sc.loadParquetAlignments(args.inputPath)
+      } else {
+        sc.loadAlignments(args.inputPath)
+      }
+    }).adamSave(args)
   }
 
   private def createKnownSnpsTable(sc: SparkContext): SnpTable = CreateKnownSnpsTable.time {

adam-core/src/main/scala/org/bdgenomics/adam/models/RecordGroupDictionary.scala

@@ -71,6 +71,10 @@ class RecordGroupDictionary(val recordGroups: Seq[RecordGroup]) extends Serializ
   assert(recordGroupMap.size == recordGroups.length,
     "Read group dictionary contains multiple samples with identical read group names.")
 
+  def ++(that: RecordGroupDictionary): RecordGroupDictionary = {
+    new RecordGroupDictionary(recordGroups ++ that.recordGroups)
+  }
+
   /**
    * Returns the numerical index for a given record group name.
    *

adam-core/src/main/scala/org/bdgenomics/adam/models/SequenceDictionary.scala

@@ -94,7 +94,6 @@ class SequenceDictionary(val records: Vector[SequenceRecord]) extends Serializab
 
   def +(record: SequenceRecord): SequenceDictionary = this ++ SequenceDictionary(record)
   def ++(that: SequenceDictionary): SequenceDictionary = {
-    assert(this.isCompatibleWith(that))
     new SequenceDictionary(records ++ that.records.filter(r => !byName.contains(r.name)))
   }
 

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

@@ -229,12 +229,29 @@ class ADAMContext(val sc: SparkContext) extends Serializable with Logging {
   def loadBam(
     filePath: String): RDD[AlignmentRecord] = {
 
-    // We need to separately read the header, so that we can inject the sequence dictionary
-    // data into each individual Read (see the argument to samRecordConverter.convert,
-    // below).
-    val samHeader = SAMHeaderReader.readSAMHeaderFrom(new Path(filePath), sc.hadoopConfiguration)
-    val seqDict = adamBamDictionaryLoad(samHeader)
-    val readGroups = adamBamLoadReadGroups(samHeader)
+    val (seqDict, readGroups) = FileSystem.get(sc.hadoopConfiguration)
+      .globStatus(new Path(filePath))
+      .map(fs => fs.getPath)
+      .flatMap(fp => {
+
+        try {
+          // We need to separately read the header, so that we can inject the sequence dictionary
+          // data into each individual Read (see the argument to samRecordConverter.convert,
+          // below).
+          val samHeader = SAMHeaderReader.readSAMHeaderFrom(fp, sc.hadoopConfiguration)
+          log.info("Loaded header from " + fp)
+          val sd = adamBamDictionaryLoad(samHeader)
+          val rg = adamBamLoadReadGroups(samHeader)
+          Some((sd, rg))
+        } catch {
+          case _: Throwable => {
+            log.error("Loading failed for " + fp)
+            None
+          }
+        }
+      }).reduce((kv1, kv2) => {
+        (kv1._1 ++ kv2._1, kv1._2 ++ kv2._2)
+      })
 
     val job = HadoopUtil.newJob(sc)
     val records = sc.newAPIHadoopFile(filePath, classOf[AnySAMInputFormat], classOf[LongWritable],