ParquetStore extends Store { public List<Path> getFilesForGroup(final String group) throws IOException { final Path dir = new Path(getDataDir() + "/" + getSnapshotPath(currentSnapshot) + "/" + GRAPH + "/" + GROUP + "=" + group); final FileStatus[] files = fs.listStatus(dir, path -> path.getName().endsWith(".parquet")); return Arrays .stream(files) .map(FileStatus::getPath) .collect(Collectors.toList()); }  @Override void initialise(final String graphId, final Schema schema, final StoreProperties properties); static String getSnapshotPath(final long snapshot); Path getGraphPartitionerPath(); FileSystem getFS(); SchemaUtils getSchemaUtils(); @Override Set<StoreTrait> getTraits(); String getDataDir(); String getTempFilesDir(); String getFile(final String group, final Partition partition); String getFile(final String group, final Integer partitionId); static String getFile(final Integer partitionId); String getFileForReversedEdges(final String group, final Partition partition); String getFileForReversedEdges(final String group, final Integer partitionId); List<Path> getFilesForGroup(final String group); Path getGroupPath(final String group); static String getGroupSubDir(final String group, final boolean reversed); Path getGroupPathForReversedEdges(final String group); String getGraphPath(); @SuppressFBWarnings(value = "BC_UNCONFIRMED_CAST_OF_RETURN_VALUE", justification = "The properties should always be ParquetStoreProperties") @Override ParquetStoreProperties getProperties(); @Override void validateSchemas(); long getLatestSnapshot(); void setLatestSnapshot(final long snapshot); GraphPartitioner getGraphPartitioner(); static final String GROUP; static final String GRAPH; static final String VERTEX; static final String SOURCE; static final String DESTINATION; static final String DIRECTED; static final String SNAPSHOT; static final String REVERSED_EDGES; static final String PARTITION; static final int LENGTH_OF_PARTITION_NUMBER_IN_FILENAME; @SuppressFBWarnings("MS_MUTABLE_ARRAY")
    static final Serialiser[] SERIALISERS; }
@Test public void shouldCorrectlyUseCompressionOption(@TempDir java.nio.file.Path tempDir) throws Exception { for (final String compressionType : Sets.newHashSet("GZIP", "SNAPPY", "UNCOMPRESSED")) { final Schema schema = new Schema.Builder() .type("int", new TypeDefinition.Builder() .clazz(Integer.class) .serialiser(new IntegerParquetSerialiser()) .build()) .type("string", new TypeDefinition.Builder() .clazz(String.class) .serialiser(new StringParquetSerialiser()) .build()) .type(DIRECTED_EITHER, Boolean.class) .entity("entity", new SchemaEntityDefinition.Builder() .vertex("string") .property("property1", "int") .aggregate(false) .build()) .edge("edge", new SchemaEdgeDefinition.Builder() .source("string") .destination("string") .property("property2", "int") .directed(DIRECTED_EITHER) .aggregate(false) .build()) .vertexSerialiser(new StringParquetSerialiser()) .build(); final ParquetStoreProperties parquetStoreProperties = TestUtils.getParquetStoreProperties(tempDir); parquetStoreProperties.setCompressionCodecName(compressionType); final ParquetStore parquetStore = (ParquetStore) ParquetStore.createStore("graphId", schema, parquetStoreProperties); final List<Element> elements = new ArrayList<>(); elements.add(new Entity.Builder() .group("entity") .vertex("A") .property("property1", 1) .build()); elements.add(new Edge.Builder() .group("edge") .source("B") .dest("C") .property("property2", 100) .build()); final AddElements add = new AddElements.Builder().input(elements).build(); parquetStore.execute(add, new Context()); final List<Path> files = parquetStore.getFilesForGroup("entity"); for (final Path path : files) { final ParquetMetadata parquetMetadata = ParquetFileReader .readFooter(new Configuration(), path, ParquetMetadataConverter.NO_FILTER); for (final BlockMetaData blockMetadata : parquetMetadata.getBlocks()) { blockMetadata.getColumns().forEach(c -> assertEquals(compressionType, c.getCodec().name())); } } } }