Elasticsearch5SearchIndex implements SearchIndex, SearchIndexWithVertexPropertyCountByValue { @Override public void flush(Graph graph) { bulkUpdateService.flush(); if (shouldRefreshIndexOnFlush()) { indexRefreshTracker.refresh(client); } } Elasticsearch5SearchIndex(Graph graph, GraphConfiguration config); PropertyNameVisibilitiesStore getPropertyNameVisibilitiesStore(); Set<String> getIndexNamesFromElasticsearch(); @Override void clearCache(); @Override void addElement(
        Graph graph,
        Element element,
        Set<String> additionalVisibilities,
        Set<String> additionalVisibilitiesToDelete,
        Authorizations authorizations
    ); @Override void updateElement(
        Graph graph,
        ExistingElementMutation<TElement> elementMutation,
        Authorizations authorizations
    ); @Override void addElementExtendedData(
        Graph graph,
        ElementLocation elementLocation,
        Iterable<ExtendedDataMutation> extendedData,
        Iterable<AdditionalExtendedDataVisibilityAddMutation> additionalExtendedDataVisibilities,
        Iterable<AdditionalExtendedDataVisibilityDeleteMutation> additionalExtendedDataVisibilityDeletes,
        Authorizations authorizations
    ); @Override void deleteExtendedData(Graph graph, ExtendedDataRowId rowId, Authorizations authorizations); @Override void deleteExtendedData(
        Graph graph,
        ElementLocation elementLocation,
        String tableName,
        String row,
        String columnName,
        String key,
        Visibility visibility,
        Authorizations authorizations
    ); void alterExtendedDataElementTypeVisibility(
        Graph graph,
        ElementMutation<T> elementMutation,
        Iterable<ExtendedDataRow> rows,
        Visibility oldVisibility,
        Visibility newVisibility
    ); @Override void addExtendedData(
        Graph graph,
        ElementLocation sourceElementLocation,
        Iterable<ExtendedDataRow> extendedDatas,
        Authorizations authorizations
    ); @Override void alterElementVisibility(
        Graph graph,
        ExistingElementMutation<T> elementMutation,
        Visibility oldVisibility,
        Visibility newVisibility,
        Authorizations authorizations
    ); @Override void markElementHidden(Graph graph, Element element, Visibility visibility, Authorizations authorizations); @Override void markElementVisible(
        Graph graph,
        ElementLocation elementLocation,
        Visibility visibility,
        Authorizations authorizations
    ); @Override void markPropertyHidden(
        Graph graph,
        ElementLocation elementLocation,
        Property property,
        Visibility visibility,
        Authorizations authorizations
    ); @Override void markPropertyVisible(
        Graph graph,
        ElementLocation elementLocation,
        Property property,
        Visibility visibility,
        Authorizations authorizations
    ); boolean isServerPluginInstalled(); void handleBulkFailure(BulkItem<?> bulkItem, BulkItemResponse bulkItemResponse, AtomicBoolean retry); boolean supportsExactMatchSearch(PropertyDefinition propertyDefinition); VertexiumMetricRegistry getMetricsRegistry(); String getPropertyVisibilityHashFromPropertyName(String propertyName); String getAggregationName(String name); String replaceFieldnameDots(String fieldName); String[] getAllMatchingPropertyNames(Graph graph, String propertyName, Authorizations authorizations); String[] addHashesToPropertyName(String propertyName, Collection<String> hashes); Collection<String> getQueryableExtendedDataVisibilities(Graph graph, Authorizations authorizations); Collection<String> getQueryableElementTypeVisibilityPropertyNames(Graph graph, Authorizations authorizations); Collection<String> getQueryablePropertyNames(Graph graph, Authorizations authorizations); @Override void deleteElement(Graph graph, ElementId elementId, Authorizations authorizations); @Override SearchIndexSecurityGranularity getSearchIndexSecurityGranularity(); @Override GraphQuery queryGraph(Graph graph, String queryString, Authorizations authorizations); @Override VertexQuery queryVertex(Graph graph, Vertex vertex, String queryString, Authorizations authorizations); @Override Query queryExtendedData(Graph graph, Element element, String tableName, String queryString, Authorizations authorizations); @Override SimilarToGraphQuery querySimilarTo(Graph graph, String[] similarToFields, String similarToText, Authorizations authorizations); @Override boolean isFieldLevelSecuritySupported(); @Override boolean isDeleteElementSupported(); void addPropertyToIndex(
        Graph graph,
        IndexInfo indexInfo,
        String propertyName,
        Object propertyValue,
        Visibility propertyVisibility
    ); @Override @Deprecated Map<Object, Long> getVertexPropertyCountByValue(Graph graph, String propertyName, Authorizations authorizations); IndexInfo ensureIndexCreatedAndInitialized(String indexName); @Override void deleteProperty(Graph graph, Element element, PropertyDescriptor property, Authorizations authorizations); @Override void deleteProperties(Graph graph, Element element, Collection<PropertyDescriptor> propertyList, Authorizations authorizations); @Override void addElements(Graph graph, Iterable<? extends Element> elements, Authorizations authorizations); @Override MultiVertexQuery queryGraph(Graph graph, String[] vertexIds, String queryString, Authorizations authorizations); @Override boolean isQuerySimilarToTextSupported(); @Override void flush(Graph graph); @Override void shutdown(); @Override boolean isFieldBoostSupported(); IndexInfo addPropertiesToIndex(Graph graph, Element element, Iterable<Property> properties); @Override synchronized void truncate(Graph graph); @Override void drop(Graph graph); IndexSelectionStrategy getIndexSelectionStrategy(); Client getClient(); ElasticsearchSearchIndexConfiguration getConfig(); IdStrategy getIdStrategy(); IndexRefreshTracker getIndexRefreshTracker(); QueryBuilder getAdditionalVisibilitiesFilter(Authorizations authorizations); boolean isPropertyInIndex(Graph graph, String propertyName, Visibility visibility); boolean isPropertyInIndex(Graph graph, String propertyName); String[] getPropertyNames(Graph graph, String propertyName, Authorizations authorizations); static final String ELEMENT_ID_FIELD_NAME; static final String ELEMENT_TYPE_FIELD_NAME; static final String VISIBILITY_FIELD_NAME; static final String HIDDEN_VERTEX_FIELD_NAME; static final String HIDDEN_PROPERTY_FIELD_NAME; static final String OUT_VERTEX_ID_FIELD_NAME; static final String IN_VERTEX_ID_FIELD_NAME; static final String EDGE_LABEL_FIELD_NAME; static final String EXTENDED_DATA_TABLE_NAME_FIELD_NAME; static final String EXTENDED_DATA_TABLE_ROW_ID_FIELD_NAME; static final String EXTENDED_DATA_TABLE_COLUMN_VISIBILITIES_FIELD_NAME; static final String ADDITIONAL_VISIBILITY_FIELD_NAME; static final String ADDITIONAL_VISIBILITY_METADATA_PREFIX; static final String EXACT_MATCH_FIELD_NAME; static final String EXACT_MATCH_PROPERTY_NAME_SUFFIX; static final String GEO_PROPERTY_NAME_SUFFIX; static final String GEO_POINT_PROPERTY_NAME_SUFFIX; static final String LOWERCASER_NORMALIZER_NAME; static final int EXACT_MATCH_IGNORE_ABOVE_LIMIT; static final String FIELDNAME_DOT_REPLACEMENT; static final int MAX_RETRIES; static final String AGGREGATION_HAS_NOT_SUFFIX; static final Pattern AGGREGATION_NAME_PATTERN; }
