CompressedSource extends FileBasedSource<T> { public static <T> CompressedSource<T> from(FileBasedSource<T> sourceDelegate) { return new CompressedSource<>(sourceDelegate, CompressionMode.AUTO); } private  CompressedSource(
      FileBasedSource<T> sourceDelegate, DecompressingChannelFactory channelFactory); private  CompressedSource(
      FileBasedSource<T> sourceDelegate,
      DecompressingChannelFactory channelFactory,
      Metadata metadata,
      long minBundleSize,
      long startOffset,
      long endOffset); static CompressedSource<T> from(FileBasedSource<T> sourceDelegate); CompressedSource<T> withDecompression(DecompressingChannelFactory channelFactory); CompressedSource<T> withCompression(Compression compression); @Override void validate(); @Override void populateDisplayData(DisplayData.Builder builder); @Override final Coder<T> getOutputCoder(); final DecompressingChannelFactory getChannelFactory();  }
@Test public void testHeterogeneousCompression() throws Exception { String baseName = "test-input"; byte[] generated = generateInput(1000); List<Byte> expected = new ArrayList<>(); File uncompressedFile = tmpFolder.newFile(baseName + ".bin"); generated = generateInput(1000, 1); Files.write(generated, uncompressedFile); expected.addAll(Bytes.asList(generated)); File gzipFile = tmpFolder.newFile(baseName + ".gz"); generated = generateInput(1000, 2); writeFile(gzipFile, generated, CompressionMode.GZIP); expected.addAll(Bytes.asList(generated)); File bzip2File = tmpFolder.newFile(baseName + ".bz2"); generated = generateInput(1000, 3); writeFile(bzip2File, generated, CompressionMode.BZIP2); expected.addAll(Bytes.asList(generated)); File zstdFile = tmpFolder.newFile(baseName + ".zst"); generated = generateInput(1000, 4); writeFile(zstdFile, generated, CompressionMode.ZSTD); expected.addAll(Bytes.asList(generated)); File lzoFile = tmpFolder.newFile(baseName + ".lzo_deflate"); generated = generateInput(1000, 4); writeFile(lzoFile, generated, CompressionMode.LZO); expected.addAll(Bytes.asList(generated)); String filePattern = new File(tmpFolder.getRoot().toString(), baseName + ".*").toString(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filePattern, 1)); List<Byte> actual = SourceTestUtils.readFromSource(source, PipelineOptionsFactory.create()); assertEquals(HashMultiset.create(actual), HashMultiset.create(expected)); }
@Test public void testEmptyGzipProgress() throws IOException { File tmpFile = tmpFolder.newFile("empty.gz"); String filename = tmpFile.toPath().toString(); writeFile(tmpFile, new byte[0], CompressionMode.GZIP); PipelineOptions options = PipelineOptionsFactory.create(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filename, 1)); try (BoundedReader<Byte> readerOrig = source.createReader(options)) { assertThat(readerOrig, instanceOf(CompressedReader.class)); CompressedReader<Byte> reader = (CompressedReader<Byte>) readerOrig; assertEquals(0.0, reader.getFractionConsumed(), delta); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); assertFalse(reader.start()); assertEquals(1.0, reader.getFractionConsumed(), delta); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(0, reader.getSplitPointsRemaining()); } }
@Test public void testGzipProgress() throws IOException { int numRecords = 3; File tmpFile = tmpFolder.newFile("nonempty.gz"); String filename = tmpFile.toPath().toString(); writeFile(tmpFile, new byte[numRecords], CompressionMode.GZIP); PipelineOptions options = PipelineOptionsFactory.create(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filename, 1)); try (BoundedReader<Byte> readerOrig = source.createReader(options)) { assertThat(readerOrig, instanceOf(CompressedReader.class)); CompressedReader<Byte> reader = (CompressedReader<Byte>) readerOrig; assertEquals(0.0, reader.getFractionConsumed(), delta); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); for (int i = 0; i < numRecords; ++i) { if (i == 0) { assertTrue(reader.start()); } else { assertTrue(reader.advance()); } assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); } assertFalse(reader.advance()); assertEquals(1.0, reader.getFractionConsumed(), delta); assertEquals(1, reader.getSplitPointsConsumed()); assertEquals(0, reader.getSplitPointsRemaining()); } }
@Test public void testUnsplittable() throws IOException { String baseName = "test-input"; File compressedFile = tmpFolder.newFile(baseName + ".gz"); byte[] input = generateInput(10000); writeFile(compressedFile, input, CompressionMode.GZIP); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(compressedFile.getPath(), 1)); List<Byte> expected = Lists.newArrayList(); for (byte i : input) { expected.add(i); } PipelineOptions options = PipelineOptionsFactory.create(); BoundedReader<Byte> reader = source.createReader(options); List<Byte> actual = Lists.newArrayList(); for (boolean hasNext = reader.start(); hasNext; hasNext = reader.advance()) { actual.add(reader.getCurrent()); if (actual.size() % 9 == 0) { Double fractionConsumed = reader.getFractionConsumed(); assertNotNull(fractionConsumed); assertNull(reader.splitAtFraction(fractionConsumed)); } } assertEquals(expected.size(), actual.size()); assertEquals(Sets.newHashSet(expected), Sets.newHashSet(actual)); }
@Test public void testSplittableProgress() throws IOException { File tmpFile = tmpFolder.newFile("nonempty.txt"); String filename = tmpFile.toPath().toString(); Files.write(new byte[2], tmpFile); PipelineOptions options = PipelineOptionsFactory.create(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filename, 1)); try (BoundedReader<Byte> readerOrig = source.createReader(options)) { assertThat(readerOrig, not(instanceOf(CompressedReader.class))); assertThat(readerOrig, instanceOf(FileBasedReader.class)); FileBasedReader<Byte> reader = (FileBasedReader<Byte>) readerOrig; assertEquals(0.0, reader.getFractionConsumed(), delta); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(BoundedReader.SPLIT_POINTS_UNKNOWN, reader.getSplitPointsRemaining()); assertTrue(reader.start()); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(BoundedReader.SPLIT_POINTS_UNKNOWN, reader.getSplitPointsRemaining()); assertTrue(reader.advance()); assertEquals(1, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); assertFalse(reader.advance()); assertEquals(1.0, reader.getFractionConsumed(), delta); assertEquals(2, reader.getSplitPointsConsumed()); assertEquals(0, reader.getSplitPointsRemaining()); } }
@Test public void testHeterogeneousCompression() throws Exception { String baseName = "test-input"; byte[] generated = generateInput(1000); List<Byte> expected = new ArrayList<>(); File uncompressedFile = tmpFolder.newFile(baseName + ".bin"); generated = generateInput(1000, 1); Files.write(generated, uncompressedFile); expected.addAll(Bytes.asList(generated)); File gzipFile = tmpFolder.newFile(baseName + ".gz"); generated = generateInput(1000, 2); writeFile(gzipFile, generated, CompressionMode.GZIP); expected.addAll(Bytes.asList(generated)); File bzip2File = tmpFolder.newFile(baseName + ".bz2"); generated = generateInput(1000, 3); writeFile(bzip2File, generated, CompressionMode.BZIP2); expected.addAll(Bytes.asList(generated)); File zstdFile = tmpFolder.newFile(baseName + ".zst"); generated = generateInput(1000, 4); writeFile(zstdFile, generated, CompressionMode.ZSTD); expected.addAll(Bytes.asList(generated)); String filePattern = new File(tmpFolder.getRoot().toString(), baseName + ".*").toString(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filePattern, 1)); List<Byte> actual = SourceTestUtils.readFromSource(source, PipelineOptionsFactory.create()); assertEquals(HashMultiset.create(actual), HashMultiset.create(expected)); }
@Test public void testEmptyGzipProgress() throws IOException { File tmpFile = tmpFolder.newFile("empty.gz"); String filename = tmpFile.toPath().toString(); writeFile(tmpFile, new byte[0], CompressionMode.GZIP); PipelineOptions options = PipelineOptionsFactory.create(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filename, 1)); try (BoundedReader<Byte> readerOrig = source.createReader(options)) { assertThat(readerOrig, instanceOf(CompressedReader.class)); CompressedReader<Byte> reader = (CompressedReader<Byte>) readerOrig; assertEquals(0.0, reader.getFractionConsumed(), 1e-6); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); assertFalse(reader.start()); assertEquals(1.0, reader.getFractionConsumed(), 1e-6); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(0, reader.getSplitPointsRemaining()); } }
@Test public void testGzipProgress() throws IOException { int numRecords = 3; File tmpFile = tmpFolder.newFile("nonempty.gz"); String filename = tmpFile.toPath().toString(); writeFile(tmpFile, new byte[numRecords], CompressionMode.GZIP); PipelineOptions options = PipelineOptionsFactory.create(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filename, 1)); try (BoundedReader<Byte> readerOrig = source.createReader(options)) { assertThat(readerOrig, instanceOf(CompressedReader.class)); CompressedReader<Byte> reader = (CompressedReader<Byte>) readerOrig; assertEquals(0.0, reader.getFractionConsumed(), 1e-6); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); for (int i = 0; i < numRecords; ++i) { if (i == 0) { assertTrue(reader.start()); } else { assertTrue(reader.advance()); } assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); } assertFalse(reader.advance()); assertEquals(1.0, reader.getFractionConsumed(), 1e-6); assertEquals(1, reader.getSplitPointsConsumed()); assertEquals(0, reader.getSplitPointsRemaining()); } }
@Test public void testSplittableProgress() throws IOException { File tmpFile = tmpFolder.newFile("nonempty.txt"); String filename = tmpFile.toPath().toString(); Files.write(new byte[2], tmpFile); PipelineOptions options = PipelineOptionsFactory.create(); CompressedSource<Byte> source = CompressedSource.from(new ByteSource(filename, 1)); try (BoundedReader<Byte> readerOrig = source.createReader(options)) { assertThat(readerOrig, not(instanceOf(CompressedReader.class))); assertThat(readerOrig, instanceOf(FileBasedReader.class)); FileBasedReader<Byte> reader = (FileBasedReader<Byte>) readerOrig; assertEquals(0.0, reader.getFractionConsumed(), 1e-6); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(BoundedReader.SPLIT_POINTS_UNKNOWN, reader.getSplitPointsRemaining()); assertTrue(reader.start()); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(BoundedReader.SPLIT_POINTS_UNKNOWN, reader.getSplitPointsRemaining()); assertTrue(reader.advance()); assertEquals(1, reader.getSplitPointsConsumed()); assertEquals(1, reader.getSplitPointsRemaining()); assertFalse(reader.advance()); assertEquals(1.0, reader.getFractionConsumed(), 1e-6); assertEquals(2, reader.getSplitPointsConsumed()); assertEquals(0, reader.getSplitPointsRemaining()); } }