AddElementsHandler implements OperationHandler<AddElements> { @Override public Void doOperation(final AddElements operation, final Context context, final Store store) throws OperationException { addElements(operation, context, (ParquetStore) store); return null; }  @Override Void doOperation(final AddElements operation,
                            final Context context,
                            final Store store);  }
@Test public void testOnePartitionOneGroup(@TempDir java.nio.file.Path tempDir) throws OperationException, IOException, StoreException { final List<Element> elementsToAdd = new ArrayList<>(); elementsToAdd.addAll(AggregateAndSortDataTest.generateData()); elementsToAdd.addAll(AggregateAndSortDataTest.generateData()); final AddElements add = new AddElements.Builder() .input(elementsToAdd) .build(); final Context context = new Context(); final Schema schema = TestUtils.gafferSchema("schemaUsingLongVertexType"); final ParquetStoreProperties storeProperties = new ParquetStoreProperties(); final String testDir = tempDir.toString(); storeProperties.setDataDir(testDir + "/data"); storeProperties.setTempFilesDir(testDir + "/tmpdata"); final ParquetStore store = (ParquetStore) ParquetStore.createStore("graphId", schema, storeProperties); final FileSystem fs = FileSystem.get(new Configuration()); final SparkSession sparkSession = SparkSessionProvider.getSparkSession(); new AddElementsHandler().doOperation(add, context, store); final long snapshotId = store.getLatestSnapshot(); final Path snapshotPath = new Path(testDir + "/data", ParquetStore.getSnapshotPath(snapshotId)); assertTrue(fs.exists(snapshotPath)); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/." + ParquetStore.getFile(0) + ".crc"))); final Row[] results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(40, results.length); for (int i = 0; i < 40; i++) { assertEquals((long) i / 2, (long) results[i].getAs(ParquetStore.VERTEX)); assertEquals(i % 2 == 0 ? 'b' : 'a', ((byte[]) results[i].getAs("byte"))[0]); assertEquals(i % 2 == 0 ? 8f : 6f, results[i].getAs("float"), 0.01f); assertEquals(11L * 2 * (i / 2), (long) results[i].getAs("long")); assertEquals(i % 2 == 0 ? 14 : 12, (int) results[i].getAs("short")); assertEquals(i % 2 == 0 ? 100000L : 200000L, (long) results[i].getAs("date")); assertEquals(2, (int) results[i].getAs("count")); assertArrayEquals(i % 2 == 0 ? new String[]{"A", "C"} : new String[]{"A", "B"}, (String[]) ((WrappedArray<String>) results[i].getAs("treeSet")).array()); final FreqMap mergedFreqMap1 = new FreqMap(); mergedFreqMap1.put("A", 2L); mergedFreqMap1.put("B", 2L); final FreqMap mergedFreqMap2 = new FreqMap(); mergedFreqMap2.put("A", 2L); mergedFreqMap2.put("C", 2L); assertEquals(JavaConversions$.MODULE$.mapAsScalaMap(i % 2 == 0 ? mergedFreqMap2 : mergedFreqMap1), results[i].getAs("freqMap")); } }
@Test public void testOnePartitionAllGroups(@TempDir java.nio.file.Path tempDir) throws IOException, OperationException, StoreException { final List<Element> elementsToAdd = new ArrayList<>(); elementsToAdd.addAll(AggregateAndSortDataTest.generateData()); elementsToAdd.addAll(AggregateAndSortDataTest.generateData()); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false)); Collections.shuffle(elementsToAdd); final AddElements add = new AddElements.Builder() .input(elementsToAdd) .build(); final Context context = new Context(); final Schema schema = TestUtils.gafferSchema("schemaUsingLongVertexType"); final ParquetStoreProperties storeProperties = new ParquetStoreProperties(); final String testDir = tempDir.toString(); storeProperties.setDataDir(testDir + "/data"); storeProperties.setTempFilesDir(testDir + "/tmpdata"); final ParquetStore store = (ParquetStore) ParquetStore.createStore("graphId", schema, storeProperties); final FileSystem fs = FileSystem.get(new Configuration()); final SparkSession sparkSession = SparkSessionProvider.getSparkSession(); new AddElementsHandler().doOperation(add, context, store); final long snapshotId = store.getLatestSnapshot(); final Path snapshotPath = new Path(testDir + "/data", ParquetStore.getSnapshotPath(snapshotId)); assertTrue(fs.exists(snapshotPath)); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/." + ParquetStore.getFile(0) + ".crc"))); Row[] results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(40, results.length); for (int i = 0; i < 40; i++) { assertEquals((long) i / 2, (long) results[i].getAs(ParquetStore.VERTEX)); assertEquals(i % 2 == 0 ? 'b' : 'a', ((byte[]) results[i].getAs("byte"))[0]); assertEquals(i % 2 == 0 ? 8f : 6f, results[i].getAs("float"), 0.01f); assertEquals(11L * 2 * (i / 2), (long) results[i].getAs("long")); assertEquals(i % 2 == 0 ? 14 : 12, (int) results[i].getAs("short")); assertEquals(i % 2 == 0 ? 100000L : 200000L, (long) results[i].getAs("date")); assertEquals(2, (int) results[i].getAs("count")); assertArrayEquals(i % 2 == 0 ? new String[]{"A", "C"} : new String[]{"A", "B"}, (String[]) ((WrappedArray<String>) results[i].getAs("treeSet")).array()); final FreqMap mergedFreqMap1 = new FreqMap(); mergedFreqMap1.put("A", 2L); mergedFreqMap1.put("B", 2L); final FreqMap mergedFreqMap2 = new FreqMap(); mergedFreqMap2.put("A", 2L); mergedFreqMap2.put("C", 2L); assertEquals(JavaConversions$.MODULE$.mapAsScalaMap(i % 2 == 0 ? mergedFreqMap2 : mergedFreqMap1), results[i].getAs("freqMap")); } assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, "graph/group=BasicEntity2/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(4, results.length); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L), results[0]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L), results[1]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L), results[2]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L), results[3]); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/." + ParquetStore.getFile(0) + ".crc"))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, true) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, true) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(6, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L)), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L)), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L)), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L)), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L)), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L)), results[5]); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, true) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(6, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L)), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L)), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L)), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L)), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L)), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L)), results[5]); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(4, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[3]); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, true) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(4, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[3]); }
@Test public void testRepeatedCallsOfAddElementsHandler(@TempDir java.nio.file.Path tempDir) throws IOException, OperationException, StoreException { final List<Element> elementsToAdd = new ArrayList<>(); elementsToAdd.addAll(AggregateAndSortDataTest.generateData()); elementsToAdd.addAll(AggregateAndSortDataTest.generateData()); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L))); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true)); elementsToAdd.add(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false)); Collections.shuffle(elementsToAdd); final AddElements add = new AddElements.Builder() .input(elementsToAdd) .build(); final Context context = new Context(); final Schema schema = TestUtils.gafferSchema("schemaUsingLongVertexType"); final ParquetStoreProperties storeProperties = new ParquetStoreProperties(); final String testDir = tempDir.toString(); storeProperties.setDataDir(testDir + "/data"); storeProperties.setTempFilesDir(testDir + "/tmpdata"); final ParquetStore store = (ParquetStore) ParquetStore.createStore("graphId", schema, storeProperties); final FileSystem fs = FileSystem.get(new Configuration()); final SparkSession sparkSession = SparkSessionProvider.getSparkSession(); new AddElementsHandler().doOperation(add, context, store); new AddElementsHandler().doOperation(add, context, store); long snapshotId = store.getLatestSnapshot(); Path snapshotPath = new Path(testDir + "/data", ParquetStore.getSnapshotPath(snapshotId)); assertTrue(fs.exists(snapshotPath)); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/." + ParquetStore.getFile(0) + ".crc"))); Row[] results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(40, results.length); for (int i = 0; i < 40; i++) { assertEquals((long) i / 2, (long) results[i].getAs(ParquetStore.VERTEX)); assertEquals(i % 2 == 0 ? 'b' : 'a', ((byte[]) results[i].getAs("byte"))[0]); assertEquals(i % 2 == 0 ? 16f : 12f, results[i].getAs("float"), 0.01f); assertEquals(11L * 2 * 2 * (i / 2), (long) results[i].getAs("long")); assertEquals(i % 2 == 0 ? 28 : 24, (int) results[i].getAs("short")); assertEquals(i % 2 == 0 ? 100000L : 200000L, (long) results[i].getAs("date")); assertEquals(4, (int) results[i].getAs("count")); assertArrayEquals(i % 2 == 0 ? new String[]{"A", "C"} : new String[]{"A", "B"}, (String[]) ((WrappedArray<String>) results[i].getAs("treeSet")).array()); final FreqMap mergedFreqMap1 = new FreqMap(); mergedFreqMap1.put("A", 4L); mergedFreqMap1.put("B", 4L); final FreqMap mergedFreqMap2 = new FreqMap(); mergedFreqMap2.put("A", 4L); mergedFreqMap2.put("C", 4L); assertEquals(JavaConversions$.MODULE$.mapAsScalaMap(i % 2 == 0 ? mergedFreqMap2 : mergedFreqMap1), results[i].getAs("freqMap")); } assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(8, results.length); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L), results[0]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L), results[1]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L), results[2]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L), results[3]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L), results[4]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L), results[5]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L), results[6]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L), results[7]); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(6, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L), (short) 2), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L), (short) 2), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L), (short) 2), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L), (short) 2), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L), (short) 2), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L), (short) 2), results[5]); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, true) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(6, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L), (short) 2), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L), (short) 2), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L), (short) 2), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L), (short) 2), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L), (short) 2), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L), (short) 2), results[5]); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(8, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[5]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[6]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[7]); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, true) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(8, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[5]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[6]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[7]); elementsToAdd.clear(); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L)); elementsToAdd.add(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L)); new AddElementsHandler().doOperation(add, context, store); snapshotId = store.getLatestSnapshot(); snapshotPath = new Path(testDir + "/data", ParquetStore.getSnapshotPath(snapshotId)); assertTrue(fs.exists(snapshotPath)); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(40, results.length); for (int i = 0; i < 40; i++) { assertEquals((long) i / 2, (long) results[i].getAs(ParquetStore.VERTEX)); assertEquals(i % 2 == 0 ? 'b' : 'a', ((byte[]) results[i].getAs("byte"))[0]); assertEquals(i % 2 == 0 ? 16f : 12f, results[i].getAs("float"), 0.01f); assertEquals(11L * 2 * 2 * (i / 2), (long) results[i].getAs("long")); assertEquals(i % 2 == 0 ? 28 : 24, (int) results[i].getAs("short")); assertEquals(i % 2 == 0 ? 100000L : 200000L, (long) results[i].getAs("date")); assertEquals(4, (int) results[i].getAs("count")); assertArrayEquals(i % 2 == 0 ? new String[]{"A", "C"} : new String[]{"A", "B"}, (String[]) ((WrappedArray<String>) results[i].getAs("treeSet")).array()); final FreqMap mergedFreqMap1 = new FreqMap(); mergedFreqMap1.put("A", 4L); mergedFreqMap1.put("B", 4L); final FreqMap mergedFreqMap2 = new FreqMap(); mergedFreqMap2.put("A", 4L); mergedFreqMap2.put("C", 4L); assertEquals(JavaConversions$.MODULE$.mapAsScalaMap(i % 2 == 0 ? mergedFreqMap2 : mergedFreqMap1), results[i].getAs("freqMap")); } assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.ENTITY_2, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(12, results.length); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L), results[0]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L), results[1]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(1L), results[2]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L), results[3]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L), results[4]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10L), results[5]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L), results[6]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L), results[7]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(100L), results[8]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L), results[9]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L), results[10]); checkEntityGroup2(WriteUnsortedDataTest.createEntityForEntityGroup_2(10000L), results[11]); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(6, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L), (short) 2), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L), (short) 2), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L), (short) 2), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L), (short) 2), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L), (short) 2), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L), (short) 2), results[5]); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE, true) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(6, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 2L, false, new Date(400L), (short) 2), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, false, new Date(400L), (short) 2), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(300L), (short) 2), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(1L, 10L, true, new Date(400L), (short) 2), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(10000L, 1000L, true, new Date(100L), (short) 2), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup(100L, 100000L, false, new Date(200L), (short) 2), results[5]); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(8, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[5]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[6]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[7]); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, true) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(8, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[5]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[6]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[7]); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)))); assertTrue(fs.exists(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/." + ParquetStore.getFile(0) + ".crc"))); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, false) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(8, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[5]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[6]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[7]); results = (Row[]) sparkSession .read() .parquet(new Path(snapshotPath, ParquetStore.getGroupSubDir(TestGroups.EDGE_2, true) + "/" + ParquetStore.getFile(0)).toString()) .collect(); assertEquals(8, results.length); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[0]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10000L, 20L, true), results[1]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[2]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(10L, 50L, true), results[3]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[4]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(100L, 200L, false), results[5]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[6]); checkEdge(WriteUnsortedDataTest.createEdgeForEdgeGroup_2(1L, 2000L, false), results[7]); }