@@ -2,7 +2,7 @@ package is.hail.expr.ir
22
33import is .hail .backend .spark .SparkBackend
44import is .hail .io .compress .BGzipInputStream
5- import is .hail .io .fs .{BGZipCompressionCodec , FS , FileStatus , Positioned , PositionedInputStream }
5+ import is .hail .io .fs .{getCodecFromPath , BGZipCompressionCodec , FS , FileStatus , Positioned , PositionedInputStream }
66import is .hail .io .tabix .{TabixLineIterator , TabixReader }
77import is .hail .types .virtual .{TBoolean , TInt32 , TInt64 , TString , TStruct , Type }
88import is .hail .utils ._
@@ -43,31 +43,36 @@ object GenericLines extends Logging {
4343 private var splitCompressed = false
4444 private val is : PositionedInputStream = {
4545 val rawIS = fs.openNoCompression(file)
46- val codec = fs.getCodecFromPath(file, gzAsBGZ)
47- if (codec == null ) {
48- assert(split || filePerPartition)
49- rawIS.seek(start)
50- rawIS
51- } else if (codec == BGZipCompressionCodec ) {
52- assert(split || filePerPartition)
53- splitCompressed = true
54- val bgzIS =
55- new BGzipInputStream (rawIS, start, end, SplittableCompressionCodec .READ_MODE .BYBLOCK )
56- new ProxyInputStream (bgzIS) with Positioned {
57- def getPosition : Long = bgzIS.getVirtualOffset
58- }
59- } else {
60- assert(! split || filePerPartition)
46+ getCodecFromPath(file, gzAsBGZ) match {
47+ case None =>
48+ assert(split || filePerPartition)
49+ rawIS.seek(start)
50+ rawIS
51+ case Some (BGZipCompressionCodec ) =>
52+ assert(split || filePerPartition)
53+ splitCompressed = true
54+ val bgzIS =
55+ new BGzipInputStream (
56+ rawIS,
57+ start,
58+ end,
59+ SplittableCompressionCodec .READ_MODE .BYBLOCK ,
60+ )
61+ new ProxyInputStream (bgzIS) with Positioned {
62+ def getPosition : Long = bgzIS.getVirtualOffset
63+ }
64+ case Some (codec) =>
65+ assert(! split || filePerPartition)
6166
6267 val delegate =
6368 BoundedInputStream
6469 .builder()
6570 .setInputStream(codec.makeInputStream(rawIS))
6671 .get()
6772
68- new ProxyInputStream (delegate) with Positioned {
69- override def getPosition : Long = delegate.getCount
70- }
73+ new ProxyInputStream (delegate) with Positioned {
74+ override def getPosition : Long = delegate.getCount
75+ }
7176 }
7277 }
7378
@@ -291,9 +296,9 @@ object GenericLines extends Logging {
291296
292297 val contexts = fileStatuses.flatMap { case (fileListEntry, fileNum) =>
293298 val size = fileListEntry.getLen
294- val codec = fs. getCodecFromPath(fileListEntry.getPath, gzAsBGZ)
299+ val codec = getCodecFromPath(fileListEntry.getPath, gzAsBGZ)
295300
296- val splittable = codec == null || codec == BGZipCompressionCodec
301+ val splittable = codec.isEmpty || codec.contains( BGZipCompressionCodec )
297302 if (splittable && ! filePerPartition) {
298303 var fileNParts = ((totalPartitions.toDouble * size) / totalSize + 0.5 ).toInt
299304 if (fileNParts == 0 )
@@ -305,7 +310,7 @@ object GenericLines extends Logging {
305310 .map { i =>
306311 val start = partScan(i)
307312 var end = partScan(i + 1 )
308- if (codec != null )
313+ if (codec.isDefined )
309314 end = makeVirtualOffset(end, 0 )
310315 Row (i, fileNum, fileListEntry.getPath, start, end, true )
311316 }
0 commit comments