@Test public void testQueryExecutionCountWhenPaging() { graph.prepareVertex("v1", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.prepareVertex("v2", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.flush(); long startingNumQueries = getNumQueries(); QueryResultsIterable<Vertex> vertices = graph.query(AUTHORIZATIONS_A).vertices(); assertEquals(startingNumQueries, getNumQueries()); assertResultsCount(2, 2, vertices); assertEquals(startingNumQueries + 2, getNumQueries()); vertices = graph.query(AUTHORIZATIONS_A).limit(1).vertices(); assertEquals(startingNumQueries + 4, getNumQueries()); assertResultsCount(1, 2, vertices); assertEquals(startingNumQueries + 4, getNumQueries()); vertices = graph.query(AUTHORIZATIONS_A).limit(10).vertices(); assertEquals(startingNumQueries + 6, getNumQueries()); assertResultsCount(2, 2, vertices); assertEquals(startingNumQueries + 6, getNumQueries()); }
@Test public void testDisallowLeadingWildcardsInQueryString() { graph.prepareVertex("v1", VISIBILITY_A).setProperty("prop1", "value1", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.flush(); try { graph.query("*alue1", AUTHORIZATIONS_A).limit(0).search().getTotalHits(); fail("Wildcard prefix of query string should have caused an exception"); } catch (Exception e) { if (!(getRootCause(e) instanceof NotSerializableExceptionWrapper)) { fail("Wildcard prefix of query string should have caused a NotSerializableExceptionWrapper exception"); } } }
@Test public void testQueryReturningElasticsearchEdge() { graph.prepareVertex("v1", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.prepareVertex("v2", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.prepareEdge("e1", "v1", "v2", LABEL_LABEL1, VISIBILITY_A).save(AUTHORIZATIONS_A); graph.flush(); QueryResultsIterable<Edge> edges = graph.query(AUTHORIZATIONS_A) .edges(FetchHints.NONE); assertResultsCount(1, 1, edges); Edge e1 = toList(edges).get(0); assertEquals(LABEL_LABEL1, e1.getLabel()); assertEquals("v1", e1.getVertexId(Direction.OUT)); assertEquals("v2", e1.getVertexId(Direction.IN)); assertEquals("e1", e1.getId()); }
@Test public void testQueryReturningElasticsearchVertex() { graph.prepareVertex("v1", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.prepareVertex("v2", VISIBILITY_B).save(AUTHORIZATIONS_B); graph.prepareEdge("e1", "v1", "v2", LABEL_LABEL1, VISIBILITY_A).save(AUTHORIZATIONS_A); graph.flush(); QueryResultsIterable<Vertex> vertices = graph.query(AUTHORIZATIONS_B) .vertices(FetchHints.NONE); assertResultsCount(1, 1, vertices); Vertex vertex = toList(vertices).get(0); assertEquals("v2", vertex.getId()); }
@Test(expected = VertexiumNotSupportedException.class) public void testRetrievingVerticesFromElasticsearchEdge() { graph.prepareVertex("v1", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.prepareVertex("v2", VISIBILITY_A).save(AUTHORIZATIONS_A); graph.prepareEdge("e1", "v1", "v2", LABEL_LABEL1, VISIBILITY_A).save(AUTHORIZATIONS_A); graph.flush(); QueryResultsIterable<Edge> edges = graph.query(AUTHORIZATIONS_A) .edges(FetchHints.NONE); assertResultsCount(1, 1, edges); toList(edges).get(0).getVertices(AUTHORIZATIONS_A); }
@Test public void testQueryPagingVsScrollApi() { for (int i = 0; i < ElasticsearchResource.TEST_QUERY_PAGING_LIMIT * 2; i++) { graph.prepareVertex("v" + i, VISIBILITY_A) .addPropertyValue("k1", "prop1", "joe", VISIBILITY_A) .save(AUTHORIZATIONS_A); } graph.flush(); int resultCount = count(graph.query(AUTHORIZATIONS_A) .limit(ElasticsearchResource.TEST_QUERY_PAGING_LIMIT - 1) .vertices()); assertEquals(ElasticsearchResource.TEST_QUERY_PAGING_LIMIT - 1, resultCount); resultCount = count(graph.query(AUTHORIZATIONS_A) .limit(ElasticsearchResource.TEST_QUERY_PAGING_LIMIT + 1) .vertices()); assertEquals(ElasticsearchResource.TEST_QUERY_PAGING_LIMIT + 1, resultCount); resultCount = count(graph.query(AUTHORIZATIONS_A) .vertices()); assertEquals(ElasticsearchResource.TEST_QUERY_PAGING_LIMIT * 2, resultCount); }
@Test public void testMultipleThreadsFlushing() throws InterruptedException { assumeTrue(benchmarkEnabled()); AtomicBoolean startSignal = new AtomicBoolean(); AtomicBoolean run = new AtomicBoolean(true); AtomicBoolean writing = new AtomicBoolean(false); AtomicBoolean writeThenFlushComplete = new AtomicBoolean(false); CountDownLatch threadsReadyCountdown = new CountDownLatch(2); Runnable waitForStart = () -> { try { while (!startSignal.get()) { synchronized (startSignal) { threadsReadyCountdown.countDown(); startSignal.wait(); } } } catch (Exception ex) { throw new VertexiumException("thread failed", ex); } }; Thread constantWriteThread = new Thread(() -> { waitForStart.run(); int i = 0; while (run.get()) { graph.prepareVertex("v" + i, new Visibility("")) .addPropertyValue("k1", "name1", "value1", new Visibility("")) .save(AUTHORIZATIONS_ALL); writing.set(true); i++; } }); Thread writeThenFlushThread = new Thread(() -> { try { waitForStart.run(); while (!writing.get()) { Thread.sleep(10); } for (int i = 0; i < 5; i++) { graph.prepareVertex("vWriteTheFlush", new Visibility("")) .addPropertyValue("k1", "name1", "value1", new Visibility("")) .save(AUTHORIZATIONS_ALL); graph.flush(); } writeThenFlushComplete.set(true); } catch (Exception ex) { throw new VertexiumException("thread failed", ex); } }); constantWriteThread.start(); writeThenFlushThread.start(); threadsReadyCountdown.await(); Thread.sleep(100); synchronized (startSignal) { startSignal.set(true); startSignal.notifyAll(); } int timeout = 5000; long startTime = System.currentTimeMillis(); while (!writeThenFlushComplete.get() && (System.currentTimeMillis() - startTime < timeout)) { Thread.sleep(10); } long endTime = System.currentTimeMillis(); run.set(false); constantWriteThread.join(); writeThenFlushThread.join(); if (endTime - startTime > timeout) { fail("timeout waiting for flush"); } }
@Test public void testManyWritesToSameElement() throws InterruptedException { int threadCount = 10; int numberOfTimerToWrite = 100; Thread[] threads = new Thread[threadCount]; for (int i = 0; i < threads.length; i++) { threads[i] = new Thread(() -> { for (int write = 0; write < numberOfTimerToWrite; write++) { String keyAndValue = Thread.currentThread().getId() + "-" + write; getGraph().prepareVertex("v1", VISIBILITY_EMPTY) .addPropertyValue(keyAndValue, "name", keyAndValue, VISIBILITY_EMPTY) .save(AUTHORIZATIONS_EMPTY); getGraph().flush(); } }); threads[i].setName("testManyWritesToSameElement-" + threads[i].getId()); } for (Thread thread : threads) { thread.start(); } for (Thread thread : threads) { thread.join(); } Vertex v1 = getGraph().getVertex("v1", AUTHORIZATIONS_EMPTY); assertEquals(threadCount * numberOfTimerToWrite, count(v1.getProperties("name"))); }
@Test public void testManyWritesToSameElementNoFlushTillEnd() throws InterruptedException { int threadCount = 5; int numberOfTimerToWrite = 20; Thread[] threads = new Thread[threadCount]; for (int i = 0; i < threads.length; i++) { threads[i] = new Thread(() -> { for (int write = 0; write < numberOfTimerToWrite; write++) { String keyAndValue = Thread.currentThread().getId() + "-" + write; getGraph().prepareVertex("v1", VISIBILITY_EMPTY) .addPropertyValue(keyAndValue, "name", keyAndValue, VISIBILITY_EMPTY) .save(AUTHORIZATIONS_EMPTY); } getGraph().flush(); }); threads[i].setName("testManyWritesToSameElementNoFlushTillEnd-" + threads[i].getId()); } for (Thread thread : threads) { thread.start(); } for (Thread thread : threads) { thread.join(); } Vertex v1 = getGraph().getVertex("v1", AUTHORIZATIONS_EMPTY); assertEquals(threadCount * numberOfTimerToWrite, count(v1.getProperties("name"))); }
@Test public void testUnclosedScrollApi() { int verticesToCreate = ElasticsearchResource.TEST_QUERY_PAGE_SIZE * 2; for (int i = 0; i < verticesToCreate; i++) { getGraph().prepareVertex("v" + i, VISIBILITY_EMPTY) .addPropertyValue("k1", "name", "value1", VISIBILITY_EMPTY) .save(AUTHORIZATIONS_EMPTY); } getGraph().flush(); assertEquals(0, getCurrentScrolls()); QueryResultsIterable<Vertex> vertices = getGraph().query(AUTHORIZATIONS_EMPTY) .has("name", "value1") .limit((Long) null) .vertices(); assertEquals(0, getCurrentScrolls()); assertEquals(verticesToCreate, vertices.getTotalHits()); assertEquals(2, getCurrentScrolls()); Iterator<Vertex> it = vertices.iterator(); assertEquals(2, getCurrentScrolls()); assertTrue(it.hasNext()); it.next(); it = null; vertices = null; assertEquals(2, getCurrentScrolls()); System.gc(); System.gc(); }
@Test public void testNumberOfRefreshes() throws InterruptedException { getGraph().prepareVertex("vPRIME", VISIBILITY_EMPTY) .addPropertyValue("k1", "name", "value1", VISIBILITY_EMPTY) .save(AUTHORIZATIONS_EMPTY); getGraph().flush(); int verticesToCreate = 100; int insertIterations = 50; int queryIterations = 10; long startRefreshes = getRefreshCount(); Thread insertThread = new Thread(() -> { for (int it = 0; it < insertIterations; it++) { System.out.println("update " + it); for (int i = 0; i < verticesToCreate; i++) { getGraph().prepareVertex("v", VISIBILITY_EMPTY) .addPropertyValue("k" + i, "name", "value" + i, VISIBILITY_EMPTY) .save(AUTHORIZATIONS_EMPTY); } getGraph().flush(); } }); Thread queryThread = new Thread(() -> { for (int it = 0; it < queryIterations; it++) { System.out.println("query " + it); for (int i = 0; i < verticesToCreate; i++) { toList(getGraph().query(AUTHORIZATIONS_EMPTY) .has("name", "value" + i) .vertices()); } } }); long startTime = System.currentTimeMillis(); queryThread.start(); insertThread.start(); queryThread.join(); insertThread.join(); long endTime = System.currentTimeMillis(); System.out.println("time: " + (endTime - startTime)); long endRefreshCount = getRefreshCount(); long totalRefreshes = endRefreshCount - startRefreshes; System.out.println("refreshes: " + totalRefreshes); assertTrue( "total refreshes should be well below insert iterations times the number of vertices inserted", totalRefreshes < insertIterations * 2 ); }
@Test public void testCloseIterableClearsScrollWithNoIterators() throws IOException { int verticesToCreate = ElasticsearchResource.TEST_QUERY_PAGE_SIZE * 2; for (int i = 0; i < verticesToCreate; i++) { getGraph().prepareVertex("v" + i, VISIBILITY_EMPTY) .addPropertyValue("k1", "name", "value1", VISIBILITY_EMPTY) .save(AUTHORIZATIONS_EMPTY); } getGraph().flush(); assertEquals(0, getCurrentScrolls()); QueryResultsIterable<Vertex> vertices = getGraph().query(AUTHORIZATIONS_EMPTY) .has("name", "value1") .limit((Long) null) .vertices(); assertEquals(0, getCurrentScrolls()); assertEquals(verticesToCreate, vertices.getTotalHits()); assertEquals(2, getCurrentScrolls()); vertices.close(); assertEquals(0, getCurrentScrolls()); }
@Test public void testCompleteIteratorsClearsScroll() throws IOException { int verticesToCreate = ElasticsearchResource.TEST_QUERY_PAGE_SIZE * 2; for (int i = 0; i < verticesToCreate; i++) { getGraph().prepareVertex("v" + i, VISIBILITY_EMPTY) .addPropertyValue("k1", "name", "value1", VISIBILITY_EMPTY) .save(AUTHORIZATIONS_EMPTY); } getGraph().flush(); assertEquals(0, getCurrentScrolls()); QueryResultsIterable<Vertex> vertices = getGraph().query(AUTHORIZATIONS_EMPTY) .has("name", "value1") .limit((Long) null) .vertices(); assertEquals(0, getCurrentScrolls()); assertEquals(verticesToCreate, vertices.getTotalHits()); assertEquals(2, getCurrentScrolls()); Iterator<Vertex> iterator1 = vertices.iterator(); assertEquals(2, getCurrentScrolls()); Iterator<Vertex> iterator2 = vertices.iterator(); assertEquals(4, getCurrentScrolls()); Iterator<Vertex> iterator3 = vertices.iterator(); assertEquals(6, getCurrentScrolls()); iterator1.forEachRemaining(vertex -> {}); assertEquals(4, getCurrentScrolls()); CloseableUtils.closeQuietly(iterator2); assertEquals(2, getCurrentScrolls()); vertices.close(); assertEquals(0, getCurrentScrolls()); }