[ADAM-1040] Remove implicit GenomicRDD to RDD conversion.

Resolves bigdatagenomics#1040.

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

@@ -56,7 +56,7 @@ class ADAM2Fasta(val args: ADAM2FastaArgs) extends BDGSparkCommand[ADAM2FastaArg
  val contigs = contigFragments.mergeFragments()
 
  val cc = if (args.coalesce > 0) {
- if (args.coalesce > contigs.partitions.size || args.forceShuffle) {
+ if (args.coalesce > contigs.rdd.partitions.size || args.forceShuffle) {
  contigs.transform(_.coalesce(args.coalesce, shuffle = true))
  } else {
  contigs.transform(_.coalesce(args.coalesce, shuffle = false))

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

@@ -53,8 +53,8 @@ object AlleleCountHelper extends Serializable {
  case _ => None
  }
 
- def countAlleles(adamVariants: RDD[Genotype], args: AlleleCountArgs) {
- val usefulData = adamVariants.map(p => (
+ def countAlleles(adamVariants: GenotypeRDD, args: AlleleCountArgs) {
+ val usefulData = adamVariants.rdd.map(p => (
  p.getVariant.getContigName,
  p.getVariant.getStart,
  p.getVariant.getReferenceAllele,

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

@@ -46,7 +46,7 @@ class PrintGenes(protected val args: PrintGenesArgs)
  val companion = PrintGenes
 
  def run(sc: SparkContext): Unit = {
- val genes: RDD[Gene] = sc.loadGenes(args.gtfInput)
+ val genes: RDD[Gene] = sc.loadGenes(args.gtfInput).rdd
 
  genes.map(printGene).collect().foreach(println)
  }

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

@@ -127,7 +127,7 @@ class Transform(protected val args: TransformArgs) extends BDGSparkCommand[Trans
  def apply(rdd: AlignmentRecordRDD): AlignmentRecordRDD = {
 
  var adamRecords = rdd
- val sc = rdd.context
+ val sc = rdd.rdd.context
  val sl = StorageLevel.fromString(args.storageLevel)
 
  val stringencyOpt = Option(args.stringency).map(ValidationStringency.valueOf(_))
@@ -192,7 +192,7 @@ class Transform(protected val args: TransformArgs) extends BDGSparkCommand[Trans
 
  if (args.coalesce != -1) {
  log.info("Coalescing the number of partitions to '%d'".format(args.coalesce))
- if (args.coalesce > adamRecords.partitions.size || args.forceShuffle) {
+ if (args.coalesce > adamRecords.rdd.partitions.size || args.forceShuffle) {
  adamRecords = adamRecords.transform(_.coalesce(args.coalesce, shuffle = true))
  } else {
  adamRecords = adamRecords.transform(_.coalesce(args.coalesce, shuffle = false))
@@ -359,6 +359,6 @@ class Transform(protected val args: TransformArgs) extends BDGSparkCommand[Trans
  }
 
  private def createKnownSnpsTable(sc: SparkContext): SnpTable = CreateKnownSnpsTable.time {
- Option(args.knownSnpsFile).fold(SnpTable())(f => SnpTable(sc.loadVariants(f).map(new RichVariant(_))))
+ Option(args.knownSnpsFile).fold(SnpTable())(f => SnpTable(sc.loadVariants(f).rdd.map(new RichVariant(_))))
  }
 }

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

@@ -67,7 +67,7 @@ class Vcf2ADAM(val args: Vcf2ADAMArgs) extends BDGSparkCommand[Vcf2ADAMArgs] wit
 
  val variantContextRdd = sc.loadVcf(args.vcfPath, sdOpt = dictionary)
  var variantContextsToSave = if (args.coalesce > 0) {
- if (args.coalesce > variantContextRdd.partitions.size || args.forceShuffle) {
+ if (args.coalesce > variantContextRdd.rdd.partitions.size || args.forceShuffle) {
  variantContextRdd.transform(_.coalesce(args.coalesce, shuffle = true))
  } else {
  variantContextRdd.transform(_.coalesce(args.coalesce, shuffle = false))

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

@@ -22,6 +22,7 @@ import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import org.bdgenomics.adam.models.VariantContext
 import org.bdgenomics.adam.rdd.ADAMContext._
+import org.bdgenomics.adam.rdd.variation.DatabaseVariantAnnotationRDD
 import org.bdgenomics.adam.rich.RichVariant
 import org.bdgenomics.formats.avro._
 import org.bdgenomics.utils.cli._
@@ -52,14 +53,14 @@ class VcfAnnotation2ADAM(val args: VcfAnnotation2ADAMArgs) extends BDGSparkComma
 
  def run(sc: SparkContext) {
  log.info("Reading VCF file from %s".format(args.vcfFile))
- val annotations: RDD[DatabaseVariantAnnotation] = sc.loadVcfAnnotations(args.vcfFile)
+ val annotations = sc.loadVcfAnnotations(args.vcfFile)
 
  if (args.currentAnnotations != null) {
- val existingAnnotations: RDD[DatabaseVariantAnnotation] = sc.loadVariantAnnotations(args.currentAnnotations)
- val keyedAnnotations = existingAnnotations.keyBy(anno => new RichVariant(anno.getVariant))
- val joinedAnnotations = keyedAnnotations.join(annotations.keyBy(anno => new RichVariant(anno.getVariant)))
+ val existingAnnotations = sc.loadVariantAnnotations(args.currentAnnotations)
+ val keyedAnnotations = existingAnnotations.rdd.keyBy(anno => new RichVariant(anno.getVariant))
+ val joinedAnnotations = keyedAnnotations.join(annotations.rdd.keyBy(anno => new RichVariant(anno.getVariant)))
  val mergedAnnotations = joinedAnnotations.map(kv => VariantContext.mergeAnnotations(kv._2._1, kv._2._2))
- mergedAnnotations.saveAsParquet(args)
+ DatabaseVariantAnnotationRDD(mergedAnnotations, existingAnnotations.sequences).saveAsParquet(args)
  } else {
  annotations.saveAsParquet(args)
  }

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

@@ -169,7 +169,7 @@ class View(val args: ViewArgs) extends BDGSparkCommand[ViewArgs] {
  reads.save(args)
  } else {
  if (args.printCount) {
- println(reads.count())
+ println(reads.rdd.count())
  } else {
  println(reads.saveAsSamString())
  }

adam-cli/src/test/scala/org/bdgenomics/adam/cli/Fasta2ADAMSuite.scala

@@ -28,8 +28,8 @@ class Fasta2ADAMSuite extends ADAMFunSuite {
  val cmd = Fasta2ADAM(Array(inputPath, convertPath)).run(sc)
 
  val contigFragments = sc.loadParquetContigFragments(convertPath)
- assert(contigFragments.count() === 26)
- val first = contigFragments.first()
+ assert(contigFragments.rdd.count() === 26)
+ val first = contigFragments.rdd.first()
  assert(first.getContig.getContigName === "gi|224384749|gb|CM000682.1|")
  assert(first.getDescription === "Homo sapiens chromosome 20, GRCh37 primary reference assembly")
  assert(first.getFragmentNumber === 0)

adam-cli/src/test/scala/org/bdgenomics/adam/cli/Features2ADAMSuite.scala

@@ -80,7 +80,7 @@ class Features2ADAMSuite extends ADAMFunSuite {
 
  val schema = Projection(featureId, contigName, start, end, score)
  val rdd = sc.loadFeatures(outputPath, projection = Some(schema))
- val converted = rdd.collect.toSeq.sortBy(f => f.getStart)
+ val converted = rdd.rdd.collect.toSeq.sortBy(f => f.getStart)
 
  assert(converted.size === 10)
  assert(converted(0).getContigName == "chr5")

adam-cli/src/test/scala/org/bdgenomics/adam/cli/FlagStatSuite.scala

@@ -50,7 +50,7 @@ class FlagStatSuite extends ADAMFunSuite {
  AlignmentRecordField.mapq,
  AlignmentRecordField.failedVendorQualityChecks)
 
- val adamFile: RDD[AlignmentRecord] = sc.loadAlignments(args.inputPath, projection = Some(projection))
+ val adamFile: RDD[AlignmentRecord] = sc.loadAlignments(args.inputPath, projection = Some(projection)).rdd
 
  val (failedVendorQuality, passedVendorQuality) = apply(adamFile)
 

adam-core/src/main/scala/org/bdgenomics/adam/algorithms/consensus/ConsensusGeneratorFromKnowns.scala

@@ -48,7 +48,7 @@ private[adam] class ConsensusGeneratorFromKnowns(file: String,
  * @return Returns an option which wraps an RDD of indel realignment targets.
  */
  def targetsToAdd(): Option[RDD[IndelRealignmentTarget]] = {
- val rdd: RDD[Variant] = sc.loadVariants(file)
+ val rdd: RDD[Variant] = sc.loadVariants(file).rdd
 
  Some(rdd.filter(v => v.getReferenceAllele.length != v.getAlternateAllele.length)
  .map(v => ReferenceRegion(v.getContigName, v.getStart, v.getStart + v.getReferenceAllele.length))

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

@@ -58,7 +58,7 @@ private[adam] object IndelTable {
  * @return Returns a table with the known indels populated.
  */
  def apply(knownIndelsFile: String, sc: SparkContext): IndelTable = {
- val rdd: RDD[Variant] = sc.loadVariants(knownIndelsFile)
+ val rdd: RDD[Variant] = sc.loadVariants(knownIndelsFile).rdd
  apply(rdd)
  }
 

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

@@ -105,8 +105,6 @@ object ADAMContext {
  implicit def iterableToJavaCollection[A](i: Iterable[A]): java.util.Collection[A] = asJavaCollection(i)
 
  implicit def setToJavaSet[A](set: Set[A]): java.util.Set[A] = setAsJavaSet(set)
-
- implicit def genomicRDDToRDD[T, U <: GenomicRDD[T, U]](gRdd: GenomicRDD[T, U]): RDD[T] = gRdd.rdd
 }
 
 import org.bdgenomics.adam.rdd.ADAMContext._
@@ -558,8 +556,8 @@ class ADAMContext(@transient val sc: SparkContext) extends Serializable with Log
 
  stringency match {
  case ValidationStringency.STRICT | ValidationStringency.LENIENT =>
- val count1 = reads1.cache.count
- val count2 = reads2.cache.count
+ val count1 = reads1.rdd.cache.count
+ val count2 = reads2.rdd.cache.count
 
  if (count1 != count2) {
  val msg = s"Fastq 1 ($filePath1) has $count1 reads, fastq 2 ($filePath2) has $count2 reads"
@@ -573,7 +571,7 @@ class ADAMContext(@transient val sc: SparkContext) extends Serializable with Log
  case ValidationStringency.SILENT =>
  }
 
- UnalignedReadRDD.fromRdd(reads1 ++ reads2)
+ UnalignedReadRDD.fromRdd(reads1.rdd ++ reads2.rdd)
  }
 
  def loadUnpairedFastq(
@@ -865,7 +863,7 @@ class ADAMContext(@transient val sc: SparkContext) extends Serializable with Log
  //TODO(ryan): S3ByteAccess
  new TwoBitFile(new LocalFileByteAccess(new File(filePath)))
  } else {
- ReferenceContigMap(loadSequences(filePath, fragmentLength = fragmentLength))
+ ReferenceContigMap(loadSequences(filePath, fragmentLength = fragmentLength).rdd)
  }
  }
 

adam-core/src/test/scala/org/bdgenomics/adam/algorithms/consensus/ConsensusGeneratorFromReadsSuite.scala

@@ -29,7 +29,7 @@ class ConsensusGeneratorFromReadsSuite extends ADAMFunSuite {
 
  def artificial_reads: RDD[AlignmentRecord] = {
  val path = resourcePath("artificial.sam")
- sc.loadAlignments(path)
+ sc.loadAlignments(path).rdd
  }
 
  sparkTest("checking search for consensus list for artificial reads") {

adam-core/src/test/scala/org/bdgenomics/adam/converters/FastaConverterSuite.scala

@@ -210,7 +210,7 @@ class FastaConverterSuite extends ADAMFunSuite {
 
  sparkTest("convert reference fasta file") {
  //Loading "human_g1k_v37_chr1_59kb.fasta"
- val referenceSequences = sc.loadSequences(chr1File, fragmentLength = 10).collect()
+ val referenceSequences = sc.loadSequences(chr1File, fragmentLength = 10).rdd.collect()
  assert(referenceSequences.forall(_.getContig.getContigName.toString == "1"))
  assert(referenceSequences.slice(0, referenceSequences.length - 2).forall(_.getFragmentSequence.length == 10))
 

adam-core/src/test/scala/org/bdgenomics/adam/models/GeneSuite.scala

@@ -35,7 +35,7 @@ class GeneSuite extends ADAMFunSuite {
 
  val fixedParentIds = features.reassignParentIds
 
- val genes = fixedParentIds.toGenes()
+ val genes = fixedParentIds.toGenes().rdd
  assert(genes.count() === 4)
 
  val transcripts = genes.flatMap(_.transcripts)
@@ -51,7 +51,7 @@ class GeneSuite extends ADAMFunSuite {
 
  val fixedParentIds = features.reassignParentIds
 
- val genes = fixedParentIds.toGenes()
+ val genes = fixedParentIds.toGenes().rdd
  assert(genes.count() === 8)
 
  val transcripts = genes.flatMap(_.transcripts)
@@ -102,7 +102,7 @@ class GeneSuite extends ADAMFunSuite {
  val features = sc.loadFeatures(path)
  val fixedParentIds = features.reassignParentIds
  val genes = fixedParentIds.toGenes()
- val transcripts: Seq[Transcript] = genes.flatMap(g => g.transcripts).take(100)
+ val transcripts: Seq[Transcript] = genes.rdd.flatMap(g => g.transcripts).take(100)
 
  transcripts.foreach { transcript =>
  val mySequence = transcript.extractSplicedmRNASequence(chr20sequence.get)

adam-core/src/test/scala/org/bdgenomics/adam/projections/FieldEnumerationSuite.scala

@@ -42,15 +42,15 @@ class FieldEnumerationSuite extends ADAMFunSuite {
  rRdd.saveAsParquet(TestSaveArgs(readsParquetFilepath))
 
  val p1 = Projection(AlignmentRecordField.readName)
- val reads1: RDD[AlignmentRecord] = sc.loadAlignments(readsParquetFilepath, projection = Some(p1))
+ val reads1: RDD[AlignmentRecord] = sc.loadAlignments(readsParquetFilepath, projection = Some(p1)).rdd
  assert(reads1.count() === 200)
 
  val first1 = reads1.first()
  assert(first1.getReadName.toString === "simread:1:26472783:false")
  assert(first1.getReadMapped === false)
 
  val p2 = Projection(AlignmentRecordField.readName, AlignmentRecordField.readMapped)
- val reads2: RDD[AlignmentRecord] = sc.loadAlignments(readsParquetFilepath, projection = Some(p2))
+ val reads2: RDD[AlignmentRecord] = sc.loadAlignments(readsParquetFilepath, projection = Some(p2)).rdd
  assert(reads2.count() === 200)
 
  val first2 = reads2.first()