QueryGenerator { public ParquetQuery getParquetQuery(final Operation operation) throws IOException, OperationException { if (operation instanceof GetAllElements) { return getPathsAndFiltersForAllElements((GetAllElements) operation); } else if (operation instanceof GetElements) { return getPathsAndFiltersForGetElements((GetElements) operation); } else { throw new OperationException("QueryGenerator can only handle GetAllElements and GetElements operations"); } } QueryGenerator(final ParquetStore store); ParquetQuery getParquetQuery(final Operation operation);  }
@Test public void testQueryGeneratorForGetAllElements(@TempDir java.nio.file.Path tempDir) throws IOException, OperationException { final String folder = String.format("file: final String snapshotFolder = folder + "/" + ParquetStore.getSnapshotPath(1000L); CalculatePartitionerTest.writeData(snapshotFolder, new SchemaUtils(schema)); final ParquetStoreProperties storeProperties = new ParquetStoreProperties(); storeProperties.setDataDir(folder); storeProperties.setTempFilesDir(folder + "/tmpdata"); final ParquetStore store = (ParquetStore) ParquetStore.createStore("graphId", schema, storeProperties); GetAllElements getAllElements = new GetAllElements.Builder().build(); ParquetQuery query = new QueryGenerator(store).getParquetQuery(getAllElements); final List<ParquetFileQuery> expected = new ArrayList<>(); for (final String group : Arrays.asList(TestGroups.ENTITY, TestGroups.ENTITY_2, TestGroups.EDGE, TestGroups.EDGE_2)) { final Path groupFolderPath = store.getGroupPath(group); for (int partition = 0; partition < 10; partition++) { final Path pathForPartitionFile = new Path(groupFolderPath, ParquetStore.getFile(partition)); expected.add(new ParquetFileQuery(pathForPartitionFile, null, true)); } } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getAllElements = new GetAllElements.Builder().view(new View.Builder().edge(TestGroups.EDGE).build()).build(); query = new QueryGenerator(store).getParquetQuery(getAllElements); expected.clear(); Path groupFolderPath = store.getGroupPath(TestGroups.EDGE); for (int partition = 0; partition < 10; partition++) { final Path pathForPartitionFile = new Path(groupFolderPath, ParquetStore.getFile(partition)); expected.add(new ParquetFileQuery(pathForPartitionFile, null, true)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getAllElements = new GetAllElements.Builder() .view(new View.Builder() .edge(TestGroups.EDGE, new ViewElementDefinition.Builder() .preAggregationFilter(new ElementFilter.Builder() .select("count") .execute(new IsMoreThan(10)) .build()) .build()) .build()) .build(); query = new QueryGenerator(store).getParquetQuery(getAllElements); expected.clear(); for (int partition = 0; partition < 10; partition++) { final Path pathForPartitionFile = new Path(groupFolderPath, ParquetStore.getFile(partition)); expected.add(new ParquetFileQuery(pathForPartitionFile, gt(FilterApi.intColumn("count"), 10), true)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getAllElements = new GetAllElements.Builder() .view(new View.Builder() .edge(TestGroups.EDGE, new ViewElementDefinition.Builder() .preAggregationFilter(new ElementFilter.Builder() .select("count") .execute(new IsEvenFilter()) .build()) .build()) .build()) .build(); query = new QueryGenerator(store).getParquetQuery(getAllElements); expected.clear(); for (int partition = 0; partition < 10; partition++) { final Path pathForPartitionFile = new Path(groupFolderPath, ParquetStore.getFile(partition)); expected.add(new ParquetFileQuery(pathForPartitionFile, null, false)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getAllElements = new GetAllElements.Builder() .view(new View.Builder() .edge(TestGroups.EDGE, new ViewElementDefinition.Builder() .preAggregationFilter(new ElementFilter.Builder() .select("count") .execute(new IsEvenFilter()) .select("count") .execute(new IsMoreThan(10)) .build()) .build()) .build()) .build(); query = new QueryGenerator(store).getParquetQuery(getAllElements); expected.clear(); for (int partition = 0; partition < 10; partition++) { final Path pathForPartitionFile = new Path(groupFolderPath, ParquetStore.getFile(partition)); expected.add(new ParquetFileQuery(pathForPartitionFile, gt(FilterApi.intColumn("count"), 10), false)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); }
@Test public void testQueryGeneratorForGetElementsWithEntitySeeds(@TempDir java.nio.file.Path tempDir) throws IOException, OperationException { final String folder = String.format("file: final String snapshotFolder = folder + "/" + ParquetStore.getSnapshotPath(1000L); CalculatePartitionerTest.writeData(snapshotFolder, new SchemaUtils(schema)); final ParquetStoreProperties storeProperties = new ParquetStoreProperties(); storeProperties.setDataDir(folder); storeProperties.setTempFilesDir(folder + "/tmpdata"); final ParquetStore store = (ParquetStore) ParquetStore.createStore("graphId", schema, storeProperties); GetElements getElements = new GetElements.Builder() .input(new EntitySeed(0L)) .seedMatching(SeedMatching.SeedMatchingType.RELATED) .build(); ParquetQuery query = new QueryGenerator(store).getParquetQuery(getElements); final List<ParquetFileQuery> expected = new ArrayList<>(); final FilterPredicate vertex0 = eq(FilterApi.longColumn(ParquetStore.VERTEX), 0L); final FilterPredicate source0 = eq(FilterApi.longColumn(ParquetStore.SOURCE), 0L); final FilterPredicate destination0 = eq(FilterApi.longColumn(ParquetStore.DESTINATION), 0L); for (final String group : Arrays.asList(TestGroups.ENTITY, TestGroups.ENTITY_2)) { final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, false)); final Path pathForPartitionFile = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartitionFile, vertex0, true)); } for (final String group : Arrays.asList(TestGroups.EDGE, TestGroups.EDGE_2)) { final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, false)); final Path pathForPartitionFile = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartitionFile, source0, true)); final Path reversedGroupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, true)); final Path pathForReversedPartitionFile = new Path(reversedGroupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForReversedPartitionFile, destination0, true)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getElements = new GetElements.Builder() .input(new EntitySeed(0L), new EntitySeed(1000000L)) .seedMatching(SeedMatching.SeedMatchingType.RELATED) .build(); query = new QueryGenerator(store).getParquetQuery(getElements); expected.clear(); final FilterPredicate vertex1000000 = eq(FilterApi.longColumn(ParquetStore.VERTEX), 1000000L); final FilterPredicate source1000000 = eq(FilterApi.longColumn(ParquetStore.SOURCE), 1000000L); final FilterPredicate destination1000000 = eq(FilterApi.longColumn(ParquetStore.DESTINATION), 1000000L); for (final String group : Arrays.asList(TestGroups.ENTITY, TestGroups.ENTITY_2)) { final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, false)); final Path pathForPartitionFile1 = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartitionFile1, vertex0, true)); final Path pathForPartitionFile2 = new Path(groupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartitionFile2, vertex1000000, true)); } for (final String group : Arrays.asList(TestGroups.EDGE, TestGroups.EDGE_2)) { final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, false)); final Path reversedGroupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, true)); final Path pathForPartitionFile1 = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartitionFile1, source0, true)); final Path pathForPartitionFile2 = new Path(groupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartitionFile2, source1000000, true)); final Path pathForPartitionFile3 = new Path(reversedGroupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartitionFile3, destination0, true)); final Path pathForPartitionFile4 = new Path(reversedGroupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartitionFile4, destination1000000, true)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getElements = new GetElements.Builder() .input(new EntitySeed(0L), new EntitySeed(1000000L)) .seedMatching(SeedMatching.SeedMatchingType.RELATED) .view(new View.Builder() .edge(TestGroups.EDGE, new ViewElementDefinition.Builder() .preAggregationFilter(new ElementFilter.Builder() .select("count") .execute(new IsMoreThan(10)) .build()) .build()) .build()) .build(); query = new QueryGenerator(store).getParquetQuery(getElements); expected.clear(); final FilterPredicate source0AndCount = and(gt(FilterApi.intColumn("count"), 10), eq(FilterApi.longColumn(ParquetStore.SOURCE), 0L)); final FilterPredicate source1000000AndCount = and(gt(FilterApi.intColumn("count"), 10), eq(FilterApi.longColumn(ParquetStore.SOURCE), 1000000L)); final FilterPredicate destination0AndCount = and(gt(FilterApi.intColumn("count"), 10), eq(FilterApi.longColumn(ParquetStore.DESTINATION), 0L)); final FilterPredicate destination1000000AndCount = and(gt(FilterApi.intColumn("count"), 10), eq(FilterApi.longColumn(ParquetStore.DESTINATION), 1000000L)); final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(TestGroups.EDGE, false)); final Path reversedGroupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(TestGroups.EDGE, true)); final Path pathForPartitionFile1 = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartitionFile1, source0AndCount, true)); final Path pathForPartitionFile2 = new Path(groupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartitionFile2, source1000000AndCount, true)); final Path pathForPartitionFile3 = new Path(reversedGroupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartitionFile3, destination0AndCount, true)); final Path pathForPartitionFile4 = new Path(reversedGroupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartitionFile4, destination1000000AndCount, true)); assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getElements = new GetElements.Builder() .input(new EntitySeed(0L), new EntitySeed(1000000L)) .seedMatching(SeedMatching.SeedMatchingType.RELATED) .view(new View.Builder() .edge(TestGroups.EDGE, new ViewElementDefinition.Builder() .preAggregationFilter(new ElementFilter.Builder() .select("count") .execute(new IsEvenFilter()) .build()) .build()) .build()) .build(); query = new QueryGenerator(store).getParquetQuery(getElements); expected.clear(); expected.add(new ParquetFileQuery(pathForPartitionFile1, source0, false)); expected.add(new ParquetFileQuery(pathForPartitionFile2, source1000000, false)); expected.add(new ParquetFileQuery(pathForPartitionFile3, destination0, false)); expected.add(new ParquetFileQuery(pathForPartitionFile4, destination1000000, false)); assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); }
@Test public void testQueryGeneratorForGetElementsWithEdgeSeeds(@TempDir java.nio.file.Path tempDir) throws IOException, OperationException { final String folder = String.format("file: final String snapshotFolder = folder + "/" + ParquetStore.getSnapshotPath(1000L); CalculatePartitionerTest.writeData(snapshotFolder, new SchemaUtils(schema)); final ParquetStoreProperties storeProperties = new ParquetStoreProperties(); storeProperties.setDataDir(folder); storeProperties.setTempFilesDir(folder + "/tmpdata"); final ParquetStore store = (ParquetStore) ParquetStore.createStore("graphId", schema, storeProperties); GetElements getElements = new GetElements.Builder() .input(new EdgeSeed(0L, 1L, DirectedType.DIRECTED), new EdgeSeed(10L, 1000L, DirectedType.UNDIRECTED), new EdgeSeed(10000L, 10L, DirectedType.EITHER)) .seedMatching(SeedMatching.SeedMatchingType.EQUAL) .build(); ParquetQuery query = new QueryGenerator(store).getParquetQuery(getElements); final List<ParquetFileQuery> expected = new ArrayList<>(); final FilterPredicate source0 = eq(FilterApi.longColumn(ParquetStore.SOURCE), 0L); final FilterPredicate source10 = eq(FilterApi.longColumn(ParquetStore.SOURCE), 10L); final FilterPredicate source10000 = eq(FilterApi.longColumn(ParquetStore.SOURCE), 10000L); final FilterPredicate destination1 = eq(FilterApi.longColumn(ParquetStore.DESTINATION), 1L); final FilterPredicate destination10 = eq(FilterApi.longColumn(ParquetStore.DESTINATION), 10L); final FilterPredicate destination1000 = eq(FilterApi.longColumn(ParquetStore.DESTINATION), 1000L); final FilterPredicate directedTrue = eq(FilterApi.booleanColumn(ParquetStore.DIRECTED), true); final FilterPredicate directedFalse = eq(FilterApi.booleanColumn(ParquetStore.DIRECTED), false); final FilterPredicate source0Destination1DirectedTrue = and(and(source0, destination1), directedTrue); final FilterPredicate source10Destination1000DirectedFalse = and(and(source10, destination1000), directedFalse); final FilterPredicate source10000Destination10DirectedEither = and(source10000, destination10); for (final String group : Arrays.asList(TestGroups.EDGE, TestGroups.EDGE_2)) { final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, false)); final Path pathForPartition0File = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartition0File, source0Destination1DirectedTrue, true)); final Path pathForPartition1File = new Path(groupFolderPath, ParquetStore.getFile(1)); expected.add(new ParquetFileQuery(pathForPartition1File, source10Destination1000DirectedFalse, true)); final Path pathForPartition9File = new Path(groupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartition9File, source10000Destination10DirectedEither, true)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); getElements = new GetElements.Builder() .input(new EdgeSeed(0L, 1L, DirectedType.DIRECTED), new EdgeSeed(10L, 1000L, DirectedType.UNDIRECTED), new EdgeSeed(10000L, 10L, DirectedType.EITHER)) .seedMatching(SeedMatching.SeedMatchingType.RELATED) .build(); query = new QueryGenerator(store).getParquetQuery(getElements); expected.clear(); final FilterPredicate vertex0 = eq(FilterApi.longColumn(ParquetStore.VERTEX), 0L); final FilterPredicate vertex1 = eq(FilterApi.longColumn(ParquetStore.VERTEX), 1L); final FilterPredicate vertex10 = eq(FilterApi.longColumn(ParquetStore.VERTEX), 10L); final FilterPredicate vertex1000 = eq(FilterApi.longColumn(ParquetStore.VERTEX), 1000L); final FilterPredicate vertex10000 = eq(FilterApi.longColumn(ParquetStore.VERTEX), 10000L); final FilterPredicate vertex0or1 = or(vertex0, vertex1); final FilterPredicate vertex10or1000 = or(vertex10, vertex1000); final FilterPredicate vertex10000or10 = or(vertex10000, vertex10); for (final String group : Arrays.asList(TestGroups.ENTITY, TestGroups.ENTITY_2)) { final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, false)); final Path pathForPartition0File = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartition0File, vertex0or1, true)); final Path pathForPartition1File = new Path(groupFolderPath, ParquetStore.getFile(1)); expected.add(new ParquetFileQuery(pathForPartition1File, or(vertex10or1000, vertex10000or10), true)); final Path pathForPartition9File = new Path(groupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartition9File, or(vertex10or1000, vertex10000or10), true)); } for (final String group : Arrays.asList(TestGroups.EDGE, TestGroups.EDGE_2)) { final Path groupFolderPath = new Path(snapshotFolder, ParquetStore.getGroupSubDir(group, false)); final Path pathForPartition0File = new Path(groupFolderPath, ParquetStore.getFile(0)); expected.add(new ParquetFileQuery(pathForPartition0File, source0Destination1DirectedTrue, true)); final Path pathForPartition1File = new Path(groupFolderPath, ParquetStore.getFile(1)); expected.add(new ParquetFileQuery(pathForPartition1File, source10Destination1000DirectedFalse, true)); final Path pathForPartition9File = new Path(groupFolderPath, ParquetStore.getFile(9)); expected.add(new ParquetFileQuery(pathForPartition9File, source10000Destination10DirectedEither, true)); } assertThat(expected, containsInAnyOrder(query.getAllParquetFileQueries().toArray())); }