diff --git a/.travis.yml b/.travis.yml index c12d5d8f90..a273efdf1d 100644 --- a/.travis.yml +++ b/.travis.yml @@ -24,7 +24,6 @@ env: - MODULE='solr' - MODULE='es' - MODULE='es' ARGS='-Pelasticsearch2' - - MODULE='es' ARGS='-Pelasticsearch2 -Dtest=**/Transport*' - MODULE='berkeleyje' - MODULE='test' - MODULE='cassandra' ARGS='-Dtest=**/diskstorage/cassandra/thrift/* -Dtest.skip.unordered=true -Dtest.skip.ssl=true -Dtest.skip.serial=true' diff --git a/docs/elasticsearch.txt b/docs/elasticsearch.txt index 1ad7fa2f42..d301e86e4f 100644 --- a/docs/elasticsearch.txt +++ b/docs/elasticsearch.txt @@ -9,7 +9,7 @@ JanusGraph supports https://www.elastic.co/[Elasticsearch] as an index backend. * *Full-Text*: Supports all `Text` predicates to search for text properties that matches a given word, prefix or regular expression. * *Geo*: Supports all `Geo` predicates to search for geo properties that are intersecting, within, disjoint to or contained in a given query geometry. Supports points, circles, boxes, lines and polygons for indexing. Supports circles, boxes and polygons for querying point properties and all shapes for querying non-point properties. Note that JTS is required when using line and polygon shapes (see <> for more information). * *Numeric Range*: Supports all numeric comparisons in `Compare`. -* *Flexible Configuration*: Supports embedded or remote operation, custom transport and discovery, and open-ended settings customization. +* *Flexible Configuration*: Supports remote operation and open-ended settings customization. * *TTL*: Supports automatically expiring indexed elements. * *Collections*: Supports indexing SET and LIST cardinality properties. * *Temporal*: Nanosecond granularity temporal indexing. @@ -38,7 +38,7 @@ For security reasons Elasticsearch must be run under a non-root account === Elasticsearch Configuration Overview -JanusGraph supports HTTP and Transport client connections to a running Elasticsearch cluster. Please see <> for details on what versions of ES will work with the different client types in JanusGraph. +JanusGraph supports HTTP client connections to a running Elasticsearch cluster. Please see <> for details on what versions of ES will work with the different client types in JanusGraph. [NOTE] JanusGraph's index options start with the string "`index.[X].`" where "`[X]`" is a user-defined name for the backend. This user-defined name must be passed to JanusGraph's ManagementSystem interface when building a mixed index, as described in <>, so that JanusGraph knows which of potentially multiple configured index backends to use. Configuration snippets in this chapter use the name `search`, whereas prose discussion of options typically write `[X]` in the same position. The exact index name is not significant as long as it is used consistently in JanusGraph's configuration and when administering indices. @@ -52,27 +52,16 @@ The Elasticsearch client is specified as follows: [source, properties] ---- -# ES REST client -index.search.elasticsearch.interface=REST_CLIENT index.search.backend=elasticsearch ---- -[source, properties] ----- -# ES TransportClient -index.search.elasticsearch.interface=TRANSPORT_CLIENT -index.search.backend=elasticsearch ----- - -The `REST_CLIENT` and `TRANSPORT_CLIENT` values tell JanusGraph to use either the REST or Transport client, respectively. One or the other must be specified. Do not specify both in the same configuration. When connecting to Elasticsearch a single or list of hostnames for the Elasticsearch instances must be provided. These are supplied via JanusGraph's `index.[X].hostname` key. [source, properties] ---- index.search.backend=elasticsearch -index.search.elasticsearch.interface=TRANSPORT_CLIENT -index.search.hostname=10.0.0.10:9300 +index.search.hostname=10.0.0.10:9200 ---- Each host or host:port pair specified here will be added to the HTTP client's round-robin list of request targets. Here's a minimal configuration that will round-robin over 10.0.0.10 on the default Elasticsearch HTTP port (9200) and 10.0.0.20 on port 7777: @@ -80,7 +69,6 @@ Each host or host:port pair specified here will be added to the HTTP client's ro [source, properties] ---- index.search.backend=elasticsearch -index.search.elasticsearch.interface=REST_CLIENT index.search.hostname=10.0.0.10, 10.0.0.20:7777 ---- @@ -111,10 +99,6 @@ After processing `ext`, JanusGraph checks for the following common options. Jan The REST client accepts the `index.[X].bulk-refresh` option. This option controls when changes are made visible to search. See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-refresh.html[?refresh documentation] for more information. This this can also be set through the `ext` mechanism. -[[es-cfg-transport-opts]] -==== Transport Client Options - -The Transport client accepts the `index.[X].client-sniff` option. This can be set just as effectively through the `ext` mechanism. However, it can also be controlled through this JanusGraph config option. This option exists for continuity with the legacy config. === Secure Elasticsearch @@ -125,8 +109,8 @@ Elasticsearch does not perform authentication or authorization. A client that c A client uses either one protocol/port or the other, but not both simultaneously. Securing the HTTP protocol port is generally done with a combination of firewalling and a reverse proxy with SSL encryption and HTTP authentication. There are a couple of ways to approach security on the native "transport" protocol port: -Tunnel ES's native "transport" protocol:: This approach can be implemented with SSL/TLS tunneling (for instance via https://www.stunnel.org/index.html[stunnel]), a VPN, or SSH port forwarding. SSL/TLS tunnels require non-trivial setup and monitoring: one or both ends of the tunnel need a certificate, and the stunnel processes need to be configured and running continuously in order for JanusGraph and Elasticsearch to communicate. The setup for most secure VPNs is likewise non-trivial. Some Elasticsearch service providers handle server-side tunnel management and provide a custom Elasticsearch `transport.type` to simplify the client setup. JanusGraph is compatible with these custom transports. See <> for information on how to override the `transport.type` and provide arbitrary `transport.*` config keys to JanusGraph's ES client. -Add a firewall rule that allows only trusted clients to connect on Elasticsearch's native protocol port:: This is typically done at the host firewall level. This doesn't require any configuration changes in JanusGraph or Elasticsearch, nor does it require helper processes like stunnel. Easy to configure, but very weak security by itself. +Tunnel ES's native "transport" protocol:: This approach can be implemented with SSL/TLS tunneling (for instance via https://www.stunnel.org/index.html[stunnel]), a VPN, or SSH port forwarding. SSL/TLS tunnels require non-trivial setup and monitoring: one or both ends of the tunnel need a certificate, and the stunnel processes need to be configured and running continuously. The setup for most secure VPNs is likewise non-trivial. Some Elasticsearch service providers handle server-side tunnel management and provide a custom Elasticsearch `transport.type` to simplify the client setup. +Add a firewall rule that allows only trusted clients to connect on Elasticsearch's native protocol port:: This is typically done at the host firewall level. Easy to configure, but very weak security by itself. [[es-cfg-index-create]] === Index Creation Options @@ -171,14 +155,7 @@ The `create.ext` mechanism for specifying index creation settings is compatible ==== Connection Issues to remote Elasticsearch cluster -Check that the Elasticsearch cluster nodes are reachable on the HTTP and native "transport" protocol ports from the JanusGraph nodes. Check the node listen port by examining the Elasticsearch node configuration logs or using a general diagnostic utility like `netstat`. Check the JanusGraph configuration. Disable sniffing to restrict the Transport client to just the configured host list. Check that the client and server have the same major version: 1.x and 2.x are not compatible. - -==== Classpath or Field errors - -When you see exception referring to lucene implementation details, make sure you don't have a conflicting version of Lucene on the classpath. Exception may look like this: - -[source, text] -java.lang.NoSuchFieldError: LUCENE_5_5_2 +Check that the Elasticsearch cluster nodes are reachable on the HTTP protocol port from the JanusGraph nodes. Check the node listen port by examining the Elasticsearch node configuration logs or using a general diagnostic utility like `netstat`. Check the JanusGraph configuration. === Optimizing Elasticsearch @@ -190,4 +167,4 @@ For additional suggestions on how to increase write performance in Elasticsearch ==== Further Reading -* Please refer to the https://www.elastic.co[Elasticsearch homepage] and available documentation for more information on Elasticsearch and how to setup an Elasticsearch cluster. \ No newline at end of file +* Please refer to the https://www.elastic.co[Elasticsearch homepage] and available documentation for more information on Elasticsearch and how to setup an Elasticsearch cluster. diff --git a/docs/versions.txt b/docs/versions.txt index f69872170e..738889e20d 100644 --- a/docs/versions.txt +++ b/docs/versions.txt @@ -13,7 +13,7 @@ JanusGraph. [options="header"] |========================== | JanusGraph | Cassandra | HBase | Bigtable | Elasticsearch | Solr | TinkerPop -| 0.1.0 | 1.2.z, 2.0.z, 2.1.z | 0.98.z, 1.0.z, 1.1.z, 1.2.z | 0.9.z | 2.z,5.z* | 5.2.z | 3.2.z | +| 0.1.0 | 1.2.z, 2.0.z, 2.1.z | 0.98.z, 1.0.z, 1.1.z, 1.2.z | 0.9.z | 1.z*,2.z,5.z | 5.2.z | 3.2.z | |========================== -*The Elasticsearch REST client is compatible with both Elasticsearch 2.z and 5.z. The transport client is only compatibile with Elasticsearch 2.z. +*Elasticsearch 1.z compatibility is deprecated and no longer tested by default diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/indexing/IndexFeatures.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/indexing/IndexFeatures.java index d56c0cf5a7..be9ec09683 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/indexing/IndexFeatures.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/indexing/IndexFeatures.java @@ -36,11 +36,12 @@ public class IndexFeatures { private final String wildcardField; private final boolean supportsNanoseconds; private final boolean supportsCustomAnalyzer; + private final boolean supportsGeoContains; private ImmutableSet supportedCardinalities; - public IndexFeatures(boolean supportsDocumentTTL, - Mapping defaultMap, - ImmutableSet supportedMap, String wildcardField, ImmutableSet supportedCardinaities, boolean supportsNanoseconds, boolean supportCustomAnalyzer) { + public IndexFeatures(boolean supportsDocumentTTL, Mapping defaultMap, ImmutableSet supportedMap, + String wildcardField, ImmutableSet supportedCardinaities, boolean supportsNanoseconds, + boolean supportCustomAnalyzer, boolean supportsGeoContains) { Preconditions.checkArgument(defaultMap!=null || defaultMap!=Mapping.DEFAULT); Preconditions.checkArgument(supportedMap!=null && !supportedMap.isEmpty() @@ -52,6 +53,7 @@ public IndexFeatures(boolean supportsDocumentTTL, this.supportedCardinalities = supportedCardinaities; this.supportsNanoseconds = supportsNanoseconds; this.supportsCustomAnalyzer = supportCustomAnalyzer; + this.supportsGeoContains = supportsGeoContains; } public boolean supportsDocumentTTL() { @@ -82,6 +84,10 @@ public boolean supportsCustomAnalyzer() { return supportsCustomAnalyzer; } + public boolean supportsGeoContains() { + return supportsGeoContains; + } + public static class Builder { private boolean supportsDocumentTTL = false; @@ -91,6 +97,7 @@ public static class Builder { private String wildcardField = "*"; private boolean supportsNanoseconds; private boolean supportsCustomAnalyzer; + private boolean supportsGeoContains = false; public Builder supportsDocumentTTL() { supportsDocumentTTL=true; @@ -127,9 +134,15 @@ public Builder supportsCustomAnalyzer() { return this; } + public Builder supportsGeoContains() { + this.supportsGeoContains = true; + return this; + } + public IndexFeatures build() { - return new IndexFeatures(supportsDocumentTTL, defaultStringMapping, - ImmutableSet.copyOf(supportedMappings), wildcardField, ImmutableSet.copyOf(supportedCardinalities), supportsNanoseconds, supportsCustomAnalyzer); + return new IndexFeatures(supportsDocumentTTL, defaultStringMapping, ImmutableSet.copyOf(supportedMappings), + wildcardField, ImmutableSet.copyOf(supportedCardinalities), supportsNanoseconds, supportsCustomAnalyzer, + supportsGeoContains); } diff --git a/janusgraph-es/pom.xml b/janusgraph-es/pom.xml index ad03ce6a62..2aeea3dbfd 100644 --- a/janusgraph-es/pom.xml +++ b/janusgraph-es/pom.xml @@ -50,11 +50,6 @@ ${project.version} test - - org.elasticsearch - elasticsearch - ${elasticsearch.version} - org.elasticsearch.client rest @@ -219,9 +214,6 @@ default-test ${default.test.jvm.opts} -Dtest.cassandra.confdir=${project.build.directory}/cassandra/conf/localhost-murmur -Dtest.cassandra.datadir=${project.build.directory}/cassandra/data/localhost-murmur - - **/Transport*.java - @@ -261,7 +253,7 @@ elasticsearch2 - ${elasticsearch.version} + 2.4.4 true diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticMajorVersion.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticMajorVersion.java index 4731320ce1..f576e5d639 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticMajorVersion.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticMajorVersion.java @@ -16,8 +16,12 @@ public enum ElasticMajorVersion { + ONE, + TWO, - FIVE + FIVE, + + ; } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchClient.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchClient.java index 7965bf228e..6c8e3476f2 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchClient.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchClient.java @@ -14,9 +14,6 @@ package org.janusgraph.diskstorage.es; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.common.xcontent.XContentBuilder; - import java.io.Closeable; import java.io.IOException; import java.util.List; @@ -30,11 +27,11 @@ public interface ElasticSearchClient extends Closeable { boolean indexExists(String indexName) throws IOException; - void createIndex(String indexName, Settings settings) throws IOException; + void createIndex(String indexName, Map settings) throws IOException; Map getIndexSettings(String indexName) throws IOException; - void createMapping(String indexName, String typeName, XContentBuilder mapping) throws IOException; + void createMapping(String indexName, String typeName, Map mapping) throws IOException; Map getMapping(String indexName, String typeName) throws IOException; @@ -42,6 +39,6 @@ public interface ElasticSearchClient extends Closeable { void bulkRequest(List requests) throws IOException; - ElasticSearchResponse search(String indexName, String type, ElasticSearchRequest request) throws IOException; + ElasticSearchResponse search(String indexName, String type, Map request) throws IOException; } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchConstants.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchConstants.java index cfb18e8ed5..76c27014ce 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchConstants.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchConstants.java @@ -32,18 +32,8 @@ public class ElasticSearchConstants { public static final String ES_SCRIPT_KEY = "script"; public static final String ES_INLINE_KEY = "inline"; public static final String ES_LANG_KEY = "lang"; - public static final String ES_VERSION_EXPECTED; - - static { - Properties props; - - try { - props = new Properties(); - props.load(JanusGraphFactory.class.getClassLoader().getResourceAsStream(ES_PROPERTIES_FILE)); - } catch (IOException e) { - throw new AssertionError(e); - } - - ES_VERSION_EXPECTED = props.getProperty("es.version"); - } + public static final String ES_TYPE_KEY = "type"; + public static final String ES_INDEX_KEY = "index"; + public static final String ES_ANALYZER = "analyzer"; + public static final String ES_GEO_COORDS_KEY = "coordinates"; } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java index b3706876dc..a05c47ec3f 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java @@ -15,28 +15,17 @@ package org.janusgraph.diskstorage.es; import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Iterators; import com.google.common.collect.LinkedListMultimap; import com.google.common.collect.Multimap; import com.spatial4j.core.shape.Rectangle; import org.apache.commons.lang.StringUtils; -import org.elasticsearch.Version; -import org.elasticsearch.common.geo.ShapeRelation; -import org.elasticsearch.common.geo.builders.LineStringBuilder; -import org.elasticsearch.common.geo.builders.PolygonBuilder; -import org.elasticsearch.common.geo.builders.ShapeBuilder; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.common.unit.DistanceUnit; -import org.elasticsearch.common.unit.Fuzziness; -import org.elasticsearch.common.xcontent.XContentBuilder; -import org.elasticsearch.common.xcontent.XContentFactory; -import org.elasticsearch.index.query.BoolQueryBuilder; -import org.elasticsearch.index.query.GeoPolygonQueryBuilder; -import org.elasticsearch.index.query.MatchQueryBuilder.Operator; -import org.elasticsearch.index.query.QueryBuilder; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.search.sort.SortOrder; +import org.apache.tinkerpop.shaded.jackson.core.JsonProcessingException; +import org.apache.tinkerpop.shaded.jackson.databind.ObjectMapper; +import org.apache.tinkerpop.shaded.jackson.databind.ObjectWriter; +import org.apache.tinkerpop.shaded.jackson.databind.SerializationFeature; import org.janusgraph.core.Cardinality; import org.janusgraph.core.JanusGraphException; import org.janusgraph.core.attribute.Cmp; @@ -54,9 +43,13 @@ import org.janusgraph.diskstorage.configuration.ConfigOption; import org.janusgraph.diskstorage.configuration.Configuration; import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_DOC_KEY; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_INLINE_KEY; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_LANG_KEY; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_SCRIPT_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_GEO_COORDS_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_TYPE_KEY; + +import org.janusgraph.diskstorage.es.compat.AbstractESCompat; +import org.janusgraph.diskstorage.es.compat.ES1Compat; +import org.janusgraph.diskstorage.es.compat.ES2Compat; +import org.janusgraph.diskstorage.es.compat.ES5Compat; import org.janusgraph.diskstorage.indexing.IndexEntry; import org.janusgraph.diskstorage.indexing.IndexFeatures; import org.janusgraph.diskstorage.indexing.IndexMutation; @@ -68,7 +61,6 @@ import org.janusgraph.graphdb.configuration.PreInitializeConfigOptions; import static org.janusgraph.diskstorage.configuration.ConfigOption.disallowEmpty; import org.janusgraph.graphdb.database.serialize.AttributeUtil; -import org.janusgraph.graphdb.internal.Order; import org.janusgraph.graphdb.query.JanusGraphPredicate; import org.janusgraph.graphdb.query.condition.And; import org.janusgraph.graphdb.query.condition.Condition; @@ -79,7 +71,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.ByteArrayInputStream; import java.io.IOException; import java.time.Instant; import java.util.ArrayList; @@ -91,6 +82,7 @@ import java.util.UUID; import java.util.stream.Collectors; import java.util.stream.IntStream; +import java.util.stream.StreamSupport; import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.INDEX_MAX_RESULT_SET_SIZE; import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.INDEX_NAME; @@ -107,12 +99,6 @@ public class ElasticSearchIndex implements IndexProvider { private static final String STRING_MAPPING_SUFFIX = "__STRING"; - private static final String NOT_ANALYZED = "not_analyzed"; - - private static final String ANALYZER = "analyzer"; - - private static final String INDEX = "index"; - public static final ConfigNamespace ELASTICSEARCH_NS = new ConfigNamespace(INDEX_NS, "elasticsearch", "Elasticsearch index configuration"); @@ -188,28 +174,42 @@ public class ElasticSearchIndex implements IndexProvider { */ public static final double DEFAULT_GEO_DIST_ERROR_PCT = 0.025; - private static final Map SPATIAL_PREDICATES = spatialPredicates(); + private static final ObjectWriter mapWriter; + static { + final ObjectMapper mapper = new ObjectMapper(); + mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + mapWriter = mapper.writerWithView(Map.class); + } + private final AbstractESCompat compat; private final ElasticSearchClient client; private final String indexName; private final int maxResultsSize; - private final String scriptLang; private final boolean useExternalMappings; public ElasticSearchIndex(Configuration config) throws BackendException { indexName = config.get(INDEX_NAME); useExternalMappings = config.get(USE_EXTERNAL_MAPPINGS); - checkExpectedClientVersion(); - final ElasticSearchSetup.Connection c = interfaceConfiguration(config); client = c.getClient(); maxResultsSize = config.get(INDEX_MAX_RESULT_SET_SIZE); log.debug("Configured ES query result set max size to {}", maxResultsSize); - scriptLang = client.getMajorVersion() == ElasticMajorVersion.TWO ? "groovy" : "painless"; - log.debug("Using {} script language", scriptLang); + switch (client.getMajorVersion()) { + case ONE: + compat = new ES1Compat(); + break; + case TWO: + compat = new ES2Compat(); + break; + case FIVE: + compat = new ES5Compat(); + break; + default: + throw new PermanentBackendException("Unsupported Elasticsearch version: " + client.getMajorVersion()); + } try { client.clusterHealthRequest(config.get(HEALTH_REQUEST_TIMEOUT)); @@ -236,12 +236,11 @@ private void checkForOrCreateIndex(Configuration config) throws IOException { //Create index if it does not useExternalMappings and if it does not already exist if (!useExternalMappings && !client.indexExists(indexName)) { - - Settings.Builder settings = Settings.builder(); + final Map settings = new HashMap<>(); ElasticSearchSetup.applySettingsFromJanusGraphConf(settings, config, ES_CREATE_EXTRAS_NS); settings.put("index.max_result_window", Integer.MAX_VALUE); - client.createIndex(indexName, settings.build()); + client.createIndex(indexName, settings); try { final long sleep = config.get(CREATE_SLEEP); @@ -284,13 +283,6 @@ private static String getDualMappingName(String key) { return key + STRING_MAPPING_SUFFIX; } - private static Map spatialPredicates() { - return ImmutableMap.of(Geo.WITHIN, ShapeRelation.WITHIN, - Geo.CONTAINS, ShapeRelation.CONTAINS, - Geo.INTERSECT, ShapeRelation.INTERSECTS, - Geo.DISJOINT, ShapeRelation.DISJOINT); - } - @Override public void register(String store, String key, KeyInformation information, BaseTransaction tx) throws BackendException { Class dataType = information.getDataType(); @@ -323,102 +315,73 @@ public void register(String store, String key, KeyInformation information, BaseT private void pushMapping(String store, String key, KeyInformation information) throws AssertionError, PermanentBackendException, BackendException { Class dataType = information.getDataType(); Mapping map = Mapping.getMapping(information); - XContentBuilder mapping; - try { - mapping = XContentFactory.jsonBuilder(). - startObject(). - startObject("properties"). - startObject(key); - - if (AttributeUtil.isString(dataType)) { - if (map==Mapping.DEFAULT) map=Mapping.TEXT; - log.debug("Registering string type for {} with mapping {}", key, map); - mapping.field("type", "string"); - String stringAnalyzer = (String) ParameterType.STRING_ANALYZER.findParameter(information.getParameters(), null); - String textAnalyzer = (String) ParameterType.TEXT_ANALYZER.findParameter(information.getParameters(), null); - switch (map) { - case STRING: - if (stringAnalyzer != null) { - mapping.field(ANALYZER, stringAnalyzer); - } else { - mapping.field(INDEX, NOT_ANALYZED); - } - break; - case TEXT: - if (textAnalyzer != null) { - mapping.field(ANALYZER, textAnalyzer); - } - break; - case TEXTSTRING: - if (textAnalyzer != null) { - mapping.field(ANALYZER, textAnalyzer); - } - mapping.endObject(); - //add string mapping - mapping.startObject(getDualMappingName(key)); - mapping.field("type", "string"); - if (stringAnalyzer != null) { - mapping.field(ANALYZER, stringAnalyzer); - } else { - mapping.field(INDEX, NOT_ANALYZED); - } - break; - default: throw new AssertionError("Unexpected mapping: "+map); - } - } else if (dataType == Float.class) { - log.debug("Registering float type for {}", key); - mapping.field("type", "float"); - } else if (dataType == Double.class) { - log.debug("Registering double type for {}", key); - mapping.field("type", "double"); - } else if (dataType == Byte.class) { - log.debug("Registering byte type for {}", key); - mapping.field("type", "byte"); - } else if (dataType == Short.class) { - log.debug("Registering short type for {}", key); - mapping.field("type", "short"); - } else if (dataType == Integer.class) { - log.debug("Registering integer type for {}", key); - mapping.field("type", "integer"); - } else if (dataType == Long.class) { - log.debug("Registering long type for {}", key); - mapping.field("type", "long"); - } else if (dataType == Boolean.class) { - log.debug("Registering boolean type for {}", key); - mapping.field("type", "boolean"); - } else if (dataType == Geoshape.class) { - switch (map) { - case PREFIX_TREE: - int maxLevels = (int) ParameterType.INDEX_GEO_MAX_LEVELS.findParameter(information.getParameters(), DEFAULT_GEO_MAX_LEVELS); - double distErrorPct = (double) ParameterType.INDEX_GEO_DIST_ERROR_PCT.findParameter(information.getParameters(), DEFAULT_GEO_DIST_ERROR_PCT); - log.debug("Registering geo_shape type for {} with tree_levels={} and distance_error_pct={}", key, maxLevels, distErrorPct); - mapping.field("type", "geo_shape"); - mapping.field("tree", "quadtree"); - mapping.field("tree_levels", maxLevels); - mapping.field("distance_error_pct", distErrorPct); - break; - default: - log.debug("Registering geo_point type for {}", key); - mapping.field("type", "geo_point"); - } - } else if (dataType == Date.class || dataType == Instant.class) { - log.debug("Registering date type for {}", key); - mapping.field("type", "date"); - } else if (dataType == Boolean.class) { - log.debug("Registering boolean type for {}", key); - mapping.field("type", "boolean"); - } else if (dataType == UUID.class) { - log.debug("Registering uuid type for {}", key); - mapping.field("type", "string"); - mapping.field(INDEX, NOT_ANALYZED); + final Map properties = new HashMap<>(); + if (AttributeUtil.isString(dataType)) { + if (map==Mapping.DEFAULT) map=Mapping.TEXT; + log.debug("Registering string type for {} with mapping {}", key, map); + String stringAnalyzer = (String) ParameterType.STRING_ANALYZER.findParameter(information.getParameters(), null); + String textAnalyzer = (String) ParameterType.TEXT_ANALYZER.findParameter(information.getParameters(), null); + // use keyword type for string mappings unless custom string analyzer is provided + final Map stringMapping = stringAnalyzer == null ? compat.createKeywordMapping() : compat.createTextMapping(stringAnalyzer); + switch (map) { + case STRING: + properties.put(key, stringMapping); + break; + case TEXT: + properties.put(key, compat.createTextMapping(textAnalyzer)); + break; + case TEXTSTRING: + properties.put(key, compat.createTextMapping(textAnalyzer)); + properties.put(getDualMappingName(key), stringMapping); + break; + default: throw new AssertionError("Unexpected mapping: "+map); } - - mapping.endObject().endObject().endObject(); - - } catch (IOException e) { - throw new PermanentBackendException("Could not render json for put mapping request", e); + } else if (dataType == Float.class) { + log.debug("Registering float type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "float")); + } else if (dataType == Double.class) { + log.debug("Registering double type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "double")); + } else if (dataType == Byte.class) { + log.debug("Registering byte type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "byte")); + } else if (dataType == Short.class) { + log.debug("Registering short type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "short")); + } else if (dataType == Integer.class) { + log.debug("Registering integer type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "integer")); + } else if (dataType == Long.class) { + log.debug("Registering long type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "long")); + } else if (dataType == Boolean.class) { + log.debug("Registering boolean type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "boolean")); + } else if (dataType == Geoshape.class) { + switch (map) { + case PREFIX_TREE: + int maxLevels = (int) ParameterType.INDEX_GEO_MAX_LEVELS.findParameter(information.getParameters(), DEFAULT_GEO_MAX_LEVELS); + double distErrorPct = (double) ParameterType.INDEX_GEO_DIST_ERROR_PCT.findParameter(information.getParameters(), DEFAULT_GEO_DIST_ERROR_PCT); + log.debug("Registering geo_shape type for {} with tree_levels={} and distance_error_pct={}", key, maxLevels, distErrorPct); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "geo_shape", + "tree", "quadtree", + "tree_levels", maxLevels, + "distance_error_pct", distErrorPct)); + break; + default: + log.debug("Registering geo_point type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "geo_point")); + } + } else if (dataType == Date.class || dataType == Instant.class) { + log.debug("Registering date type for {}", key); + properties.put(key, ImmutableMap.of(ES_TYPE_KEY, "date")); + } else if (dataType == UUID.class) { + log.debug("Registering uuid type for {}", key); + properties.put(key, compat.createKeywordMapping()); } + final Map mapping = ImmutableMap.of("properties", properties); + try { client.createMapping(indexName, store, mapping); } catch (Exception e) { @@ -548,7 +511,7 @@ public void mutate(Map> mutations, KeyInforma requests.add(ElasticSearchMutation.createDeleteRequest(indexName, storename, docid)); } else { String script = getDeletionScript(informations, storename, mutation); - Map doc = ImmutableMap.of(ES_SCRIPT_KEY, ImmutableMap.of(ES_INLINE_KEY, script, ES_LANG_KEY, scriptLang)); + Map doc = compat.prepareScript(script).build(); requests.add(ElasticSearchMutation.createUpdateRequest(indexName, storename, docid, doc)); log.trace("Adding script {}", script); } @@ -568,8 +531,7 @@ public void mutate(Map> mutations, KeyInforma String inline = getAdditionScript(informations, storename, mutation); if (!inline.isEmpty()) { - Map script = ImmutableMap.of(ES_INLINE_KEY, inline, ES_LANG_KEY, scriptLang); - final ImmutableMap.Builder builder = ImmutableMap.builder().put(ES_SCRIPT_KEY, script); + final ImmutableMap.Builder builder = compat.prepareScript(inline); requests.add(ElasticSearchMutation.createUpdateRequest(indexName, storename, docid, builder, upsert)); log.trace("Adding script {}", inline); } @@ -589,7 +551,7 @@ public void mutate(Map> mutations, KeyInforma client.bulkRequest(requests); } } catch (Exception e) { - log.error("Failed to execute bulk Elasticsearch query", e); + log.error("Failed to execute bulk Elasticsearch mutation", e); throw convert(e); } } @@ -608,7 +570,7 @@ private String getDeletionScript(KeyInformation.IndexRetriever informations, Str break; case SET: case LIST: - String jsValue = convertToJsType(deletion.value, scriptLang, Mapping.getMapping(keyInformation)); + String jsValue = convertToJsType(deletion.value, compat.scriptLang(), Mapping.getMapping(keyInformation)); script.append("def index = ctx._source[\"").append(deletion.field).append("\"].indexOf(").append(jsValue).append("); ctx._source[\"").append(deletion.field).append("\"].remove(index);"); if (hasDualStringMapping(informations.get(storename, deletion.field))) { script.append("def index = ctx._source[\"").append(getDualMappingName(deletion.field)).append("\"].indexOf(").append(jsValue).append("); ctx._source[\"").append(getDualMappingName(deletion.field)).append("\"].remove(index);"); @@ -627,9 +589,9 @@ private String getAdditionScript(KeyInformation.IndexRetriever informations, Str switch (keyInformation.getCardinality()) { case SET: case LIST: - script.append("ctx._source[\"").append(e.field).append("\"].add(").append(convertToJsType(e.value, scriptLang, Mapping.getMapping(keyInformation))).append(");"); + script.append("ctx._source[\"").append(e.field).append("\"].add(").append(convertToJsType(e.value, compat.scriptLang(), Mapping.getMapping(keyInformation))).append(");"); if (hasDualStringMapping(keyInformation)) { - script.append("ctx._source[\"").append(getDualMappingName(e.field)).append("\"].add(").append(convertToJsType(e.value, scriptLang, Mapping.getMapping(keyInformation))).append(");"); + script.append("ctx._source[\"").append(getDualMappingName(e.field)).append("\"].add(").append(convertToJsType(e.value, compat.scriptLang(), Mapping.getMapping(keyInformation))).append(");"); } break; default: @@ -657,30 +619,13 @@ private Map getAdditionDoc(KeyInformation.IndexRetriever informat } private static String convertToJsType(Object value, String scriptLang, Mapping mapping) throws PermanentBackendException { + final String esValue; try { - XContentBuilder builder = XContentFactory.jsonBuilder().startObject(); - - Object esValue = convertToEsType(value, mapping); - if (esValue instanceof byte[]) { - builder.rawField("value", new ByteArrayInputStream((byte[]) esValue)); - } else { - builder.field("value", esValue); - } - - builder.endObject(); - String s = builder.string(); - int prefixLength = "{\"value\":".length(); - int suffixLength = "}".length(); - String result = s.substring(prefixLength, s.length() - suffixLength); - if (scriptLang.equals("groovy")) { - result = result.replace("$", "\\$"); - } - return result; + esValue = mapWriter.writeValueAsString(convertToEsType(value, mapping)); } catch (IOException e) { throw new PermanentBackendException("Could not write json"); } - - + return scriptLang.equals("groovy") ? esValue.replace("$", "\\$") : esValue; } @@ -717,7 +662,7 @@ public void restore(Map>> documents, KeyInfo } } - public QueryBuilder getFilter(Condition condition, KeyInformation.StoreRetriever informations) { + public Map getFilter(Condition condition, KeyInformation.StoreRetriever informations) { if (condition instanceof PredicateCondition) { PredicateCondition atom = (PredicateCondition) condition; Object value = atom.getValue(); @@ -730,45 +675,52 @@ public QueryBuilder getFilter(Condition condition, KeyInformation.StoreRetrie switch (numRel) { case EQUAL: - return QueryBuilders.termsQuery(key, value); + return compat.term(key, value); case NOT_EQUAL: - return QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery(key, value)); + return compat.boolMustNot(compat.term(key, value)); case LESS_THAN: - return QueryBuilders.rangeQuery(key).lt(value); + return compat.lt(key, value); case LESS_THAN_EQUAL: - return QueryBuilders.rangeQuery(key).lte(value); + return compat.lte(key, value); case GREATER_THAN: - return QueryBuilders.rangeQuery(key).gt(value); + return compat.gt(key, value); case GREATER_THAN_EQUAL: - return QueryBuilders.rangeQuery(key).gte(value); + return compat.gte(key, value); default: throw new IllegalArgumentException("Unexpected relation: " + numRel); } } else if (value instanceof String) { Mapping map = getStringMapping(informations.get(key)); - String fieldName = key; - if (map==Mapping.TEXT && !janusgraphPredicate.toString().startsWith("CONTAINS")) + if (map==Mapping.TEXT && !janusgraphPredicate.toString().startsWith(Text.CONTAINS.name())) throw new IllegalArgumentException("Text mapped string values only support CONTAINS queries and not: " + janusgraphPredicate); if (map==Mapping.STRING && janusgraphPredicate.toString().startsWith("CONTAINS")) throw new IllegalArgumentException("String mapped string values do not support CONTAINS queries: " + janusgraphPredicate); - if (map==Mapping.TEXTSTRING && !janusgraphPredicate.toString().startsWith("CONTAINS")) + + final String fieldName; + if (map==Mapping.TEXTSTRING && !janusgraphPredicate.toString().startsWith(Text.CONTAINS.name())) { fieldName = getDualMappingName(key); + } else { + fieldName = key; + } + if (janusgraphPredicate == Text.CONTAINS || janusgraphPredicate == Cmp.EQUAL) { - return QueryBuilders.matchQuery(fieldName, value).operator(Operator.AND); + return compat.match(key, value); } else if (janusgraphPredicate == Text.CONTAINS_PREFIX) { - value = ParameterType.TEXT_ANALYZER.findParameter(informations.get(key).getParameters(), null)!=null?((String) value):((String) value).toLowerCase(); - return QueryBuilders.prefixQuery(fieldName, (String) value); + if (!ParameterType.TEXT_ANALYZER.hasParameter(informations.get(key).getParameters())) + value = ((String) value).toLowerCase(); + return compat.prefix(fieldName, value); } else if (janusgraphPredicate == Text.CONTAINS_REGEX) { - value = ParameterType.TEXT_ANALYZER.findParameter(informations.get(key).getParameters(), null)!=null?((String) value):((String) value).toLowerCase(); - return QueryBuilders.regexpQuery(fieldName, (String) value); + if (!ParameterType.TEXT_ANALYZER.hasParameter(informations.get(key).getParameters())) + value = ((String) value).toLowerCase(); + return compat.regexp(fieldName, value); } else if (janusgraphPredicate == Text.PREFIX) { - return QueryBuilders.prefixQuery(fieldName, (String) value); + return compat.prefix(fieldName, value); } else if (janusgraphPredicate == Text.REGEX) { - return QueryBuilders.regexpQuery(fieldName, (String) value); + return compat.regexp(fieldName, value); } else if (janusgraphPredicate == Cmp.NOT_EQUAL) { - return QueryBuilders.boolQuery().mustNot(QueryBuilders.matchQuery(fieldName, value).operator(Operator.AND)); - } else if (janusgraphPredicate == Text.FUZZY || janusgraphPredicate == Text.CONTAINS_FUZZY){ - return QueryBuilders.matchQuery(fieldName, (String) value).fuzziness(Fuzziness.AUTO).operator(Operator.AND); + return compat.boolMustNot(compat.match(fieldName, value)); + } else if (janusgraphPredicate == Text.FUZZY || janusgraphPredicate == Text.CONTAINS_FUZZY) { + return compat.fuzzyMatch(fieldName, value); } else throw new IllegalArgumentException("Predicate is not supported for string value: " + janusgraphPredicate); } else if (value instanceof Geoshape && Mapping.getMapping(informations.get(key)) == Mapping.DEFAULT) { @@ -776,79 +728,83 @@ public QueryBuilder getFilter(Condition condition, KeyInformation.StoreRetrie Geoshape shape = (Geoshape) value; Preconditions.checkArgument(janusgraphPredicate instanceof Geo && janusgraphPredicate != Geo.CONTAINS, "Relation not supported on geopoint types: " + janusgraphPredicate); - final QueryBuilder queryBuilder; + final Map query; if (shape.getType() == Geoshape.Type.CIRCLE) { Geoshape.Point center = shape.getPoint(); - queryBuilder = QueryBuilders.geoDistanceQuery(key).lat(center.getLatitude()).lon(center.getLongitude()).distance(shape.getRadius(), DistanceUnit.KILOMETERS); + query = compat.geoDistance(key, center.getLatitude(), center.getLongitude(), shape.getRadius()); } else if (shape.getType() == Geoshape.Type.BOX) { Geoshape.Point southwest = shape.getPoint(0); Geoshape.Point northeast = shape.getPoint(1); - queryBuilder = QueryBuilders.geoBoundingBoxQuery(key).bottomRight(southwest.getLatitude(), northeast.getLongitude()).topLeft(northeast.getLatitude(), southwest.getLongitude()); + query = compat.geoBoundingBox(key, southwest.getLatitude(), southwest.getLongitude(), northeast.getLatitude(), northeast.getLongitude()); } else if (shape.getType() == Geoshape.Type.POLYGON) { - queryBuilder = QueryBuilders.geoPolygonQuery(key); - IntStream.range(0, shape.size()).forEach(i -> { - Geoshape.Point point = shape.getPoint(i); - ((GeoPolygonQueryBuilder) queryBuilder).addPoint(point.getLatitude(), point.getLongitude()); - }); + final List> points = IntStream.range(0, shape.size()) + .mapToObj(i -> ImmutableList.of(shape.getPoint(i).getLongitude(), shape.getPoint(i).getLatitude())) + .collect(Collectors.toList()); + query = compat.geoPolygon(key, points); } else { throw new IllegalArgumentException("Unsupported or invalid search shape type for geopoint: " + shape.getType()); } - return janusgraphPredicate == Geo.DISJOINT ? QueryBuilders.boolQuery().mustNot(queryBuilder) : queryBuilder; + return janusgraphPredicate == Geo.DISJOINT ? compat.boolMustNot(query) : query; } else if (value instanceof Geoshape) { - // geoshape Preconditions.checkArgument(janusgraphPredicate instanceof Geo, "Relation not supported on geoshape types: " + janusgraphPredicate); Geoshape shape = (Geoshape) value; - final ShapeBuilder sb; + final Map geo; switch (shape.getType()) { case CIRCLE: Geoshape.Point center = shape.getPoint(); - sb = ShapeBuilder.newCircleBuilder().center(center.getLongitude(), center.getLatitude()).radius(shape.getRadius(), DistanceUnit.KILOMETERS); + geo = ImmutableMap.of(ES_TYPE_KEY, "circle", + ES_GEO_COORDS_KEY, ImmutableList.of(center.getLongitude(), center.getLatitude()), + "radius", shape.getRadius() + "km"); break; case BOX: Geoshape.Point southwest = shape.getPoint(0); Geoshape.Point northeast = shape.getPoint(1); - sb = ShapeBuilder.newEnvelope().bottomRight(northeast.getLongitude(),southwest.getLatitude()).topLeft(southwest.getLongitude(),northeast.getLatitude()); + geo = ImmutableMap.of(ES_TYPE_KEY, "envelope", + ES_GEO_COORDS_KEY, ImmutableList.of(ImmutableList.of(southwest.getLongitude(),northeast.getLatitude()), + ImmutableList.of(northeast.getLongitude(),southwest.getLatitude()))); break; case LINE: - sb = ShapeBuilder.newLineString(); - IntStream.range(0, shape.size()).forEach(i -> { - Geoshape.Point point = shape.getPoint(i); - ((LineStringBuilder) sb).point(point.getLongitude(), point.getLatitude()); - }); + final List lineCoords = IntStream.range(0, shape.size()) + .mapToObj(i -> ImmutableList.of(shape.getPoint(i).getLongitude(), shape.getPoint(i).getLatitude())) + .collect(Collectors.toList()); + geo = ImmutableMap.of(ES_TYPE_KEY, "linestring", ES_GEO_COORDS_KEY, lineCoords); break; case POLYGON: - sb = ShapeBuilder.newPolygon(); - IntStream.range(0, shape.size()).forEach(i -> { - Geoshape.Point point = shape.getPoint(i); - ((PolygonBuilder) sb).point(point.getLongitude(), point.getLatitude()); - }); + final List polyCoords = IntStream.range(0, shape.size()) + .mapToObj(i -> ImmutableList.of(shape.getPoint(i).getLongitude(), shape.getPoint(i).getLatitude())) + .collect(Collectors.toList()); + geo = ImmutableMap.of(ES_TYPE_KEY, "polygon", ES_GEO_COORDS_KEY, ImmutableList.of(polyCoords)); break; case POINT: - sb = ShapeBuilder.newPoint(shape.getPoint().getLongitude(),shape.getPoint().getLatitude()); + geo = ImmutableMap.of(ES_TYPE_KEY, "point", + ES_GEO_COORDS_KEY, ImmutableList.of(shape.getPoint().getLongitude(),shape.getPoint().getLatitude())); break; default: throw new IllegalArgumentException("Unsupported or invalid search shape type: " + shape.getType()); } - return QueryBuilders.geoShapeQuery(key, sb, SPATIAL_PREDICATES.get((Geo) janusgraphPredicate)); + return compat.geoShape(key, geo, (Geo) janusgraphPredicate); } else if (value instanceof Date || value instanceof Instant) { Preconditions.checkArgument(janusgraphPredicate instanceof Cmp, "Relation not supported on date types: " + janusgraphPredicate); Cmp numRel = (Cmp) janusgraphPredicate; + if (value instanceof Instant) { + value = Date.from((Instant) value); + } switch (numRel) { case EQUAL: - return QueryBuilders.termsQuery(key, value); + return compat.term(key, value); case NOT_EQUAL: - return QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery(key, value)); + return compat.boolMustNot(compat.term(key, value)); case LESS_THAN: - return QueryBuilders.rangeQuery(key).lt(value); + return compat.lt(key, value); case LESS_THAN_EQUAL: - return QueryBuilders.rangeQuery(key).lte(value); + return compat.lte(key, value); case GREATER_THAN: - return QueryBuilders.rangeQuery(key).gt(value); + return compat.gt(key, value); case GREATER_THAN_EQUAL: - return QueryBuilders.rangeQuery(key).gte(value); + return compat.gte(key, value); default: throw new IllegalArgumentException("Unexpected relation: " + numRel); } @@ -856,53 +812,48 @@ public QueryBuilder getFilter(Condition condition, KeyInformation.StoreRetrie Cmp numRel = (Cmp) janusgraphPredicate; switch (numRel) { case EQUAL: - return QueryBuilders.termsQuery(key, value); + return compat.term(key, value); case NOT_EQUAL: - return QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery(key, value)); + return compat.boolMustNot(compat.term(key, value)); default: throw new IllegalArgumentException("Boolean types only support EQUAL or NOT_EQUAL"); } - } else if (value instanceof UUID) { if (janusgraphPredicate == Cmp.EQUAL) { - return QueryBuilders.termQuery(key, value); + return compat.term(key, value); } else if (janusgraphPredicate == Cmp.NOT_EQUAL) { - return QueryBuilders.boolQuery().mustNot(QueryBuilders.termQuery(key, value)); + return compat.boolMustNot(compat.term(key, value)); } else { throw new IllegalArgumentException("Only equal or not equal is supported for UUIDs: " + janusgraphPredicate); } } else throw new IllegalArgumentException("Unsupported type: " + value); } else if (condition instanceof Not) { - return QueryBuilders.boolQuery().mustNot(getFilter(((Not) condition).getChild(),informations)); + return compat.boolMustNot(getFilter(((Not) condition).getChild(),informations)); } else if (condition instanceof And) { - BoolQueryBuilder b = QueryBuilders.boolQuery(); - for (Condition c : condition.getChildren()) { - b.must(getFilter(c,informations)); - } - return b; + final List queries = StreamSupport.stream(condition.getChildren().spliterator(), false) + .map(c -> getFilter(c,informations)).collect(Collectors.toList()); + return compat.boolMust(queries); } else if (condition instanceof Or) { - BoolQueryBuilder b = QueryBuilders.boolQuery(); - b.minimumNumberShouldMatch(1); - for (Condition c : condition.getChildren()) { - b.should(getFilter(c,informations)); - } - return b; + final List queries = StreamSupport.stream(condition.getChildren().spliterator(), false) + .map(c -> getFilter(c,informations)).collect(Collectors.toList()); + return compat.boolShould(queries); } else throw new IllegalArgumentException("Invalid condition: " + condition); } @Override public List query(IndexQuery query, KeyInformation.IndexRetriever informations, BaseTransaction tx) throws BackendException { ElasticSearchRequest sr = new ElasticSearchRequest(); - sr.setQuery(getFilter(query.getCondition(),informations.get(query.getStore()))); + final Map esQuery = getFilter(query.getCondition(), informations.get(query.getStore())); + sr.setQuery(compat.prepareQuery(esQuery)); if (!query.getOrder().isEmpty()) { List orders = query.getOrder(); for (int i = 0; i < orders.size(); i++) { IndexQuery.OrderEntry orderEntry = orders.get(i); - String order = (orderEntry.getOrder() == Order.ASC ? SortOrder.ASC : SortOrder.DESC).toString(); + String order = orderEntry.getOrder().name(); KeyInformation information = informations.get(query.getStore()).get(orders.get(i).getKey()); Mapping mapping = Mapping.getMapping(information); Class datatype = orderEntry.getDatatype(); - sr.addSort(orders.get(i).getKey(), order, convertToEsDataType(datatype, mapping)); + sr.addSort(orders.get(i).getKey(), order.toLowerCase(), convertToEsDataType(datatype, mapping)); } } sr.setFrom(0); @@ -911,7 +862,7 @@ public List query(IndexQuery query, KeyInformation.IndexRetriever inform ElasticSearchResponse response; try { - response = client.search(indexName, query.getStore(), sr); + response = client.search(indexName, query.getStore(), compat.createRequestBody(sr)); } catch (IOException e) { throw new PermanentBackendException(e); } @@ -957,7 +908,7 @@ else if (Geoshape.class.isAssignableFrom(datatype)) { @Override public Iterable> query(RawQuery query, KeyInformation.IndexRetriever informations, BaseTransaction tx) throws BackendException { ElasticSearchRequest sr = new ElasticSearchRequest(); - sr.setQuery(QueryBuilders.queryStringQuery(query.getQuery())); + sr.setQuery(compat.queryString(query.getQuery())); sr.setFrom(query.getOffset()); if (query.hasLimit()) sr.setSize(query.getLimit()); @@ -965,7 +916,7 @@ public Iterable> query(RawQuery query, KeyInformation.In ElasticSearchResponse response; try { - response = client.search(indexName, query.getStore(), sr); + response = client.search(indexName, query.getStore(), compat.createRequestBody(sr)); } catch (IOException e) { throw new PermanentBackendException(e); } @@ -1035,7 +986,7 @@ public String mapKey2Field(String key, KeyInformation information) { @Override public IndexFeatures getFeatures() { - return ES_FEATURES; + return compat.getIndexFeatures(); } @Override @@ -1064,22 +1015,4 @@ public void clearStorage() throws BackendException { } } - private void checkExpectedClientVersion() { - /* - * This is enclosed in a catch block to prevent an unchecked exception - * from killing the startup thread. This check is just advisory -- the - * most it does is log a warning -- so there's no reason to allow it to - * emit a exception and potentially block graph startup. - */ - try { - if (!Version.CURRENT.toString().equals(ElasticSearchConstants.ES_VERSION_EXPECTED)) { - log.warn("ES client version ({}) does not match the version with which JanusGraph was compiled ({}). This might cause problems.", - Version.CURRENT, ElasticSearchConstants.ES_VERSION_EXPECTED); - } else { - log.debug("Found ES client version matching JanusGraph's compile-time version: {} (OK)", Version.CURRENT); - } - } catch (RuntimeException e) { - log.warn("Unable to check expected ES client version", e); - } - } } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchRequest.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchRequest.java index 3330733b90..45f1cea28d 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchRequest.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchRequest.java @@ -16,7 +16,6 @@ import com.google.common.collect.ImmutableMap; import org.apache.tinkerpop.shaded.jackson.annotation.JsonProperty; -import org.elasticsearch.index.query.QueryBuilder; import java.util.ArrayList; import java.util.List; @@ -24,9 +23,7 @@ public class ElasticSearchRequest { - private QueryBuilder query; - - private QueryBuilder postFilter; + private Map query; private Integer size; @@ -38,22 +35,14 @@ public ElasticSearchRequest() { this.sorts = new ArrayList<>(); } - public QueryBuilder getQuery() { + public Map getQuery() { return query; } - public void setQuery(QueryBuilder query) { + public void setQuery(Map query) { this.query = query; } - public QueryBuilder getPostFilter() { - return postFilter; - } - - public void setPostFilter(QueryBuilder postFilter) { - this.postFilter = postFilter; - } - public Integer getSize() { return size; } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchSetup.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchSetup.java index 23ce3f73bd..55bbd28f4a 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchSetup.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchSetup.java @@ -19,9 +19,6 @@ import org.apache.commons.lang.StringUtils; import org.apache.http.HttpHost; import org.elasticsearch.client.RestClient; -import org.elasticsearch.client.transport.TransportClient; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.common.transport.InetSocketTransportAddress; import org.janusgraph.diskstorage.configuration.ConfigNamespace; import org.janusgraph.diskstorage.configuration.ConfigOption; import org.janusgraph.diskstorage.configuration.Configuration; @@ -31,14 +28,14 @@ import org.slf4j.LoggerFactory; import java.io.FileInputStream; -import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; import java.lang.reflect.Array; -import java.net.InetAddress; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Properties; import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.INDEX_CONF_FILE; import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.INDEX_HOSTS; @@ -73,44 +70,6 @@ */ public enum ElasticSearchSetup { - /** - * Start an ES TransportClient connected to - * {@link org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration#INDEX_HOSTS}. - */ - TRANSPORT_CLIENT { - @Override - public Connection connect(Configuration config) throws IOException { - log.debug("Configuring TransportClient"); - - Settings.Builder settingsBuilder = settingsBuilder(config); - - if (config.has(ElasticSearchIndex.CLIENT_SNIFF)) { - String k = "client.transport.sniff"; - settingsBuilder.put(k, config.get(ElasticSearchIndex.CLIENT_SNIFF)); - log.debug("Set {}: {}", k, config.get(ElasticSearchIndex.CLIENT_SNIFF)); - } - - settingsBuilder.put("index.max_result_window", Integer.MAX_VALUE); - - TransportClient tc = TransportClient.builder().settings(settingsBuilder.build()).build(); - int defaultPort = config.has(INDEX_PORT) ? config.get(INDEX_PORT) : ElasticSearchIndex.HOST_PORT_DEFAULT; - for (String host : config.get(INDEX_HOSTS)) { - String[] hostparts = host.split(":"); - String hostname = hostparts[0]; - int hostport = defaultPort; - if (hostparts.length == 2) hostport = Integer.parseInt(hostparts[1]); - log.info("Configured remote host: {} : {}", hostname, hostport); - tc.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(hostname), hostport)); - } - - TransportElasticSearchClient client = new TransportElasticSearchClient(tc); - if (config.has(ElasticSearchIndex.BULK_REFRESH)) { - client.setBulkRefresh(!"false".equals(config.get(ElasticSearchIndex.BULK_REFRESH))); - } - return new Connection(client); - } - }, - /** * Create an ES RestClient connected to * {@link org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration#INDEX_HOSTS}. @@ -127,7 +86,7 @@ public Connection connect(Configuration config) throws IOException { String hostname = hostparts[0]; int hostport = defaultPort; if (hostparts.length == 2) hostport = Integer.parseInt(hostparts[1]); - log.info("Configured remote host: {} : {}", hostname, hostport); + log.debug("Configured remote host: {} : {}", hostname, hostport); hosts.add(new HttpHost(hostname, hostport, "http")); } RestClient rc = RestClient.builder(hosts.toArray(new HttpHost[hosts.size()])).build(); @@ -165,9 +124,9 @@ public Connection connect(Configuration config) throws IOException { * @return ES settings builder configured according to the {@code config} parameter * @throws java.io.IOException if conf-file was set but could not be read */ - private static Settings.Builder settingsBuilder(Configuration config) throws IOException { + private static Map settingsBuilder(Configuration config) throws IOException { - Settings.Builder settings = Settings.settingsBuilder(); + Map settings = new HashMap<>(); // Set JanusGraph defaults settings.put("client.transport.ignore_cluster_name", true); @@ -199,7 +158,7 @@ private static Settings.Builder settingsBuilder(Configuration config) throws IOE // Force-enable inline scripting. This is probably only useful in Node mode. String inlineScriptsKey = "script.inline"; - String inlineScriptsVal = settings.get(inlineScriptsKey); + String inlineScriptsVal = (String) settings.get(inlineScriptsKey); if (null != inlineScriptsVal && !"true".equals(inlineScriptsVal)) { log.error("JanusGraph requires Elasticsearch inline scripting but found {} set to false", inlineScriptsKey); throw new IOException("JanusGraph requires Elasticsearch inline scripting"); @@ -210,16 +169,18 @@ private static Settings.Builder settingsBuilder(Configuration config) throws IOE return settings; } - static void applySettingsFromFile(Settings.Builder settings, + static void applySettingsFromFile(Map settings, Configuration config, - ConfigOption confFileOption) throws FileNotFoundException { + ConfigOption confFileOption) throws IOException { if (config.has(confFileOption)) { String confFile = config.get(confFileOption); log.debug("Loading Elasticsearch settings from file {}", confFile); InputStream confStream = null; try { confStream = new FileInputStream(confFile); - settings.loadFromStream(confFile, confStream); + final Properties properties = new Properties(); + properties.load(confStream); + properties.stringPropertyNames().stream().forEach(key -> settings.put(key, properties.get(key))); } finally { IOUtils.closeQuietly(confStream); } @@ -228,7 +189,7 @@ static void applySettingsFromFile(Settings.Builder settings, } } - static void applySettingsFromJanusGraphConf(Settings.Builder settings, + static void applySettingsFromJanusGraphConf(Map settings, Configuration config, ConfigNamespace rootNS) { int keysLoaded = 0; diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/TransportElasticSearchClient.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/TransportElasticSearchClient.java deleted file mode 100644 index 930d4bec76..0000000000 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/TransportElasticSearchClient.java +++ /dev/null @@ -1,227 +0,0 @@ -// Copyright 2017 JanusGraph Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package org.janusgraph.diskstorage.es; - -import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; -import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; -import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest; -import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse; -import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; -import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; -import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest; -import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse; -import org.elasticsearch.action.bulk.BulkItemResponse; -import org.elasticsearch.action.bulk.BulkRequestBuilder; -import org.elasticsearch.action.bulk.BulkResponse; -import org.elasticsearch.action.delete.DeleteRequest; -import org.elasticsearch.action.index.IndexRequest; -import org.elasticsearch.action.search.SearchRequestBuilder; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.elasticsearch.client.Client; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.common.xcontent.XContentBuilder; -import org.elasticsearch.index.IndexNotFoundException; -import org.elasticsearch.node.Node; -import org.elasticsearch.rest.RestStatus; -import org.elasticsearch.script.Script; -import org.elasticsearch.script.ScriptService; -import org.elasticsearch.search.SearchHit; -import org.elasticsearch.search.SearchHits; -import org.elasticsearch.search.sort.FieldSortBuilder; -import org.elasticsearch.search.sort.SortOrder; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_DOC_KEY; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_INLINE_KEY; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_LANG_KEY; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_SCRIPT_KEY; -import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_UPSERT_KEY; - -import org.janusgraph.diskstorage.indexing.RawQuery; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; - -public class TransportElasticSearchClient implements ElasticSearchClient { - - private static final Logger log = LoggerFactory.getLogger(TransportElasticSearchClient.class); - - private Client client; - - private boolean bulkRefresh; - - public TransportElasticSearchClient(Client client) { - this.client = client; - } - - @Override - public void clusterHealthRequest(String timeout) throws IOException { - client.admin().cluster().prepareHealth().setTimeout(timeout).setWaitForYellowStatus().execute().actionGet(); - } - - @Override - public boolean indexExists(String indexName) throws IOException { - IndicesExistsResponse response = client.admin().indices().exists(new IndicesExistsRequest(indexName)).actionGet(); - return response.isExists(); - } - - @Override - public void createIndex(String indexName, Settings settings) throws IOException { - CreateIndexResponse create = client.admin().indices().prepareCreate(indexName) - .setSettings(settings).execute().actionGet(); - } - - @Override - public Map getIndexSettings(String indexName) throws IOException { - GetSettingsResponse response = client.admin().indices().getSettings(new GetSettingsRequest().indices(indexName)).actionGet(); - return response.getIndexToSettings().get(indexName).getAsMap().entrySet().stream() - .collect(Collectors.toMap(e->e.getKey().replace("index.",""), Map.Entry::getValue)); - } - - @Override - public void createMapping(String indexName, String typeName, XContentBuilder mapping) throws IOException { - client.admin().indices().preparePutMapping(indexName).setType(typeName).setSource(mapping).execute().actionGet(); - } - - @Override - public Map getMapping(String indexName, String typeName) throws IOException { - GetMappingsResponse response = client.admin().indices().getMappings(new GetMappingsRequest().indices(indexName.toLowerCase()).types(typeName)).actionGet(); - return (Map)response.getMappings().get(indexName.toLowerCase()).get(typeName).getSourceAsMap().get("properties"); - } - - @Override - public void deleteIndex(String indexName) throws IOException { - try { - client.admin().indices().delete(new DeleteIndexRequest(indexName)).actionGet(); - // We wait for one second to let ES delete the river - Thread.sleep(1000); - } catch (IndexNotFoundException e) { - // Index does not exist... Fine - } catch (InterruptedException e) { - throw new IOException(e); - } - } - - @Override - public void bulkRequest(List requests) throws IOException { - BulkRequestBuilder brb = client.prepareBulk(); - requests.stream().forEach(request -> { - String indexName = request.getIndex(); - String type = request.getType(); - String id = request.getId(); - switch (request.getRequestType()) { - case DELETE: { - brb.add(new DeleteRequest(indexName, type, id)); - break; - } case INDEX: { - brb.add(new IndexRequest(indexName, type, id).source(request.getSource())); - break; - } case UPDATE: { - UpdateRequestBuilder update = client.prepareUpdate(indexName, type, id); - if (request.getSource().containsKey(ES_SCRIPT_KEY)) { - Map script = ((Map) request.getSource().get(ES_SCRIPT_KEY)); - String inline = script.get(ES_INLINE_KEY); - String lang = script.get(ES_LANG_KEY); - update.setScript(new Script(inline, ScriptService.ScriptType.INLINE, lang, null)); - } - if (request.getSource().containsKey(ES_DOC_KEY)) { - update.setDoc((Map) request.getSource().get(ES_DOC_KEY)); - } - if (request.getSource().containsKey(ES_UPSERT_KEY)) { - update.setUpsert((Map) request.getSource().get(ES_UPSERT_KEY)); - } - brb.add(update); - break; - } default: - throw new IllegalArgumentException("Unsupported request type: " + request.getRequestType()); - } - }); - - if (!requests.isEmpty()) { - if (bulkRefresh) { - brb.setRefresh(true); - } - BulkResponse bulkItemResponses = brb.execute().actionGet(); - if (bulkItemResponses.hasFailures()) { - boolean actualFailure = false; - for(BulkItemResponse response : bulkItemResponses.getItems()) { - //The document may have been deleted, which is OK - if(response.isFailed() && response.getFailure().getStatus() != RestStatus.NOT_FOUND) { - log.error("Failed to execute ES query {}", response.getFailureMessage()); - actualFailure = true; - } - } - if(actualFailure) { - throw new IOException("Failure(s) in Elasicsearch bulk request: " + bulkItemResponses.buildFailureMessage()); - } - } - } - } - - @Override - public ElasticSearchResponse search(String indexName, String type, ElasticSearchRequest request) throws IOException { - SearchRequestBuilder srb = client.prepareSearch(indexName); - srb.setTypes(type); - srb.setQuery(request.getQuery()); - srb.setPostFilter(request.getPostFilter()); - if (request.getFrom() != null) { - srb.setFrom(request.getFrom()); - } - if (request.getSize() != null) { - srb.setSize(request.getSize()); - } - request.getSorts().stream().flatMap(item -> item.entrySet().stream()).forEach(item -> { - String key = item.getKey(); - ElasticSearchRequest.RestSortInfo sortInfo = item.getValue(); - FieldSortBuilder fsb = new FieldSortBuilder(key) - .order(SortOrder.valueOf(sortInfo.getOrder().toUpperCase())) - .unmappedType(sortInfo.getUnmappedType()); - srb.addSort(fsb); - }); - - SearchResponse response = srb.execute().actionGet(); - SearchHits hits = response.getHits(); - - List> results = new ArrayList<>(hits.hits().length); - for (SearchHit hit : hits) { - results.add(new RawQuery.Result(hit.id(),hit.getScore())); - } - - ElasticSearchResponse result = new ElasticSearchResponse(); - result.setTook(response.getTookInMillis()); - result.setTotal(hits.getTotalHits()); - result.setResults(results); - return result; - } - - @Override - public void close() throws IOException { - client.close(); - } - - @Override - public ElasticMajorVersion getMajorVersion() { - return ElasticMajorVersion.TWO; - } - - public void setBulkRefresh(boolean bulkRefresh) { - this.bulkRefresh = bulkRefresh; - } - -} diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/AbstractESCompat.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/AbstractESCompat.java new file mode 100644 index 0000000000..75fcae5ed4 --- /dev/null +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/AbstractESCompat.java @@ -0,0 +1,192 @@ +// Copyright 2017 JanusGraph Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package org.janusgraph.diskstorage.es.compat; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.janusgraph.core.Cardinality; +import org.janusgraph.core.attribute.Geo; +import org.janusgraph.core.schema.Mapping; +import org.janusgraph.diskstorage.es.ElasticSearchRequest; +import org.janusgraph.diskstorage.indexing.IndexFeatures; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_INLINE_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_LANG_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_SCRIPT_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_TYPE_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_ANALYZER; + +/** + * Base class for building Elasticsearch mapping and query objects. + */ +public abstract class AbstractESCompat { + + static final Map MATCH_ALL = ImmutableMap.of("match_all", Collections.EMPTY_MAP); + + static final IndexFeatures.Builder coreFeatures() { + return new IndexFeatures.Builder() + .setDefaultStringMapping(Mapping.TEXT) + .supportedStringMappings(Mapping.TEXT, Mapping.TEXTSTRING, Mapping.STRING) + .setWildcardField("_all") + .supportsCardinality(Cardinality.SINGLE) + .supportsCardinality(Cardinality.LIST) + .supportsCardinality(Cardinality.SET) + .supportsNanoseconds() + .supportsCustomAnalyzer() + ; + } + + public abstract IndexFeatures getIndexFeatures(); + + public Map createKeywordMapping() { + return ImmutableMap.of(ES_TYPE_KEY, "keyword"); + } + + public Map createTextMapping(String textAnalyzer) { + final ImmutableMap.Builder builder = ImmutableMap.builder().put(ES_TYPE_KEY, "text"); + return (textAnalyzer != null ? builder.put(ES_ANALYZER, textAnalyzer) : builder).build(); + } + + public String scriptLang() { + return "painless"; + } + + public ImmutableMap.Builder prepareScript(String inline) { + final Map script = ImmutableMap.of(ES_INLINE_KEY, inline, ES_LANG_KEY, scriptLang()); + return ImmutableMap.builder().put(ES_SCRIPT_KEY, script); + } + + public Map prepareQuery(Map query) { + return query; + } + + public Map term(String key, Object value) { + return ImmutableMap.of("term", ImmutableMap.of(key, value)); + } + + public Map contains(String key, List terms) { + return boolMust(terms.stream().map(term -> term(key, term)).collect(Collectors.toList())); + } + + public Map boolMust(List> queries) { + return queries.size() > 1 ? ImmutableMap.of("bool", ImmutableMap.of("must", queries)) : queries.get(0); + } + + public Map boolMustNot(Map query) { + return ImmutableMap.of("bool", ImmutableMap.of("must_not", query)); + } + + public Map boolShould(List> queries) { + return ImmutableMap.of("bool", ImmutableMap.of("should", queries)); + } + + public Map boolFilter(Map query) { + return ImmutableMap.of("bool", ImmutableMap.of("must", MATCH_ALL, "filter", query)); + } + + public Map lt(String key, Object value) { + return ImmutableMap.of("range", ImmutableMap.of(key, ImmutableMap.of("lt", value))); + } + + public Map lte(String key, Object value) { + return ImmutableMap.of("range", ImmutableMap.of(key, ImmutableMap.of("lte", value))); + } + + public Map gt(String key, Object value) { + return ImmutableMap.of("range", ImmutableMap.of(key, ImmutableMap.of("gt", value))); + } + + public Map gte(String key, Object value) { + return ImmutableMap.of("range", ImmutableMap.of(key, ImmutableMap.of("gte", value))); + } + + public Map prefix(String key, Object value) { + return ImmutableMap.of("prefix", ImmutableMap.of(key, value)); + } + + public Map regexp(String key, Object value) { + return ImmutableMap.of("regexp", ImmutableMap.of(key, value)); + } + + public Map match(String key, Object value) { + return match(key, value, null); + } + + public Map fuzzyMatch(String key, Object value) { + return match(key, value, "AUTO"); + } + + public Map match(String key, Object value, String fuzziness) { + final ImmutableMap.Builder builder = ImmutableMap.builder().put("query", value); + builder.put("operator", "and"); + if (fuzziness != null) builder.put("fuzziness", fuzziness); + return ImmutableMap.of("match", ImmutableMap.of(key, builder.build())); + } + + public Map queryString(Object query) { + return ImmutableMap.of("query_string", ImmutableMap.of("query", query)); + } + + public Map geoDistance(String key, double lat, double lon, double radius) { + return filter(ImmutableMap.of("geo_distance", ImmutableMap.of("distance", radius + "km", key, ImmutableList.of(lon, lat)))); + } + + public Map geoBoundingBox(String key, double minLat, double minLon, double maxLat, double maxLon) { + return filter(ImmutableMap.of("geo_bounding_box", ImmutableMap.of(key,ImmutableMap.of( + "top_left", ImmutableList.of(minLon, maxLat),"bottom_right", ImmutableList.of(maxLon, minLat))))); + } + + public Map geoPolygon(String key, List> points) { + return filter(ImmutableMap.of("geo_polygon", ImmutableMap.of(key, ImmutableMap.of("points", points)))); + } + + public Map geoShape(String key, Map geoShape, Geo predicate) { + final String relation = predicate == Geo.INTERSECT ? "intersects" : predicate.name().toLowerCase(); + return filter(ImmutableMap.of("geo_shape", ImmutableMap.of(key, ImmutableMap.of("shape", geoShape, "relation", relation)))); + } + + public Map filter(Map query) { + return boolFilter(query); + } + + public Map createRequestBody(ElasticSearchRequest request) { + final Map requestBody = new HashMap<>(); + + if (request.getSize() != null) { + requestBody.put("size", request.getSize()); + } + + if (request.getFrom() != null) { + requestBody.put("from", request.getFrom()); + } + + if (!request.getSorts().isEmpty()) { + requestBody.put("sort", request.getSorts()); + } + + if (request.getQuery() != null) { + requestBody.put("query", request.getQuery()); + } + + return requestBody; + } + +} diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES1Compat.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES1Compat.java new file mode 100644 index 0000000000..87e733ccc8 --- /dev/null +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES1Compat.java @@ -0,0 +1,57 @@ +// Copyright 2017 JanusGraph Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package org.janusgraph.diskstorage.es.compat; + +import com.google.common.collect.ImmutableMap; +import org.janusgraph.diskstorage.indexing.IndexFeatures; + +import java.util.Map; + +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_LANG_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_SCRIPT_KEY; + +/** + * Mapping and query object builder for Elasticsearch 1.x. + */ +public class ES1Compat extends ES2Compat { + + private static final IndexFeatures FEATURES = coreFeatures().build(); + + @Override + public IndexFeatures getIndexFeatures() { + return FEATURES; + } + + @Override + public ImmutableMap.Builder prepareScript(String script) { + return ImmutableMap.builder().put(ES_SCRIPT_KEY, ImmutableMap.of(ES_SCRIPT_KEY, script, ES_LANG_KEY, scriptLang())); + } + + @Override + public Map prepareQuery(Map query) { + return ImmutableMap.of("filtered", ImmutableMap.of("filter", query)); + } + + @Override + public Map match(String key, Object value, String fuzziness) { + return ImmutableMap.of("query", super.match(key, value, fuzziness)); + } + + @Override + public Map filter(Map query) { + return query; + } + +} diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES2Compat.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES2Compat.java new file mode 100644 index 0000000000..a6d760024f --- /dev/null +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES2Compat.java @@ -0,0 +1,56 @@ +// Copyright 2017 JanusGraph Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package org.janusgraph.diskstorage.es.compat; + +import com.google.common.collect.ImmutableMap; +import org.janusgraph.diskstorage.indexing.IndexFeatures; + +import java.util.Map; + +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_INDEX_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_TYPE_KEY; +import static org.janusgraph.diskstorage.es.ElasticSearchConstants.ES_ANALYZER; + +/** + * Mapping and query object builder for Elasticsearch 2.x. + */ +public class ES2Compat extends AbstractESCompat { + + private static final IndexFeatures FEATURES = coreFeatures().supportsGeoContains().build(); + + private static final String STRING_TYPE_NAME = "string"; + + @Override + public String scriptLang() { + return "groovy"; + } + + @Override + public Map createKeywordMapping() { + return ImmutableMap.of(ES_TYPE_KEY, STRING_TYPE_NAME, ES_INDEX_KEY, "not_analyzed"); + } + + @Override + public Map createTextMapping(String textAnalyzer) { + final ImmutableMap.Builder builder = ImmutableMap.builder().put(ES_TYPE_KEY, STRING_TYPE_NAME); + return (textAnalyzer != null ? builder.put(ES_ANALYZER, textAnalyzer) : builder).build(); + } + + @Override + public IndexFeatures getIndexFeatures() { + return FEATURES; + } + +} diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportElasticSearchConfigTest.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES5Compat.java similarity index 60% rename from janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportElasticSearchConfigTest.java rename to janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES5Compat.java index 19f968f94c..e0a47d5140 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportElasticSearchConfigTest.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/compat/ES5Compat.java @@ -12,12 +12,20 @@ // See the License for the specific language governing permissions and // limitations under the License. -package org.janusgraph.diskstorage.es; +package org.janusgraph.diskstorage.es.compat; -public class TransportElasticSearchConfigTest extends ElasticSearchConfigTest { +import org.janusgraph.diskstorage.indexing.IndexFeatures; - public ElasticSearchSetup getInterface() { - return ElasticSearchSetup.TRANSPORT_CLIENT; +/** + * Mapping and query object builder for Elasticsearch 5.x. + */ +public class ES5Compat extends AbstractESCompat { + + private static final IndexFeatures FEATURES = coreFeatures().supportsGeoContains().build(); + + @Override + public IndexFeatures getIndexFeatures() { + return FEATURES; } } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestBulkResponse.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestBulkResponse.java index efc317de35..3f4a38b49a 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestBulkResponse.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestBulkResponse.java @@ -15,6 +15,7 @@ package org.janusgraph.diskstorage.es.rest; import org.apache.tinkerpop.shaded.jackson.annotation.JsonIgnoreProperties; +import org.apache.tinkerpop.shaded.jackson.annotation.JsonRawValue; import java.util.List; import java.util.Map; @@ -49,7 +50,7 @@ public static class RestBulkItemResponse { private int status; - private Map error; + private Object error; public String getResult() { return result; @@ -67,11 +68,11 @@ public void setStatus(int status) { this.status = status; } - public Map getError() { + public Object getError() { return error; } - public void setError(Map error) { + public void setError(Object error) { this.error = error; } } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java index e137814fa9..84d8c91158 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java @@ -26,14 +26,10 @@ import org.apache.tinkerpop.shaded.jackson.databind.module.SimpleModule; import org.elasticsearch.client.Response; import org.elasticsearch.client.RestClient; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.common.xcontent.XContentBuilder; -import org.elasticsearch.rest.RestStatus; import org.janusgraph.core.attribute.Geoshape; import org.janusgraph.diskstorage.es.ElasticMajorVersion; import org.janusgraph.diskstorage.es.ElasticSearchClient; import org.janusgraph.diskstorage.es.ElasticSearchMutation; -import org.janusgraph.diskstorage.es.ElasticSearchRequest; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -66,6 +62,8 @@ public class RestElasticSearchClient implements ElasticSearchClient { mapWriter = mapper.writerWithView(Map.class); } + private static final ElasticMajorVersion DEFAULT_VERSION = ElasticMajorVersion.FIVE; + private RestClient delegate; private ElasticMajorVersion majorVersion; @@ -89,20 +87,29 @@ public ElasticMajorVersion getMajorVersion() { } final Pattern pattern = Pattern.compile("(\\d+)\\.\\d+\\.\\d+"); - majorVersion = ElasticMajorVersion.TWO; + majorVersion = DEFAULT_VERSION; try { final Response response = delegate.performRequest("GET", "/"); try (final InputStream inputStream = response.getEntity().getContent()) { final ClusterInfo info = mapper.readValue(inputStream, ClusterInfo.class); - final Matcher m = info.getVersion() != null ? pattern.matcher((String) info.getVersion().get("number")) : null; - if (m == null || !m.find() || Integer.valueOf(m.group(1)) < 5) { - majorVersion = ElasticMajorVersion.TWO; - } else { - majorVersion = ElasticMajorVersion.FIVE; + final String version = info.getVersion() != null ? (String) info.getVersion().get("number") : null; + final Matcher m = version != null ? pattern.matcher(version) : null; + switch (m != null && m.find() ? Integer.valueOf(m.group(1)) : -1) { + case 1: + majorVersion = ElasticMajorVersion.ONE; + break; + case 2: + majorVersion = ElasticMajorVersion.TWO; + break; + case 5: + majorVersion = ElasticMajorVersion.FIVE; + break; + default: + throw new IllegalArgumentException("Unsupported Elasticsearch server version: " + version); } } } catch (Exception e) { - log.warn("Unable to determine Elasticsearch server version. Assuming 2.x.", e); + log.warn("Unable to determine Elasticsearch server version. Default to {}.", majorVersion, e); } return majorVersion; @@ -137,8 +144,8 @@ public boolean indexExists(String indexName) throws IOException { } @Override - public void createIndex(String indexName, Settings settings) throws IOException { - performRequest("PUT", "/" + indexName, mapWriter.writeValueAsBytes(settings.getAsMap())); + public void createIndex(String indexName, Map settings) throws IOException { + performRequest("PUT", "/" + indexName, mapWriter.writeValueAsBytes(settings)); } @Override @@ -151,9 +158,8 @@ public Map getIndexSettings(String indexName) throws IOException { } @Override - public void createMapping(String indexName, String typeName, XContentBuilder mapping) throws IOException { - byte[] bytes = mapping.bytes().toBytes(); - performRequest("PUT", "/" + indexName + "/_mapping/" + typeName, bytes); + public void createMapping(String indexName, String typeName, Map mapping) throws IOException { + performRequest("PUT", "/" + indexName + "/_mapping/" + typeName, mapWriter.writeValueAsBytes(mapping)); } @Override @@ -198,49 +204,26 @@ public void bulkRequest(List requests) throws IOException final Response response = performRequest("POST", builder.toString(), outputStream.toByteArray()); try (final InputStream inputStream = response.getEntity().getContent()) { final RestBulkResponse bulkResponse = mapper.readValue(inputStream, RestBulkResponse.class); - List> errors = bulkResponse.getItems().stream() + final List errors = bulkResponse.getItems().stream() .flatMap(item -> item.values().stream()) - .filter(item -> item.getError() != null && item.getStatus() != RestStatus.NOT_FOUND.getStatus()) + .filter(item -> item.getError() != null && item.getStatus() != 404) .map(item -> item.getError()).collect(Collectors.toList()); if (!errors.isEmpty()) { - errors.forEach(error -> log.error("Failed to execute ES query {}", error.get("reason"))); - throw new IOException("Failure(s) in Elasicsearch bulk request: " + mapper.writeValueAsString(errors)); + errors.forEach(error -> log.error("Failed to execute ES query: {}", error)); + throw new IOException("Failure(s) in Elasicsearch bulk request: " + errors); } } } @Override - public RestSearchResponse search(String indexName, String type, ElasticSearchRequest request) throws IOException { + public RestSearchResponse search(String indexName, String type, Map request) throws IOException { final String path = "/" + indexName + "/" + type + "/_search"; - final Map requestBody = new HashMap<>(); - - if (request.getSize() != null) { - requestBody.put("size", request.getSize()); - } - - if (request.getFrom() != null) { - requestBody.put("from", request.getFrom()); - } - - if (!request.getSorts().isEmpty()) { - requestBody.put("sort", request.getSorts()); - } - - if (request.getQuery() != null) { - final Map query = mapReader.readValue(request.getQuery().buildAsBytes().array()); - requestBody.put("query", query); - } - - if (request.getPostFilter() != null) { - final Map query = mapReader.readValue(request.getPostFilter().buildAsBytes().array()); - requestBody.put("post_filter", query); - } - - final byte[] requestData = mapper.writeValueAsBytes(requestBody); + final byte[] requestData = mapper.writeValueAsBytes(request); if (log.isDebugEnabled()) { - log.debug("Elasticsearch request: " + mapper.writerWithDefaultPrettyPrinter().writeValueAsString(requestBody)); + log.debug("Elasticsearch request: " + mapper.writerWithDefaultPrettyPrinter().writeValueAsString(request)); } + Response response = performRequest("POST", path, requestData); try (final InputStream inputStream = response.getEntity().getContent()) { return mapper.readValue(inputStream, RestSearchResponse.class); diff --git a/janusgraph-es/src/main/resources/janusgraph-es.properties b/janusgraph-es/src/main/resources/janusgraph-es.properties index ba7739e9cb..41f4b0be9c 100644 --- a/janusgraph-es/src/main/resources/janusgraph-es.properties +++ b/janusgraph-es/src/main/resources/janusgraph-es.properties @@ -1,2 +1 @@ -es.version=${elasticsearch.version} es.dist.version=${elasticsearch.dist.version} diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/BerkeleyElasticsearchTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/BerkeleyElasticsearchTest.java index b11fb9e0d0..4c828b0eaa 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/BerkeleyElasticsearchTest.java +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/BerkeleyElasticsearchTest.java @@ -41,13 +41,17 @@ public class BerkeleyElasticsearchTest extends JanusGraphIndexTest { @BeforeClass public static void startElasticsearch() { - esr = new ElasticsearchRunner(); - esr.start(); + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr = new ElasticsearchRunner(); + esr.start(); + } } @AfterClass public static void stopElasticsearch() { - esr.stop(); + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr.stop(); + } } public BerkeleyElasticsearchTest() { diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchConfigTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchConfigTest.java index 4c6dc5b808..6090278102 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchConfigTest.java +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchConfigTest.java @@ -73,9 +73,9 @@ public class ElasticSearchConfigTest { private static final String ANALYZER_STANDARD = "standard"; - private ElasticsearchRunner esr; + private static final int PORT = 9200; - private int port; + private ElasticsearchRunner esr; private HttpHost host; @@ -83,9 +83,12 @@ public class ElasticSearchConfigTest { @Before public void setup() throws Exception { - esr = new ElasticsearchRunner(); - esr.start(); - port = getInterface() == ElasticSearchSetup.REST_CLIENT ? 9200 : 9300; + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr = new ElasticsearchRunner(); + esr.start(); + Thread.sleep(5000); + } + httpClient = HttpClients.createDefault(); try { host = new HttpHost(InetAddress.getByName("127.0.0.1"), 9200); @@ -96,19 +99,17 @@ public void setup() throws Exception { @After public void teardown() throws Exception { - esr.stop(); + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr.stop(); + } IOUtils.closeQuietly(httpClient); } - public ElasticSearchSetup getInterface() { - return ElasticSearchSetup.REST_CLIENT; - } - @Test public void testJanusGraphFactoryBuilder() { JanusGraphFactory.Builder builder = JanusGraphFactory.build(); builder.set("storage.backend", "inmemory"); - builder.set("index." + INDEX_NAME + ".elasticsearch.hostname", "127.0.0.1:" + port); + builder.set("index." + INDEX_NAME + ".elasticsearch.hostname", "127.0.0.1:" + PORT); JanusGraph graph = builder.open(); // Must not throw an exception assertTrue(graph.isOpen()); graph.close(); @@ -117,16 +118,16 @@ public void testJanusGraphFactoryBuilder() { @Test public void testClient() throws BackendException, InterruptedException { ModifiableConfiguration config = GraphDatabaseConfiguration.buildGraphConfiguration(); - config.set(INTERFACE, getInterface().toString(), INDEX_NAME); - config.set(INDEX_HOSTS, new String[]{ "127.0.0.1:" + port }, INDEX_NAME); + config.set(INTERFACE, ElasticSearchSetup.REST_CLIENT.toString(), INDEX_NAME); + config.set(INDEX_HOSTS, new String[]{ "127.0.0.1:" + PORT }, INDEX_NAME); Configuration indexConfig = config.restrictTo(INDEX_NAME); - IndexProvider idx = new ElasticSearchIndex(indexConfig); + IndexProvider idx = open(indexConfig); simpleWriteAndQuery(idx); idx.close(); config = GraphDatabaseConfiguration.buildGraphConfiguration(); - config.set(INTERFACE, getInterface().toString(), INDEX_NAME); - config.set(INDEX_HOSTS, new String[]{ "10.11.12.13:" + port }, INDEX_NAME); + config.set(INTERFACE, ElasticSearchSetup.REST_CLIENT.toString(), INDEX_NAME); + config.set(INDEX_HOSTS, new String[]{ "10.11.12.13:" + PORT }, INDEX_NAME); indexConfig = config.restrictTo(INDEX_NAME); Throwable failure = null; try { @@ -147,14 +148,14 @@ public void testIndexCreationOptions() throws InterruptedException, BackendExcep ModifiableConfiguration config = new ModifiableConfiguration(GraphDatabaseConfiguration.ROOT_NS, cc, BasicConfiguration.Restriction.NONE); - config.set(INTERFACE, getInterface().toString(), INDEX_NAME); - config.set(INDEX_HOSTS, new String[]{ "127.0.0.1:" + port }, INDEX_NAME); + config.set(INTERFACE, ElasticSearchSetup.REST_CLIENT.toString(), INDEX_NAME); + config.set(INDEX_HOSTS, new String[]{ "127.0.0.1:" + PORT }, INDEX_NAME); config.set(GraphDatabaseConfiguration.INDEX_NAME, "janusgraph_creation_opts", INDEX_NAME); Configuration indexConfig = config.restrictTo(INDEX_NAME); - IndexProvider idx = new ElasticSearchIndex(indexConfig); + IndexProvider idx = open(indexConfig); simpleWriteAndQuery(idx); - ElasticSearchClient client = getInterface().connect(indexConfig).getClient(); + ElasticSearchClient client = ElasticSearchSetup.REST_CLIENT.connect(indexConfig).getClient(); assertEquals(String.valueOf(shards), client.getIndexSettings("janusgraph_creation_opts").get("number_of_shards")); @@ -276,4 +277,12 @@ private void executeRequest(HttpRequestBase request) throws IOException { IOUtils.closeQuietly(res); } } + + private IndexProvider open(Configuration indexConfig) throws BackendException { + final ElasticSearchIndex idx = new ElasticSearchIndex(indexConfig); + idx.clearStorage(); + idx.close(); + return new ElasticSearchIndex(indexConfig); + } + } diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchIndexTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchIndexTest.java index 24a090c8f9..440b357549 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchIndexTest.java +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchIndexTest.java @@ -57,13 +57,17 @@ public class ElasticSearchIndexTest extends IndexProviderTest { @BeforeClass public static void startElasticsearch() { - esr = new ElasticsearchRunner(); - esr.start(); + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr = new ElasticsearchRunner(); + esr.start(); + } } @AfterClass public static void stopElasticsearch() { - esr.stop(); + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr.stop(); + } } @Override diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticsearchRunner.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticsearchRunner.java index 8100349f2e..06da1450e8 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticsearchRunner.java +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticsearchRunner.java @@ -41,6 +41,8 @@ public class ElasticsearchRunner extends DaemonRunner { public static final String ES_PID_FILE = "/tmp/janusgraph-test-es.pid"; + public static final boolean IS_EXTERNAL = Boolean.valueOf(System.getProperty("is.external.es", "false")); + public ElasticsearchRunner(String esHome) { final Pattern VERSION_PATTERN = Pattern.compile("es.dist.version=(.*)"); String version = null; diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ThriftElasticsearchTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ThriftElasticsearchTest.java index 8a3d7248f2..b109088c14 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ThriftElasticsearchTest.java +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ThriftElasticsearchTest.java @@ -34,13 +34,17 @@ public class ThriftElasticsearchTest extends JanusGraphIndexTest { @BeforeClass public static void startElasticsearch() { CassandraStorageSetup.startCleanEmbedded(); - esr = new ElasticsearchRunner(); - esr.start(); + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr = new ElasticsearchRunner(); + esr.start(); + } } @AfterClass public static void stopElasticsearch() { - esr.stop(); + if (!ElasticsearchRunner.IS_EXTERNAL) { + esr.stop(); + } } public ThriftElasticsearchTest() { diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportBerkeleyElasticsearchTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportBerkeleyElasticsearchTest.java deleted file mode 100644 index 3f86a49e6f..0000000000 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportBerkeleyElasticsearchTest.java +++ /dev/null @@ -1,38 +0,0 @@ -// Copyright 2017 JanusGraph Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package org.janusgraph.diskstorage.es; - -import org.janusgraph.diskstorage.configuration.ModifiableConfiguration; -import org.janusgraph.diskstorage.configuration.WriteConfiguration; - -import static org.janusgraph.BerkeleyStorageSetup.getBerkeleyJEConfiguration; -import static org.janusgraph.diskstorage.es.ElasticSearchIndex.BULK_REFRESH; -import static org.janusgraph.diskstorage.es.ElasticSearchIndex.INTERFACE; -import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.INDEX_HOSTS; - -public class TransportBerkeleyElasticsearchTest extends BerkeleyElasticsearchTest { - - @Override - public WriteConfiguration getConfiguration() { - ModifiableConfiguration config = getBerkeleyJEConfiguration(); - //Add index - config.set(INTERFACE, ElasticSearchSetup.TRANSPORT_CLIENT.toString(), INDEX); - config.set(INDEX_HOSTS, new String[]{ "127.0.0.1:9300" }, INDEX); - config.set(BULK_REFRESH, "true", INDEX); - return config.getConfiguration(); - - } - -} diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportElasticSearchIndexTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportElasticSearchIndexTest.java deleted file mode 100644 index f3318f63a5..0000000000 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportElasticSearchIndexTest.java +++ /dev/null @@ -1,37 +0,0 @@ -// Copyright 2017 JanusGraph Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package org.janusgraph.diskstorage.es; - -import org.janusgraph.diskstorage.configuration.Configuration; -import org.janusgraph.diskstorage.configuration.ModifiableConfiguration; -import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration; - -import static org.janusgraph.diskstorage.es.ElasticSearchIndex.BULK_REFRESH; -import static org.janusgraph.diskstorage.es.ElasticSearchIndex.INTERFACE; -import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.INDEX_HOSTS; - -public class TransportElasticSearchIndexTest extends ElasticSearchIndexTest { - - @Override - public Configuration getESTestConfig() { - final String index = "es"; - ModifiableConfiguration config = GraphDatabaseConfiguration.buildGraphConfiguration(); - config.set(INTERFACE, ElasticSearchSetup.TRANSPORT_CLIENT.toString(), index); - config.set(INDEX_HOSTS, new String[]{ "127.0.0.1:9300" }, index); - config.set(BULK_REFRESH, "true", index); - return config.restrictTo(index); - } - -} diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportThriftElasticsearchTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportThriftElasticsearchTest.java deleted file mode 100644 index 8dc8fc62c2..0000000000 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/TransportThriftElasticsearchTest.java +++ /dev/null @@ -1,39 +0,0 @@ -// Copyright 2017 JanusGraph Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package org.janusgraph.diskstorage.es; - - -import org.janusgraph.diskstorage.configuration.ModifiableConfiguration; -import org.janusgraph.diskstorage.configuration.WriteConfiguration; - -import static org.janusgraph.CassandraStorageSetup.getCassandraThriftConfiguration; -import static org.janusgraph.diskstorage.es.ElasticSearchIndex.BULK_REFRESH; -import static org.janusgraph.diskstorage.es.ElasticSearchIndex.INTERFACE; -import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.INDEX_HOSTS; - -public class TransportThriftElasticsearchTest extends ThriftElasticsearchTest { - - @Override - public WriteConfiguration getConfiguration() { - ModifiableConfiguration config = - getCassandraThriftConfiguration(TransportThriftElasticsearchTest.class.getName()); - //Add index - config.set(INTERFACE, ElasticSearchSetup.TRANSPORT_CLIENT.toString(), INDEX); - config.set(INDEX_HOSTS, new String[]{ "127.0.0.1:9300" }, INDEX); - config.set(BULK_REFRESH, "true", INDEX); - return config.getConfiguration(); - } - -} diff --git a/janusgraph-lucene/src/main/java/org/janusgraph/diskstorage/lucene/LuceneIndex.java b/janusgraph-lucene/src/main/java/org/janusgraph/diskstorage/lucene/LuceneIndex.java index 81d5d88bab..21ad03965b 100644 --- a/janusgraph-lucene/src/main/java/org/janusgraph/diskstorage/lucene/LuceneIndex.java +++ b/janusgraph-lucene/src/main/java/org/janusgraph/diskstorage/lucene/LuceneIndex.java @@ -87,7 +87,12 @@ public class LuceneIndex implements IndexProvider { private static final Version LUCENE_VERSION = Version.LUCENE_5_5_2; - private static final IndexFeatures LUCENE_FEATURES = new IndexFeatures.Builder().supportedStringMappings(Mapping.TEXT, Mapping.STRING).supportsCardinality(Cardinality.SINGLE).supportsNanoseconds().build(); + private static final IndexFeatures LUCENE_FEATURES = new IndexFeatures.Builder() + .supportedStringMappings(Mapping.TEXT, Mapping.STRING) + .supportsCardinality(Cardinality.SINGLE) + .supportsNanoseconds() + .supportsGeoContains() + .build(); /** * Default tree levels used when creating the prefix tree. diff --git a/janusgraph-solr/src/main/java/org/janusgraph/diskstorage/solr/SolrIndex.java b/janusgraph-solr/src/main/java/org/janusgraph/diskstorage/solr/SolrIndex.java index 3fd64380fa..b4dc3f3417 100644 --- a/janusgraph-solr/src/main/java/org/janusgraph/diskstorage/solr/SolrIndex.java +++ b/janusgraph-solr/src/main/java/org/janusgraph/diskstorage/solr/SolrIndex.java @@ -167,8 +167,14 @@ public static Mode parse(String mode) { ConfigOption.Type.LOCAL, false); - private static final IndexFeatures SOLR_FEATURES = new IndexFeatures.Builder().supportsDocumentTTL() - .setDefaultStringMapping(Mapping.TEXT).supportedStringMappings(Mapping.TEXT, Mapping.STRING).supportsCardinality(Cardinality.SINGLE).supportsCustomAnalyzer().build(); + private static final IndexFeatures SOLR_FEATURES = new IndexFeatures.Builder() + .supportsDocumentTTL() + .setDefaultStringMapping(Mapping.TEXT) + .supportedStringMappings(Mapping.TEXT, Mapping.STRING) + .supportsCardinality(Cardinality.SINGLE) + .supportsCustomAnalyzer() + .supportsGeoContains() + .build(); private static Map SPATIAL_PREDICATES = spatialPredicates(); diff --git a/janusgraph-test/src/main/java/org/janusgraph/diskstorage/indexing/IndexProviderTest.java b/janusgraph-test/src/main/java/org/janusgraph/diskstorage/indexing/IndexProviderTest.java index 7f0ef43cc5..185e085ce6 100644 --- a/janusgraph-test/src/main/java/org/janusgraph/diskstorage/indexing/IndexProviderTest.java +++ b/janusgraph-test/src/main/java/org/janusgraph/diskstorage/indexing/IndexProviderTest.java @@ -354,9 +354,11 @@ private void storeTest(String... stores) throws Exception { assertEquals(ImmutableSet.of("doc3"), ImmutableSet.copyOf(result)); } - result = tx.query(new IndexQuery(store, PredicateCondition.of(BOUNDARY, Geo.CONTAINS, Geoshape.point(47,10)))); - assertEquals(1, result.size()); - assertEquals(ImmutableSet.of("doc3"), ImmutableSet.copyOf(result)); + if (indexFeatures.supportsGeoContains()) { + result = tx.query(new IndexQuery(store, PredicateCondition.of(BOUNDARY, Geo.CONTAINS, Geoshape.point(47, 10)))); + assertEquals(1, result.size()); + assertEquals(ImmutableSet.of("doc3"), ImmutableSet.copyOf(result)); + } result = tx.query(new IndexQuery(store, PredicateCondition.of(BOUNDARY, Geo.INTERSECT, Geoshape.box(48,-1,49,2)))); assertEquals(2,result.size()); diff --git a/janusgraph-test/src/main/java/org/janusgraph/graphdb/JanusGraphIndexTest.java b/janusgraph-test/src/main/java/org/janusgraph/graphdb/JanusGraphIndexTest.java index 21038bbd6f..1a198fc997 100644 --- a/janusgraph-test/src/main/java/org/janusgraph/graphdb/JanusGraphIndexTest.java +++ b/janusgraph-test/src/main/java/org/janusgraph/graphdb/JanusGraphIndexTest.java @@ -1757,8 +1757,10 @@ private void testGeo(int i, int origNumV, int numV, String geoPointProperty, Str assertCount(numV-(i + 1), tx.query().has(geoShapeProperty, Geo.DISJOINT, Geoshape.circle(0.0, 0.0, distance)).vertices()); assertCount(numV-(i + 1), tx.query().has(geoShapeProperty, Geo.DISJOINT, Geoshape.circle(0.0, 0.0, distance)).edges()); - assertCount(i % 2, tx.query().has(geoShapeProperty, Geo.CONTAINS, Geoshape.point(-offset,-offset)).vertices()); - assertCount(i % 2, tx.query().has(geoShapeProperty, Geo.CONTAINS, Geoshape.point(-offset,-offset)).edges()); + if (indexFeatures.supportsGeoContains()) { + assertCount(i % 2, tx.query().has(geoShapeProperty, Geo.CONTAINS, Geoshape.point(-offset, -offset)).vertices()); + assertCount(i % 2, tx.query().has(geoShapeProperty, Geo.CONTAINS, Geoshape.point(-offset, -offset)).edges()); + } double buffer = bufferKm/111.; double min = -Math.abs(offset); diff --git a/pom.xml b/pom.xml index e8a06bda48..e0670daea1 100644 --- a/pom.xml +++ b/pom.xml @@ -87,9 +87,8 @@ the ES version, also consider the version of Lucene, and vice-versa. --> 5.5.2 - 2.4.4 5.2.2 - 5.2.2 + ${elasticsearch.rest.version} false