You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by dk...@apache.org on 2016/08/05 23:27:32 UTC
[2/9] incubator-atlas git commit: ATLAS-693 Titan 0.5.4
implementation of graph db abstraction. (jnhagelb via dkantor)
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/4fa10b6a/titan/src/main/java/com/thinkaurelius/titan/diskstorage/locking/LocalLockMediator.java
----------------------------------------------------------------------
diff --git a/titan/src/main/java/com/thinkaurelius/titan/diskstorage/locking/LocalLockMediator.java b/titan/src/main/java/com/thinkaurelius/titan/diskstorage/locking/LocalLockMediator.java
deleted file mode 100644
index 20c59e1..0000000
--- a/titan/src/main/java/com/thinkaurelius/titan/diskstorage/locking/LocalLockMediator.java
+++ /dev/null
@@ -1,345 +0,0 @@
-/*
- * Copyright 2012-2013 Aurelius LLC
- * 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 com.thinkaurelius.titan.diskstorage.locking;
-
-import com.google.common.base.Preconditions;
-import com.thinkaurelius.titan.diskstorage.util.time.Timepoint;
-import com.thinkaurelius.titan.diskstorage.util.time.TimestampProvider;
-import com.thinkaurelius.titan.diskstorage.locking.consistentkey.ExpectedValueCheckingTransaction;
-import com.thinkaurelius.titan.diskstorage.util.KeyColumn;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.Delayed;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-
-/**
- * This class resolves lock contention between two transactions on the same JVM.
- * <p/>
- * This is not just an optimization to reduce network traffic. Locks written by
- * Titan to a distributed key-value store contain an identifier, the "Rid",
- * which is unique only to the process level. The Rid can't tell which
- * transaction in a process holds any given lock. This class prevents two
- * transactions in a single process from concurrently writing the same lock to a
- * distributed key-value store.
- *
- * @author Dan LaRocque <da...@hopcount.org>
- */
-
-public class LocalLockMediator<T> {
-
- private static final Logger log = LoggerFactory
- .getLogger(LocalLockMediator.class);
-
- /**
- * Namespace for which this mediator is responsible
- *
- * @see LocalLockMediatorProvider
- */
- private final String name;
-
- private final TimestampProvider times;
-
- private DelayQueue<ExpirableKeyColumn> expiryQueue = new DelayQueue<>();
-
- private ExecutorService lockCleanerService = Executors.newFixedThreadPool(1, new ThreadFactory() {
- @Override
- public Thread newThread(Runnable runnable) {
- Thread thread = Executors.defaultThreadFactory().newThread(runnable);
- thread.setDaemon(true);
- return thread;
- }
- });
-
-
-
- /**
- * Maps a ({@code key}, {@code column}) pair to the local transaction
- * holding a lock on that pair. Values in this map may have already expired
- * according to {@link AuditRecord#expires}, in which case the lock should
- * be considered invalid.
- */
- private final ConcurrentHashMap<KeyColumn, AuditRecord<T>> locks = new ConcurrentHashMap<KeyColumn, AuditRecord<T>>();
-
- public LocalLockMediator(String name, TimestampProvider times) {
- this.name = name;
- this.times = times;
-
- Preconditions.checkNotNull(name);
- Preconditions.checkNotNull(times);
- lockCleanerService.submit(new LockCleaner());
- }
-
- /**
- * Acquire the lock specified by {@code kc}.
- * <p/>
- * <p/>
- * For any particular key-column, whatever value of {@code requestor} is
- * passed to this method must also be passed to the associated later call to
- * {@link #unlock(KeyColumn, ExpectedValueCheckingTransaction)}.
- * <p/>
- * If some requestor {@code r} calls this method on a KeyColumn {@code k}
- * and this method returns true, then subsequent calls to this method by
- * {@code r} on {@code l} merely attempt to update the {@code expiresAt}
- * timestamp. This differs from typical lock reentrance: multiple successful
- * calls to this method do not require an equal number of calls to
- * {@code #unlock()}. One {@code #unlock()} call is enough, no matter how
- * many times a {@code requestor} called {@code lock} beforehand. Note that
- * updating the timestamp may fail, in which case the lock is considered to
- * have expired and the calling context should assume it no longer holds the
- * lock specified by {@code kc}.
- * <p/>
- * The number of nanoseconds elapsed since the UNIX Epoch is not readily
- * available within the JVM. When reckoning expiration times, this method
- * uses the approximation implemented by
- * {@link com.thinkaurelius.titan.diskstorage.util.NanoTime#getApproxNSSinceEpoch(false)}.
- * <p/>
- * The current implementation of this method returns true when given an
- * {@code expiresAt} argument in the past. Future implementations may return
- * false instead.
- *
- * @param kc lock identifier
- * @param requestor the object locking {@code kc}
- * @param expires instant at which this lock will automatically expire
- * @return true if the lock is acquired, false if it was not acquired
- */
- public boolean lock(KeyColumn kc, T requestor, Timepoint expires) {
- assert null != kc;
- assert null != requestor;
-
- AuditRecord<T> audit = new AuditRecord<T>(requestor, expires);
- AuditRecord<T> inmap = locks.putIfAbsent(kc, audit);
-
- boolean success = false;
-
- if (null == inmap) {
- // Uncontended lock succeeded
- if (log.isTraceEnabled()) {
- log.trace("New local lock created: {} namespace={} txn={}",
- new Object[]{kc, name, requestor});
- }
- success = true;
- } else if (inmap.equals(audit)) {
- // requestor has already locked kc; update expiresAt
- success = locks.replace(kc, inmap, audit);
- if (log.isTraceEnabled()) {
- if (success) {
- log.trace(
- "Updated local lock expiration: {} namespace={} txn={} oldexp={} newexp={}",
- new Object[]{kc, name, requestor, inmap.expires,
- audit.expires});
- } else {
- log.trace(
- "Failed to update local lock expiration: {} namespace={} txn={} oldexp={} newexp={}",
- new Object[]{kc, name, requestor, inmap.expires,
- audit.expires});
- }
- }
- } else if (0 > inmap.expires.compareTo(times.getTime())) {
- // the recorded lock has expired; replace it
- success = locks.replace(kc, inmap, audit);
- if (log.isTraceEnabled()) {
- log.trace(
- "Discarding expired lock: {} namespace={} txn={} expired={}",
- new Object[]{kc, name, inmap.holder, inmap.expires});
- }
- } else {
- // we lost to a valid lock
- if (log.isTraceEnabled()) {
- log.trace(
- "Local lock failed: {} namespace={} txn={} (already owned by {})",
- new Object[]{kc, name, requestor, inmap});
- }
- }
-
- if (success) {
- expiryQueue.add(new ExpirableKeyColumn(kc, expires));
- }
- return success;
- }
-
- /**
- * Release the lock specified by {@code kc} and which was previously
- * locked by {@code requestor}, if it is possible to release it.
- *
- * @param kc lock identifier
- * @param requestor the object which previously locked {@code kc}
- */
- public boolean unlock(KeyColumn kc, T requestor) {
-
- if (!locks.containsKey(kc)) {
- log.info("Local unlock failed: no locks found for {}", kc);
- return false;
- }
-
- AuditRecord<T> unlocker = new AuditRecord<T>(requestor, null);
-
- AuditRecord<T> holder = locks.get(kc);
-
- if (!holder.equals(unlocker)) {
- log.error("Local unlock of {} by {} failed: it is held by {}",
- new Object[]{kc, unlocker, holder});
- return false;
- }
-
- boolean removed = locks.remove(kc, unlocker);
-
- if (removed) {
- expiryQueue.remove(kc);
- if (log.isTraceEnabled()) {
- log.trace("Local unlock succeeded: {} namespace={} txn={}",
- new Object[]{kc, name, requestor});
- }
- } else {
- log.warn("Local unlock warning: lock record for {} disappeared "
- + "during removal; this suggests the lock either expired "
- + "while we were removing it, or that it was erroneously "
- + "unlocked multiple times.", kc);
- }
-
- // Even if !removed, we're finished unlocking, so return true
- return true;
- }
-
- public String toString() {
- return "LocalLockMediator [" + name + ", ~" + locks.size()
- + " current locks]";
- }
-
- /**
- * A record containing the local transaction that holds a lock and the
- * lock's expiration time.
- */
- private static class AuditRecord<T> {
-
- /**
- * The local transaction that holds/held the lock.
- */
- private final T holder;
- /**
- * The expiration time of a the lock.
- */
- private final Timepoint expires;
- /**
- * Cached hashCode.
- */
- private int hashCode;
-
- private AuditRecord(T holder, Timepoint expires) {
- this.holder = holder;
- this.expires = expires;
- }
-
- /**
- * This implementation depends only on the lock holder and not on the
- * lock expiration time.
- */
- @Override
- public int hashCode() {
- if (0 == hashCode)
- hashCode = holder.hashCode();
-
- return hashCode;
- }
-
- /**
- * This implementation depends only on the lock holder and not on the
- * lock expiration time.
- */
- @Override
- public boolean equals(Object obj) {
- if (this == obj)
- return true;
- if (obj == null)
- return false;
- if (getClass() != obj.getClass())
- return false;
- /*
- * This warning suppression is harmless because we are only going to
- * call other.holder.equals(...), and since equals(...) is part of
- * Object, it is guaranteed to be defined no matter the concrete
- * type of parameter T.
- */
- @SuppressWarnings("rawtypes")
- AuditRecord other = (AuditRecord) obj;
- if (holder == null) {
- if (other.holder != null)
- return false;
- } else if (!holder.equals(other.holder))
- return false;
- return true;
- }
-
- @Override
- public String toString() {
- return "AuditRecord [txn=" + holder + ", expires=" + expires + "]";
- }
-
- }
-
- private class LockCleaner implements Runnable {
-
- @Override
- public void run() {
- try {
- while (true) {
- log.debug("Lock Cleaner service started");
- ExpirableKeyColumn lock = expiryQueue.take();
- log.debug("Expiring key column " + lock.getKeyColumn());
- locks.remove(lock.getKeyColumn());
- }
- } catch (InterruptedException e) {
- log.debug("Received interrupt. Exiting");
- }
- }
- }
-
- private static class ExpirableKeyColumn implements Delayed {
-
- private Timepoint expiryTime;
- private KeyColumn kc;
-
- public ExpirableKeyColumn(KeyColumn keyColumn, Timepoint expiryTime) {
- this.kc = keyColumn;
- this.expiryTime = expiryTime;
- }
-
- @Override
- public long getDelay(TimeUnit unit) {
- return expiryTime.getTimestamp(TimeUnit.NANOSECONDS);
- }
-
- @Override
- public int compareTo(Delayed o) {
- if (this.expiryTime.getTimestamp(TimeUnit.NANOSECONDS) < ((ExpirableKeyColumn) o).expiryTime.getTimestamp(TimeUnit.NANOSECONDS)) {
- return -1;
- }
- if (this.expiryTime.getTimestamp(TimeUnit.NANOSECONDS) > ((ExpirableKeyColumn) o).expiryTime.getTimestamp(TimeUnit.NANOSECONDS)) {
- return 1;
- }
- return 0;
- }
-
- public KeyColumn getKeyColumn() {
- return kc;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/4fa10b6a/titan/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java
----------------------------------------------------------------------
diff --git a/titan/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java b/titan/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java
deleted file mode 100644
index 3b5620c..0000000
--- a/titan/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java
+++ /dev/null
@@ -1,975 +0,0 @@
-/*
- * Copyright 2012-2013 Aurelius LLC
- * 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 com.thinkaurelius.titan.diskstorage.solr;
-
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Sets;
-import com.thinkaurelius.titan.core.Order;
-import com.thinkaurelius.titan.core.TitanElement;
-import com.thinkaurelius.titan.core.attribute.Cmp;
-import com.thinkaurelius.titan.core.attribute.Geo;
-import com.thinkaurelius.titan.core.attribute.Geoshape;
-import com.thinkaurelius.titan.core.attribute.Text;
-import com.thinkaurelius.titan.core.schema.Mapping;
-import com.thinkaurelius.titan.diskstorage.BackendException;
-import com.thinkaurelius.titan.diskstorage.BaseTransaction;
-import com.thinkaurelius.titan.diskstorage.BaseTransactionConfig;
-import com.thinkaurelius.titan.diskstorage.BaseTransactionConfigurable;
-import com.thinkaurelius.titan.diskstorage.PermanentBackendException;
-import com.thinkaurelius.titan.diskstorage.TemporaryBackendException;
-import com.thinkaurelius.titan.diskstorage.configuration.ConfigNamespace;
-import com.thinkaurelius.titan.diskstorage.configuration.ConfigOption;
-import com.thinkaurelius.titan.diskstorage.configuration.Configuration;
-import com.thinkaurelius.titan.diskstorage.indexing.IndexEntry;
-import com.thinkaurelius.titan.diskstorage.indexing.IndexFeatures;
-import com.thinkaurelius.titan.diskstorage.indexing.IndexMutation;
-import com.thinkaurelius.titan.diskstorage.indexing.IndexProvider;
-import com.thinkaurelius.titan.diskstorage.indexing.IndexQuery;
-import com.thinkaurelius.titan.diskstorage.indexing.KeyInformation;
-import com.thinkaurelius.titan.diskstorage.indexing.RawQuery;
-import com.thinkaurelius.titan.diskstorage.util.DefaultTransaction;
-import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration;
-import com.thinkaurelius.titan.graphdb.configuration.PreInitializeConfigOptions;
-import com.thinkaurelius.titan.graphdb.database.serialize.AttributeUtil;
-import com.thinkaurelius.titan.graphdb.database.serialize.attribute.AbstractDecimal;
-import com.thinkaurelius.titan.graphdb.query.TitanPredicate;
-import com.thinkaurelius.titan.graphdb.query.condition.And;
-import com.thinkaurelius.titan.graphdb.query.condition.Condition;
-import com.thinkaurelius.titan.graphdb.query.condition.Not;
-import com.thinkaurelius.titan.graphdb.query.condition.Or;
-import com.thinkaurelius.titan.graphdb.query.condition.PredicateCondition;
-import com.thinkaurelius.titan.graphdb.types.ParameterType;
-import org.apache.commons.lang.StringUtils;
-import org.apache.http.client.HttpClient;
-import org.apache.solr.client.solrj.SolrClient;
-import org.apache.solr.client.solrj.SolrQuery;
-import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.impl.HttpClientUtil;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
-import org.apache.solr.client.solrj.impl.Krb5HttpClientConfigurer;
-import org.apache.solr.client.solrj.impl.LBHttpSolrClient;
-import org.apache.solr.client.solrj.request.CollectionAdminRequest;
-import org.apache.solr.client.solrj.request.UpdateRequest;
-import org.apache.solr.client.solrj.response.CollectionAdminResponse;
-import org.apache.solr.client.solrj.response.QueryResponse;
-import org.apache.solr.client.solrj.util.ClientUtils;
-import org.apache.solr.common.SolrDocument;
-import org.apache.solr.common.SolrInputDocument;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.Slice;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.zookeeper.KeeperException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.TimeZone;
-import java.util.UUID;
-
-import static com.thinkaurelius.titan.core.attribute.Cmp.*;
-import static com.thinkaurelius.titan.core.schema.Mapping.*;
-
-/**
- * NOTE: Copied from titan for supporting sol5. Do not change
- */
-@PreInitializeConfigOptions
-public class Solr5Index implements IndexProvider {
-
- private static final Logger logger = LoggerFactory.getLogger(Solr5Index.class);
-
-
- private static final String DEFAULT_ID_FIELD = "id";
-
- private enum Mode {
- HTTP, CLOUD;
-
- public static Mode parse(String mode) {
- for (Mode m : Mode.values()) {
- if (m.toString().equalsIgnoreCase(mode)) return m;
- }
- throw new IllegalArgumentException("Unrecognized mode: "+mode);
- }
- }
-
- public static final ConfigNamespace SOLR_NS =
- new ConfigNamespace(GraphDatabaseConfiguration.INDEX_NS, "solr", "Solr index configuration");
-
- public static final ConfigOption<String> SOLR_MODE = new ConfigOption<String>(SOLR_NS,"mode",
- "The operation mode for Solr which is either via HTTP (`http`) or using SolrCloud (`cloud`)",
- ConfigOption.Type.GLOBAL_OFFLINE, "cloud");
-
- public static final ConfigOption<Boolean> DYNAMIC_FIELDS = new ConfigOption<Boolean>(SOLR_NS,"dyn-fields",
- "Whether to use dynamic fields (which appends the data type to the field name). If dynamic fields is disabled" +
- "the user must map field names and define them explicitly in the schema.",
- ConfigOption.Type.GLOBAL_OFFLINE, true);
-
- public static final ConfigOption<String[]> KEY_FIELD_NAMES = new ConfigOption<String[]>(SOLR_NS,"key-field-names",
- "Field name that uniquely identifies each document in Solr. Must be specified as a list of `collection=field`.",
- ConfigOption.Type.GLOBAL, String[].class);
-
- public static final ConfigOption<String> TTL_FIELD = new ConfigOption<String>(SOLR_NS,"ttl_field",
- "Name of the TTL field for Solr collections.",
- ConfigOption.Type.GLOBAL_OFFLINE, "ttl");
-
- /** SolrCloud Configuration */
-
- public static final ConfigOption<String> ZOOKEEPER_URL = new ConfigOption<String>(SOLR_NS,"zookeeper-url",
- "URL of the Zookeeper instance coordinating the SolrCloud cluster",
- ConfigOption.Type.MASKABLE, "localhost:2181");
-
- public static final ConfigOption<Integer> NUM_SHARDS = new ConfigOption<Integer>(SOLR_NS,"num-shards",
- "Number of shards for a collection. This applies when creating a new collection which is only supported under the SolrCloud operation mode.",
- ConfigOption.Type.GLOBAL_OFFLINE, 1);
-
- public static final ConfigOption<Integer> MAX_SHARDS_PER_NODE = new ConfigOption<Integer>(SOLR_NS,"max-shards-per-node",
- "Maximum number of shards per node. This applies when creating a new collection which is only supported under the SolrCloud operation mode.",
- ConfigOption.Type.GLOBAL_OFFLINE, 1);
-
- public static final ConfigOption<Integer> REPLICATION_FACTOR = new ConfigOption<Integer>(SOLR_NS,"replication-factor",
- "Replication factor for a collection. This applies when creating a new collection which is only supported under the SolrCloud operation mode.",
- ConfigOption.Type.GLOBAL_OFFLINE, 1);
-
-
- /** HTTP Configuration */
-
- public static final ConfigOption<String[]> HTTP_URLS = new ConfigOption<String[]>(SOLR_NS,"http-urls",
- "List of URLs to use to connect to Solr Servers (LBHttpSolrClient is used), don't add core or collection name to the URL.",
- ConfigOption.Type.MASKABLE, new String[] { "http://localhost:8983/solr" });
-
- public static final ConfigOption<Integer> HTTP_CONNECTION_TIMEOUT = new ConfigOption<Integer>(SOLR_NS,"http-connection-timeout",
- "Solr HTTP connection timeout.",
- ConfigOption.Type.MASKABLE, 5000);
-
- public static final ConfigOption<Boolean> HTTP_ALLOW_COMPRESSION = new ConfigOption<Boolean>(SOLR_NS,"http-compression",
- "Enable/disable compression on the HTTP connections made to Solr.",
- ConfigOption.Type.MASKABLE, false);
-
- public static final ConfigOption<Integer> HTTP_MAX_CONNECTIONS_PER_HOST = new ConfigOption<Integer>(SOLR_NS,"http-max-per-host",
- "Maximum number of HTTP connections per Solr host.",
- ConfigOption.Type.MASKABLE, 20);
-
- public static final ConfigOption<Integer> HTTP_GLOBAL_MAX_CONNECTIONS = new ConfigOption<Integer>(SOLR_NS,"http-max",
- "Maximum number of HTTP connections in total to all Solr servers.",
- ConfigOption.Type.MASKABLE, 100);
-
- public static final ConfigOption<Boolean> WAIT_SEARCHER = new ConfigOption<Boolean>(SOLR_NS, "wait-searcher",
- "When mutating - wait for the index to reflect new mutations before returning. This can have a negative impact on performance.",
- ConfigOption.Type.LOCAL, false);
-
-
-
- private static final IndexFeatures SOLR_FEATURES = new IndexFeatures.Builder().supportsDocumentTTL()
- .setDefaultStringMapping(TEXT).supportedStringMappings(TEXT, STRING).build();
-
- private final SolrClient solrClient;
- private final Configuration configuration;
- private final Mode mode;
- private final boolean dynFields;
- private final Map<String, String> keyFieldIds;
- private final String ttlField;
- private final int maxResults;
- private final boolean waitSearcher;
-
- public Solr5Index(final Configuration config) throws BackendException {
- Preconditions.checkArgument(config!=null);
- configuration = config;
-
- mode = Mode.parse(config.get(SOLR_MODE));
- dynFields = config.get(DYNAMIC_FIELDS);
- keyFieldIds = parseKeyFieldsForCollections(config);
- maxResults = config.get(GraphDatabaseConfiguration.INDEX_MAX_RESULT_SET_SIZE);
- ttlField = config.get(TTL_FIELD);
- waitSearcher = config.get(WAIT_SEARCHER);
-
- if (mode==Mode.CLOUD) {
- HttpClientUtil.setConfigurer(new Krb5HttpClientConfigurer());
- String zookeeperUrl = config.get(Solr5Index.ZOOKEEPER_URL);
- CloudSolrClient cloudServer = new CloudSolrClient(zookeeperUrl, true);
- cloudServer.connect();
- solrClient = cloudServer;
- } else if (mode==Mode.HTTP) {
- HttpClientUtil.setConfigurer(new Krb5HttpClientConfigurer());
- HttpClient clientParams = HttpClientUtil.createClient(new ModifiableSolrParams() {{
- add(HttpClientUtil.PROP_ALLOW_COMPRESSION, config.get(HTTP_ALLOW_COMPRESSION).toString());
- add(HttpClientUtil.PROP_CONNECTION_TIMEOUT, config.get(HTTP_CONNECTION_TIMEOUT).toString());
- add(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, config.get(HTTP_MAX_CONNECTIONS_PER_HOST).toString());
- add(HttpClientUtil.PROP_MAX_CONNECTIONS, config.get(HTTP_GLOBAL_MAX_CONNECTIONS).toString());
- }});
-
- solrClient = new LBHttpSolrClient(clientParams, config.get(HTTP_URLS));
-
-
- } else {
- throw new IllegalArgumentException("Unsupported Solr operation mode: " + mode);
- }
- }
-
- private Map<String, String> parseKeyFieldsForCollections(Configuration config) throws BackendException {
- Map<String, String> keyFieldNames = new HashMap<String, String>();
- String[] collectionFieldStatements = config.has(KEY_FIELD_NAMES)?config.get(KEY_FIELD_NAMES):new String[0];
- for (String collectionFieldStatement : collectionFieldStatements) {
- String[] parts = collectionFieldStatement.trim().split("=");
- if (parts.length != 2) {
- throw new PermanentBackendException("Unable to parse the collection name / key field name pair. It should be of the format collection=field");
- }
- String collectionName = parts[0];
- String keyFieldName = parts[1];
- keyFieldNames.put(collectionName, keyFieldName);
- }
- return keyFieldNames;
- }
-
- private String getKeyFieldId(String collection) {
- String field = keyFieldIds.get(collection);
- if (field==null) field = DEFAULT_ID_FIELD;
- return field;
- }
-
- /**
- * Unlike the ElasticSearch Index, which is schema free, Solr requires a schema to
- * support searching. This means that you will need to modify the solr schema with the
- * appropriate field definitions in order to work properly. If you have a running instance
- * of Solr and you modify its schema with new fields, don't forget to re-index!
- * @param store Index store
- * @param key New key to register
- * @param information Datatype to register for the key
- * @param tx enclosing transaction
- * @throws com.thinkaurelius.titan.diskstorage.BackendException
- */
- @Override
- public void register(String store, String key, KeyInformation information, BaseTransaction tx) throws BackendException {
- if (mode==Mode.CLOUD) {
- CloudSolrClient client = (CloudSolrClient) solrClient;
- try {
- createCollectionIfNotExists(client, configuration, store);
- } catch (IOException e) {
- throw new PermanentBackendException(e);
- } catch (SolrServerException e) {
- throw new PermanentBackendException(e);
- } catch (InterruptedException e) {
- throw new PermanentBackendException(e);
- } catch (KeeperException e) {
- throw new PermanentBackendException(e);
- }
- }
- //Since all data types must be defined in the schema.xml, pre-registering a type does not work
- }
-
- @Override
- public void mutate(Map<String, Map<String, IndexMutation>> mutations, KeyInformation.IndexRetriever informations, BaseTransaction tx) throws BackendException {
- logger.debug("Mutating SOLR");
- try {
- for (Map.Entry<String, Map<String, IndexMutation>> stores : mutations.entrySet()) {
- String collectionName = stores.getKey();
- String keyIdField = getKeyFieldId(collectionName);
-
- List<String> deleteIds = new ArrayList<String>();
- Collection<SolrInputDocument> changes = new ArrayList<SolrInputDocument>();
-
- for (Map.Entry<String, IndexMutation> entry : stores.getValue().entrySet()) {
- String docId = entry.getKey();
- IndexMutation mutation = entry.getValue();
- Preconditions.checkArgument(!(mutation.isNew() && mutation.isDeleted()));
- Preconditions.checkArgument(!mutation.isNew() || !mutation.hasDeletions());
- Preconditions.checkArgument(!mutation.isDeleted() || !mutation.hasAdditions());
-
- //Handle any deletions
- if (mutation.hasDeletions()) {
- if (mutation.isDeleted()) {
- logger.trace("Deleting entire document {}", docId);
- deleteIds.add(docId);
- } else {
- HashSet<IndexEntry> fieldDeletions = Sets.newHashSet(mutation.getDeletions());
- if (mutation.hasAdditions()) {
- for (IndexEntry indexEntry : mutation.getAdditions()) {
- fieldDeletions.remove(indexEntry);
- }
- }
- deleteIndividualFieldsFromIndex(collectionName, keyIdField, docId, fieldDeletions);
- }
- }
-
- if (mutation.hasAdditions()) {
- int ttl = mutation.determineTTL();
-
- SolrInputDocument doc = new SolrInputDocument();
- doc.setField(keyIdField, docId);
-
- boolean isNewDoc = mutation.isNew();
-
- if (isNewDoc)
- logger.trace("Adding new document {}", docId);
-
- for (IndexEntry e : mutation.getAdditions()) {
- final Object fieldValue = convertValue(e.value);
- doc.setField(e.field, isNewDoc
- ? fieldValue : new HashMap<String, Object>(1) {{ put("set", fieldValue); }});
- }
- if (ttl>0) {
- Preconditions.checkArgument(isNewDoc,"Solr only supports TTL on new documents [%s]",docId);
- doc.setField(ttlField, String.format("+%dSECONDS", ttl));
- }
- changes.add(doc);
- }
- }
-
- commitDeletes(collectionName, deleteIds);
- commitDocumentChanges(collectionName, changes);
- }
- } catch (Exception e) {
- throw storageException(e);
- }
- }
-
- private Object convertValue(Object value) throws BackendException {
- if (value instanceof Geoshape)
- return GeoToWktConverter.convertToWktString((Geoshape) value);
- // in order to serialize/deserialize properly Solr will have to have an
- // access to Titan source which has Decimal type, so for now we simply convert to
- // double and let Solr do the same thing or fail.
- if (value instanceof AbstractDecimal)
- return ((AbstractDecimal) value).doubleValue();
- if (value instanceof UUID)
- return value.toString();
- return value;
- }
-
- @Override
- public void restore(Map<String, Map<String, List<IndexEntry>>> documents, KeyInformation.IndexRetriever informations, BaseTransaction tx) throws BackendException {
- try {
- for (Map.Entry<String, Map<String, List<IndexEntry>>> stores : documents.entrySet()) {
- final String collectionName = stores.getKey();
-
- List<String> deleteIds = new ArrayList<String>();
- List<SolrInputDocument> newDocuments = new ArrayList<SolrInputDocument>();
-
- for (Map.Entry<String, List<IndexEntry>> entry : stores.getValue().entrySet()) {
- final String docID = entry.getKey();
- final List<IndexEntry> content = entry.getValue();
-
- if (content == null || content.isEmpty()) {
- if (logger.isTraceEnabled())
- logger.trace("Deleting document [{}]", docID);
-
- deleteIds.add(docID);
- continue;
- }
-
- newDocuments.add(new SolrInputDocument() {{
- setField(getKeyFieldId(collectionName), docID);
-
- for (IndexEntry addition : content) {
- Object fieldValue = addition.value;
- setField(addition.field, convertValue(fieldValue));
- }
- }});
- }
-
- commitDeletes(collectionName, deleteIds);
- commitDocumentChanges(collectionName, newDocuments);
- }
- } catch (Exception e) {
- throw new TemporaryBackendException("Could not restore Solr index", e);
- }
- }
-
- private void deleteIndividualFieldsFromIndex(String collectionName, String keyIdField, String docId, HashSet<IndexEntry> fieldDeletions) throws SolrServerException, IOException {
- if (fieldDeletions.isEmpty()) return;
-
- Map<String, String> fieldDeletes = new HashMap<String, String>(1) {{ put("set", null); }};
-
- SolrInputDocument doc = new SolrInputDocument();
- doc.addField(keyIdField, docId);
- StringBuilder sb = new StringBuilder();
- for (IndexEntry fieldToDelete : fieldDeletions) {
- doc.addField(fieldToDelete.field, fieldDeletes);
- sb.append(fieldToDelete).append(",");
- }
-
- if (logger.isTraceEnabled())
- logger.trace("Deleting individual fields [{}] for document {}", sb.toString(), docId);
-
- UpdateRequest singleDocument = newUpdateRequest();
- singleDocument.add(doc);
- solrClient.request(singleDocument, collectionName);
- }
-
- private void commitDocumentChanges(String collectionName, Collection<SolrInputDocument> documents) throws SolrServerException, IOException {
- if (documents.size() == 0) return;
-
- try {
- solrClient.request(newUpdateRequest().add(documents), collectionName);
- } catch (HttpSolrClient.RemoteSolrException rse) {
- logger.error("Unable to save documents to Solr as one of the shape objects stored were not compatible with Solr.", rse);
- logger.error("Details in failed document batch: ");
- for (SolrInputDocument d : documents) {
- Collection<String> fieldNames = d.getFieldNames();
- for (String name : fieldNames) {
- logger.error(name + ":" + d.getFieldValue(name).toString());
- }
- }
-
- throw rse;
- }
- }
-
- private void commitDeletes(String collectionName, List<String> deleteIds) throws SolrServerException, IOException {
- if (deleteIds.size() == 0) return;
- solrClient.request(newUpdateRequest().deleteById(deleteIds), collectionName);
- }
-
- @Override
- public List<String> query(IndexQuery query, KeyInformation.IndexRetriever informations, BaseTransaction tx) throws BackendException {
- List<String> result;
- String collection = query.getStore();
- String keyIdField = getKeyFieldId(collection);
- SolrQuery solrQuery = new SolrQuery("*:*");
- String queryFilter = buildQueryFilter(query.getCondition(), informations.get(collection));
- solrQuery.addFilterQuery(queryFilter);
- if (!query.getOrder().isEmpty()) {
- List<IndexQuery.OrderEntry> orders = query.getOrder();
- for (IndexQuery.OrderEntry order1 : orders) {
- String item = order1.getKey();
- SolrQuery.ORDER order = order1.getOrder() == Order.ASC ? SolrQuery.ORDER.asc : SolrQuery.ORDER.desc;
- solrQuery.addSort(new SolrQuery.SortClause(item, order));
- }
- }
- solrQuery.setStart(0);
- if (query.hasLimit()) {
- solrQuery.setRows(query.getLimit());
- } else {
- solrQuery.setRows(maxResults);
- }
- try {
- QueryResponse response = solrClient.query(collection, solrQuery);
-
- if (logger.isDebugEnabled())
- logger.debug("Executed query [{}] in {} ms", query.getCondition(), response.getElapsedTime());
-
- int totalHits = response.getResults().size();
-
- if (!query.hasLimit() && totalHits >= maxResults)
- logger.warn("Query result set truncated to first [{}] elements for query: {}", maxResults, query);
-
- result = new ArrayList<String>(totalHits);
- for (SolrDocument hit : response.getResults()) {
- result.add(hit.getFieldValue(keyIdField).toString());
- }
- } catch (IOException e) {
- logger.error("Query did not complete : ", e);
- throw new PermanentBackendException(e);
- } catch (SolrServerException e) {
- logger.error("Unable to query Solr index.", e);
- throw new PermanentBackendException(e);
- }
- return result;
- }
-
- @Override
- public Iterable<RawQuery.Result<String>> query(RawQuery query, KeyInformation.IndexRetriever informations, BaseTransaction tx) throws BackendException {
- List<RawQuery.Result<String>> result;
- String collection = query.getStore();
- String keyIdField = getKeyFieldId(collection);
- SolrQuery solrQuery = new SolrQuery(query.getQuery())
- .addField(keyIdField)
- .setIncludeScore(true)
- .setStart(query.getOffset())
- .setRows(query.hasLimit() ? query.getLimit() : maxResults);
-
- try {
- QueryResponse response = solrClient.query(collection, solrQuery);
- if (logger.isDebugEnabled())
- logger.debug("Executed query [{}] in {} ms", query.getQuery(), response.getElapsedTime());
-
- int totalHits = response.getResults().size();
- if (!query.hasLimit() && totalHits >= maxResults) {
- logger.warn("Query result set truncated to first [{}] elements for query: {}", maxResults, query);
- }
- result = new ArrayList<RawQuery.Result<String>>(totalHits);
-
- for (SolrDocument hit : response.getResults()) {
- double score = Double.parseDouble(hit.getFieldValue("score").toString());
- result.add(new RawQuery.Result<String>(hit.getFieldValue(keyIdField).toString(), score));
- }
- } catch (IOException e) {
- logger.error("Query did not complete : ", e);
- throw new PermanentBackendException(e);
- } catch (SolrServerException e) {
- logger.error("Unable to query Solr index.", e);
- throw new PermanentBackendException(e);
- }
- return result;
- }
-
- private static String escapeValue(Object value) {
- return ClientUtils.escapeQueryChars(value.toString());
- }
-
- public String buildQueryFilter(Condition<TitanElement> condition, KeyInformation.StoreRetriever informations) {
- if (condition instanceof PredicateCondition) {
- PredicateCondition<String, TitanElement> atom = (PredicateCondition<String, TitanElement>) condition;
- Object value = atom.getValue();
- String key = atom.getKey();
- TitanPredicate titanPredicate = atom.getPredicate();
-
- if (value instanceof Number) {
- String queryValue = escapeValue(value);
- Preconditions.checkArgument(titanPredicate instanceof Cmp, "Relation not supported on numeric types: " + titanPredicate);
- Cmp numRel = (Cmp) titanPredicate;
- switch (numRel) {
- case EQUAL:
- return (key + ":" + queryValue);
- case NOT_EQUAL:
- return ("-" + key + ":" + queryValue);
- case LESS_THAN:
- //use right curly to mean up to but not including value
- return (key + ":[* TO " + queryValue + "}");
- case LESS_THAN_EQUAL:
- return (key + ":[* TO " + queryValue + "]");
- case GREATER_THAN:
- //use left curly to mean greater than but not including value
- return (key + ":{" + queryValue + " TO *]");
- case GREATER_THAN_EQUAL:
- return (key + ":[" + queryValue + " TO *]");
- default: throw new IllegalArgumentException("Unexpected relation: " + numRel);
- }
- } else if (value instanceof String) {
- Mapping map = getStringMapping(informations.get(key));
- assert map== TEXT || map== STRING;
- if (map== TEXT && !titanPredicate.toString().startsWith("CONTAINS"))
- throw new IllegalArgumentException("Text mapped string values only support CONTAINS queries and not: " + titanPredicate);
- if (map== STRING && titanPredicate.toString().startsWith("CONTAINS"))
- throw new IllegalArgumentException("String mapped string values do not support CONTAINS queries: " + titanPredicate);
-
- //Special case
- if (titanPredicate == Text.CONTAINS) {
- //e.g. - if terms tomorrow and world were supplied, and fq=text:(tomorrow world)
- //sample data set would return 2 documents: one where text = Tomorrow is the World,
- //and the second where text = Hello World. Hence, we are decomposing the query string
- //and building an AND query explicitly because we need AND semantics
- value = ((String) value).toLowerCase();
- List<String> terms = Text.tokenize((String) value);
-
- if (terms.isEmpty()) {
- return "";
- } else if (terms.size() == 1) {
- return (key + ":(" + escapeValue(terms.get(0)) + ")");
- } else {
- And<TitanElement> andTerms = new And<TitanElement>();
- for (String term : terms) {
- andTerms.add(new PredicateCondition<String, TitanElement>(key, titanPredicate, term));
- }
- return buildQueryFilter(andTerms, informations);
- }
- }
- if (titanPredicate == Text.PREFIX || titanPredicate == Text.CONTAINS_PREFIX) {
- return (key + ":" + escapeValue(value) + "*");
- } else if (titanPredicate == Text.REGEX || titanPredicate == Text.CONTAINS_REGEX) {
- return (key + ":/" + value + "/");
- } else if (titanPredicate == EQUAL) {
- return (key + ":\"" + escapeValue(value) + "\"");
- } else if (titanPredicate == NOT_EQUAL) {
- return ("-" + key + ":\"" + escapeValue(value) + "\"");
- } else {
- throw new IllegalArgumentException("Relation is not supported for string value: " + titanPredicate);
- }
- } else if (value instanceof Geoshape) {
- Geoshape geo = (Geoshape)value;
- if (geo.getType() == Geoshape.Type.CIRCLE) {
- Geoshape.Point center = geo.getPoint();
- return ("{!geofilt sfield=" + key +
- " pt=" + center.getLatitude() + "," + center.getLongitude() +
- " d=" + geo.getRadius() + "} distErrPct=0"); //distance in kilometers
- } else if (geo.getType() == Geoshape.Type.BOX) {
- Geoshape.Point southwest = geo.getPoint(0);
- Geoshape.Point northeast = geo.getPoint(1);
- return (key + ":[" + southwest.getLatitude() + "," + southwest.getLongitude() +
- " TO " + northeast.getLatitude() + "," + northeast.getLongitude() + "]");
- } else if (geo.getType() == Geoshape.Type.POLYGON) {
- List<Geoshape.Point> coordinates = getPolygonPoints(geo);
- StringBuilder poly = new StringBuilder(key + ":\"IsWithin(POLYGON((");
- for (Geoshape.Point coordinate : coordinates) {
- poly.append(coordinate.getLongitude()).append(" ").append(coordinate.getLatitude()).append(", ");
- }
- //close the polygon with the first coordinate
- poly.append(coordinates.get(0).getLongitude()).append(" ").append(coordinates.get(0).getLatitude());
- poly.append(")))\" distErrPct=0");
- return (poly.toString());
- }
- } else if (value instanceof Date) {
- String queryValue = escapeValue(toIsoDate((Date)value));
- Preconditions.checkArgument(titanPredicate instanceof Cmp, "Relation not supported on date types: " + titanPredicate);
- Cmp numRel = (Cmp) titanPredicate;
-
- switch (numRel) {
- case EQUAL:
- return (key + ":" + queryValue);
- case NOT_EQUAL:
- return ("-" + key + ":" + queryValue);
- case LESS_THAN:
- //use right curly to mean up to but not including value
- return (key + ":[* TO " + queryValue + "}");
- case LESS_THAN_EQUAL:
- return (key + ":[* TO " + queryValue + "]");
- case GREATER_THAN:
- //use left curly to mean greater than but not including value
- return (key + ":{" + queryValue + " TO *]");
- case GREATER_THAN_EQUAL:
- return (key + ":[" + queryValue + " TO *]");
- default: throw new IllegalArgumentException("Unexpected relation: " + numRel);
- }
- } else if (value instanceof Boolean) {
- Cmp numRel = (Cmp) titanPredicate;
- String queryValue = escapeValue(value);
- switch (numRel) {
- case EQUAL:
- return (key + ":" + queryValue);
- case NOT_EQUAL:
- return ("-" + key + ":" + queryValue);
- default:
- throw new IllegalArgumentException("Boolean types only support EQUAL or NOT_EQUAL");
- }
- } else if (value instanceof UUID) {
- if (titanPredicate == EQUAL) {
- return (key + ":\"" + escapeValue(value) + "\"");
- } else if (titanPredicate == NOT_EQUAL) {
- return ("-" + key + ":\"" + escapeValue(value) + "\"");
- } else {
- throw new IllegalArgumentException("Relation is not supported for uuid value: " + titanPredicate);
- }
- } else throw new IllegalArgumentException("Unsupported type: " + value);
- } else if (condition instanceof Not) {
- String sub = buildQueryFilter(((Not)condition).getChild(),informations);
- if (StringUtils.isNotBlank(sub)) return "-("+sub+")";
- else return "";
- } else if (condition instanceof And) {
- int numChildren = ((And) condition).size();
- StringBuilder sb = new StringBuilder();
- for (Condition<TitanElement> c : condition.getChildren()) {
- String sub = buildQueryFilter(c, informations);
-
- if (StringUtils.isBlank(sub))
- continue;
-
- // we don't have to add "+" which means AND iff
- // a. it's a NOT query,
- // b. expression is a single statement in the AND.
- if (!sub.startsWith("-") && numChildren > 1)
- sb.append("+");
-
- sb.append(sub).append(" ");
- }
- return sb.toString();
- } else if (condition instanceof Or) {
- StringBuilder sb = new StringBuilder();
- int element=0;
- for (Condition<TitanElement> c : condition.getChildren()) {
- String sub = buildQueryFilter(c,informations);
- if (StringUtils.isBlank(sub)) continue;
- if (element==0) sb.append("(");
- else sb.append(" OR ");
- sb.append(sub);
- element++;
- }
- if (element>0) sb.append(")");
- return sb.toString();
- } else {
- throw new IllegalArgumentException("Invalid condition: " + condition);
- }
- return null;
- }
-
- private String toIsoDate(Date value) {
- TimeZone tz = TimeZone.getTimeZone("UTC");
- DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
- df.setTimeZone(tz);
- return df.format(value);
- }
-
- private List<Geoshape.Point> getPolygonPoints(Geoshape polygon) {
- List<Geoshape.Point> locations = new ArrayList<Geoshape.Point>();
-
- int index = 0;
- boolean hasCoordinates = true;
- while (hasCoordinates) {
- try {
- locations.add(polygon.getPoint(index));
- } catch (ArrayIndexOutOfBoundsException ignore) {
- //just means we asked for a point past the size of the list
- //of known coordinates
- hasCoordinates = false;
- }
- }
-
- return locations;
- }
-
- /**
- * Solr handles all transactions on the server-side. That means all
- * commit, optimize, or rollback applies since the last commit/optimize/rollback.
- * Solr documentation recommends best way to update Solr is in one process to avoid
- * race conditions.
- *
- * @return New Transaction Handle
- * @throws com.thinkaurelius.titan.diskstorage.BackendException
- */
- @Override
- public BaseTransactionConfigurable beginTransaction(BaseTransactionConfig config) throws BackendException {
- return new DefaultTransaction(config);
- }
-
- @Override
- public void close() throws BackendException {
- logger.trace("Shutting down connection to Solr", solrClient);
- try {
- solrClient.close();
- } catch (IOException e) {
- throw new TemporaryBackendException(e);
- }
- }
-
- @Override
- public void clearStorage() throws BackendException {
- try {
- if (mode!=Mode.CLOUD) throw new UnsupportedOperationException("Operation only supported for SolrCloud");
- logger.debug("Clearing storage from Solr: {}", solrClient);
- ZkStateReader zkStateReader = ((CloudSolrClient) solrClient).getZkStateReader();
- zkStateReader.updateClusterState();
- ClusterState clusterState = zkStateReader.getClusterState();
- for (String collection : clusterState.getCollections()) {
- logger.debug("Clearing collection [{}] in Solr",collection);
- UpdateRequest deleteAll = newUpdateRequest();
- deleteAll.deleteByQuery("*:*");
- solrClient.request(deleteAll, collection);
- }
-
- } catch (SolrServerException e) {
- logger.error("Unable to clear storage from index due to server error on Solr.", e);
- throw new PermanentBackendException(e);
- } catch (IOException e) {
- logger.error("Unable to clear storage from index due to low-level I/O error.", e);
- throw new PermanentBackendException(e);
- } catch (Exception e) {
- logger.error("Unable to clear storage from index due to general error.", e);
- throw new PermanentBackendException(e);
- }
- }
-
- @Override
- public boolean supports(KeyInformation information, TitanPredicate titanPredicate) {
- Class<?> dataType = information.getDataType();
- Mapping mapping = getMapping(information);
- if (mapping!= DEFAULT && !AttributeUtil.isString(dataType)) return false;
-
- if (Number.class.isAssignableFrom(dataType)) {
- return titanPredicate instanceof Cmp;
- } else if (dataType == Geoshape.class) {
- return titanPredicate == Geo.WITHIN;
- } else if (AttributeUtil.isString(dataType)) {
- switch(mapping) {
- case DEFAULT:
- case TEXT:
- return titanPredicate == Text.CONTAINS || titanPredicate == Text.CONTAINS_PREFIX || titanPredicate == Text.CONTAINS_REGEX;
- case STRING:
- return titanPredicate == EQUAL || titanPredicate== NOT_EQUAL || titanPredicate==Text.REGEX || titanPredicate==Text.PREFIX;
- // case TEXTSTRING:
- // return (titanPredicate instanceof Text) || titanPredicate == Cmp.EQUAL || titanPredicate==Cmp.NOT_EQUAL;
- }
- } else if (dataType == Date.class) {
- if (titanPredicate instanceof Cmp) return true;
- } else if (dataType == Boolean.class) {
- return titanPredicate == EQUAL || titanPredicate == NOT_EQUAL;
- } else if (dataType == UUID.class) {
- return titanPredicate == EQUAL || titanPredicate== NOT_EQUAL;
- }
- return false;
- }
-
- @Override
- public boolean supports(KeyInformation information) {
- Class<?> dataType = information.getDataType();
- Mapping mapping = getMapping(information);
- if (Number.class.isAssignableFrom(dataType) || dataType == Geoshape.class || dataType == Date.class || dataType == Boolean.class || dataType == UUID.class) {
- if (mapping== DEFAULT) return true;
- } else if (AttributeUtil.isString(dataType)) {
- if (mapping== DEFAULT || mapping== TEXT || mapping== STRING) return true;
- }
- return false;
- }
-
- @Override
- public String mapKey2Field(String key, KeyInformation keyInfo) {
- Preconditions.checkArgument(!StringUtils.containsAny(key, new char[]{' '}),"Invalid key name provided: %s",key);
- if (!dynFields) return key;
- if (ParameterType.MAPPED_NAME.hasParameter(keyInfo.getParameters())) return key;
- String postfix;
- Class datatype = keyInfo.getDataType();
- if (AttributeUtil.isString(datatype)) {
- Mapping map = getStringMapping(keyInfo);
- switch (map) {
- case TEXT: postfix = "_t"; break;
- case STRING: postfix = "_s"; break;
- default: throw new IllegalArgumentException("Unsupported string mapping: " + map);
- }
- } else if (AttributeUtil.isWholeNumber(datatype)) {
- if (datatype.equals(Long.class)) postfix = "_l";
- else postfix = "_i";
- } else if (AttributeUtil.isDecimal(datatype)) {
- if (datatype.equals(Float.class)) postfix = "_f";
- else postfix = "_d";
- } else if (datatype.equals(Geoshape.class)) {
- postfix = "_g";
- } else if (datatype.equals(Date.class)) {
- postfix = "_dt";
- } else if (datatype.equals(Boolean.class)) {
- postfix = "_b";
- } else if (datatype.equals(UUID.class)) {
- postfix = "_uuid";
- } else throw new IllegalArgumentException("Unsupported data type ["+datatype+"] for field: " + key);
- return key+postfix;
- }
-
- @Override
- public IndexFeatures getFeatures() {
- return SOLR_FEATURES;
- }
-
- /*
- ################# UTILITY METHODS #######################
- */
-
- private static Mapping getStringMapping(KeyInformation information) {
- assert AttributeUtil.isString(information.getDataType());
- Mapping map = getMapping(information);
- if (map== DEFAULT) map = TEXT;
- return map;
- }
-
- private UpdateRequest newUpdateRequest() {
- UpdateRequest req = new UpdateRequest();
- req.setAction(UpdateRequest.ACTION.COMMIT, true, true);
- if (waitSearcher) {
- req.setAction(UpdateRequest.ACTION.COMMIT, true, true);
- }
- return req;
- }
-
- private BackendException storageException(Exception solrException) {
- return new TemporaryBackendException("Unable to complete query on Solr.", solrException);
- }
-
- private static void createCollectionIfNotExists(CloudSolrClient client, Configuration config, String collection)
- throws IOException, SolrServerException, KeeperException, InterruptedException {
- if (!checkIfCollectionExists(client, collection)) {
- Integer numShards = config.get(NUM_SHARDS);
- Integer maxShardsPerNode = config.get(MAX_SHARDS_PER_NODE);
- Integer replicationFactor = config.get(REPLICATION_FACTOR);
-
- CollectionAdminRequest.Create createRequest = new CollectionAdminRequest.Create();
-
- createRequest.setConfigName(collection);
- createRequest.setCollectionName(collection);
- createRequest.setNumShards(numShards);
- createRequest.setMaxShardsPerNode(maxShardsPerNode);
- createRequest.setReplicationFactor(replicationFactor);
-
- CollectionAdminResponse createResponse = createRequest.process(client);
- if (createResponse.isSuccess()) {
- logger.trace("Collection {} successfully created.", collection);
- } else {
- throw new SolrServerException(Joiner.on("\n").join(createResponse.getErrorMessages()));
- }
- }
-
- waitForRecoveriesToFinish(client, collection);
- }
-
- /**
- * Checks if the collection has already been created in Solr.
- */
- private static boolean checkIfCollectionExists(CloudSolrClient server, String collection) throws KeeperException, InterruptedException {
- ZkStateReader zkStateReader = server.getZkStateReader();
- zkStateReader.updateClusterState();
- ClusterState clusterState = zkStateReader.getClusterState();
- return clusterState.getCollectionOrNull(collection) != null;
- }
-
- /**
- * Wait for all the collection shards to be ready.
- */
- private static void waitForRecoveriesToFinish(CloudSolrClient server, String collection) throws KeeperException, InterruptedException {
- ZkStateReader zkStateReader = server.getZkStateReader();
- try {
- boolean cont = true;
-
- while (cont) {
- boolean sawLiveRecovering = false;
- zkStateReader.updateClusterState();
- ClusterState clusterState = zkStateReader.getClusterState();
- Map<String, Slice> slices = clusterState.getSlicesMap(collection);
- Preconditions.checkNotNull("Could not find collection:" + collection, slices);
-
- for (Map.Entry<String, Slice> entry : slices.entrySet()) {
- Map<String, Replica> shards = entry.getValue().getReplicasMap();
- for (Map.Entry<String, Replica> shard : shards.entrySet()) {
- String state = shard.getValue().getStr(ZkStateReader.STATE_PROP);
- if ((state.equals(Replica.State.RECOVERING)
- || state.equals(Replica.State.DOWN))
- && clusterState.liveNodesContain(shard.getValue().getStr(
- ZkStateReader.NODE_NAME_PROP))) {
- sawLiveRecovering = true;
- }
- }
- }
- if (!sawLiveRecovering) {
- cont = false;
- } else {
- Thread.sleep(1000);
- }
- }
- } finally {
- logger.info("Exiting solr wait");
- }
- }
-
- private static class GeoToWktConverter {
- /**
- * {@link com.thinkaurelius.titan.core.attribute.Geoshape} stores Points in the String format: point[X.0,Y.0].
- * Solr needs it to be in Well-Known Text format: POINT(X.0 Y.0)
- */
- static String convertToWktString(Geoshape fieldValue) throws BackendException {
- if (fieldValue.getType() == Geoshape.Type.POINT) {
- Geoshape.Point point = fieldValue.getPoint();
- return "POINT(" + point.getLongitude() + " " + point.getLatitude() + ")";
- } else {
- throw new PermanentBackendException("Cannot index " + fieldValue.getType());
- }
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/4fa10b6a/titan/src/main/java/com/thinkaurelius/titan/graphdb/query/graph/GraphCentricQueryBuilder.java
----------------------------------------------------------------------
diff --git a/titan/src/main/java/com/thinkaurelius/titan/graphdb/query/graph/GraphCentricQueryBuilder.java b/titan/src/main/java/com/thinkaurelius/titan/graphdb/query/graph/GraphCentricQueryBuilder.java
deleted file mode 100644
index c1a983b..0000000
--- a/titan/src/main/java/com/thinkaurelius/titan/graphdb/query/graph/GraphCentricQueryBuilder.java
+++ /dev/null
@@ -1,457 +0,0 @@
-/*
- * Copyright 2012-2013 Aurelius LLC
- * 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 com.thinkaurelius.titan.graphdb.query.graph;
-
-import com.google.common.base.Preconditions;
-import com.google.common.base.Predicate;
-import com.google.common.collect.*;
-import com.thinkaurelius.titan.core.*;
-import com.thinkaurelius.titan.core.attribute.Cmp;
-import com.thinkaurelius.titan.core.Cardinality;
-import com.thinkaurelius.titan.core.schema.SchemaStatus;
-import com.thinkaurelius.titan.core.schema.TitanSchemaType;
-import com.thinkaurelius.titan.graphdb.database.IndexSerializer;
-import com.thinkaurelius.titan.graphdb.internal.ElementCategory;
-import com.thinkaurelius.titan.graphdb.internal.InternalRelationType;
-import com.thinkaurelius.titan.graphdb.internal.OrderList;
-import com.thinkaurelius.titan.graphdb.query.*;
-import com.thinkaurelius.titan.graphdb.query.condition.*;
-import com.thinkaurelius.titan.graphdb.transaction.StandardTitanTx;
-import com.thinkaurelius.titan.graphdb.types.*;
-import com.thinkaurelius.titan.graphdb.types.system.ImplicitKey;
-import com.thinkaurelius.titan.util.datastructures.Interval;
-import com.thinkaurelius.titan.util.datastructures.PointInterval;
-import com.tinkerpop.blueprints.Edge;
-import com.tinkerpop.blueprints.Vertex;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-import java.util.*;
-
-/**
- *
- * Builds a {@link TitanGraphQuery}, optimizes the query and compiles the result into a {@link GraphCentricQuery} which
- * is then executed through a {@link QueryProcessor}.
- * This class from titan-0.5.4 has no major changes except adding a few logs for debugging index usage
- *
- * @author Matthias Broecheler (me@matthiasb.com)
- */
-public class GraphCentricQueryBuilder implements TitanGraphQuery<GraphCentricQueryBuilder> {
-
- private static final Logger log = LoggerFactory.getLogger(GraphCentricQueryBuilder.class);
-
- /**
- * Transaction in which this query is executed.
- */
- private final StandardTitanTx tx;
- /**
- * Serializer used to serialize the query conditions into backend queries.
- */
- private final IndexSerializer serializer;
- /**
- * The constraints added to this query. None by default.
- */
- private List<PredicateCondition<String, TitanElement>> constraints;
- /**
- * The order in which the elements should be returned. None by default.
- */
- private OrderList orders = new OrderList();
- /**
- * The limit of this query. No limit by default.
- */
- private int limit = Query.NO_LIMIT;
-
- public GraphCentricQueryBuilder(StandardTitanTx tx, IndexSerializer serializer) {
- log.debug("Loaded shaded version of class GraphCentricQueryBuilder");
- Preconditions.checkNotNull(tx);
- Preconditions.checkNotNull(serializer);
- this.tx = tx;
- this.serializer = serializer;
- this.constraints = new ArrayList<PredicateCondition<String, TitanElement>>(5);
- }
-
- /* ---------------------------------------------------------------
- * Query Construction
- * ---------------------------------------------------------------
- */
-
- private GraphCentricQueryBuilder has(String key, TitanPredicate predicate, Object condition) {
- Preconditions.checkNotNull(key);
- Preconditions.checkNotNull(predicate);
- Preconditions.checkArgument(predicate.isValidCondition(condition), "Invalid condition: %s", condition);
- constraints.add(new PredicateCondition<String, TitanElement>(key, predicate, condition));
- return this;
- }
-
- @Override
- public GraphCentricQueryBuilder has(String key, com.tinkerpop.blueprints.Predicate predicate, Object condition) {
- Preconditions.checkNotNull(key);
- TitanPredicate titanPredicate = TitanPredicate.Converter.convert(predicate);
- return has(key, titanPredicate, condition);
- }
-
- @Override
- public GraphCentricQueryBuilder has(PropertyKey key, TitanPredicate predicate, Object condition) {
- Preconditions.checkNotNull(key);
- Preconditions.checkNotNull(predicate);
- return has(key.getName(), predicate, condition);
- }
-
- @Override
- public GraphCentricQueryBuilder has(String key) {
- return has(key, Cmp.NOT_EQUAL, (Object) null);
- }
-
- @Override
- public GraphCentricQueryBuilder hasNot(String key) {
- return has(key, Cmp.EQUAL, (Object) null);
- }
-
- @Override
- @Deprecated
- public <T extends Comparable<T>> GraphCentricQueryBuilder has(String s, T t, Compare compare) {
- return has(s, compare, t);
- }
-
- @Override
- public GraphCentricQueryBuilder has(String key, Object value) {
- return has(key, Cmp.EQUAL, value);
- }
-
- @Override
- public GraphCentricQueryBuilder hasNot(String key, Object value) {
- return has(key, Cmp.NOT_EQUAL, value);
- }
-
- @Override
- public <T extends Comparable<?>> GraphCentricQueryBuilder interval(String s, T t1, T t2) {
- has(s, Cmp.GREATER_THAN_EQUAL, t1);
- return has(s, Cmp.LESS_THAN, t2);
- }
-
- @Override
- public GraphCentricQueryBuilder limit(final int limit) {
- Preconditions.checkArgument(limit >= 0, "Non-negative limit expected: %s", limit);
- this.limit = limit;
- return this;
- }
-
- @Override
- public GraphCentricQueryBuilder orderBy(String key, Order order) {
- Preconditions.checkArgument(tx.containsPropertyKey(key),"Provided key does not exist: %s",key);
- return orderBy(tx.getPropertyKey(key), order);
- }
-
- @Override
- public GraphCentricQueryBuilder orderBy(PropertyKey key, Order order) {
- Preconditions.checkArgument(key!=null && order!=null,"Need to specify and key and an order");
- Preconditions.checkArgument(Comparable.class.isAssignableFrom(key.getDataType()),
- "Can only order on keys with comparable data type. [%s] has datatype [%s]", key.getName(), key.getDataType());
- Preconditions.checkArgument(key.getCardinality()== Cardinality.SINGLE, "Ordering is undefined on multi-valued key [%s]", key.getName());
- Preconditions.checkArgument(!orders.containsKey(key));
- orders.add(key, order);
- return this;
- }
-
- /* ---------------------------------------------------------------
- * Query Execution
- * ---------------------------------------------------------------
- */
-
- @Override
- public Iterable<Vertex> vertices() {
- GraphCentricQuery query = constructQuery(ElementCategory.VERTEX);
- return Iterables.filter(new QueryProcessor<GraphCentricQuery, TitanElement, JointIndexQuery>(query, tx.elementProcessor), Vertex.class);
- }
-
- @Override
- public Iterable<Edge> edges() {
- GraphCentricQuery query = constructQuery(ElementCategory.EDGE);
- return Iterables.filter(new QueryProcessor<GraphCentricQuery, TitanElement, JointIndexQuery>(query, tx.elementProcessor), Edge.class);
- }
-
- @Override
- public Iterable<TitanProperty> properties() {
- GraphCentricQuery query = constructQuery(ElementCategory.PROPERTY);
- return Iterables.filter(new QueryProcessor<GraphCentricQuery, TitanElement, JointIndexQuery>(query, tx.elementProcessor), TitanProperty.class);
- }
-
- private QueryDescription describe(ElementCategory category) {
- return new StandardQueryDescription(1,constructQuery(category));
- }
-
- public QueryDescription describeForVertices() {
- return describe(ElementCategory.VERTEX);
- }
-
- public QueryDescription describeForEdges() {
- return describe(ElementCategory.EDGE);
- }
-
- public QueryDescription describeForProperties() {
- return describe(ElementCategory.PROPERTY);
- }
-
- /* ---------------------------------------------------------------
- * Query Construction
- * ---------------------------------------------------------------
- */
-
- private static final int DEFAULT_NO_LIMIT = 1000;
- private static final int MAX_BASE_LIMIT = 20000;
- private static final int HARD_MAX_LIMIT = 100000;
-
- private static final double EQUAL_CONDITION_SCORE = 4;
- private static final double OTHER_CONDITION_SCORE = 1;
- private static final double ORDER_MATCH = 2;
- private static final double ALREADY_MATCHED_ADJUSTOR = 0.1;
- private static final double CARDINALITY_SINGE_SCORE = 1000;
- private static final double CARDINALITY_OTHER_SCORE = 1000;
-
- public GraphCentricQuery constructQuery(final ElementCategory resultType) {
- Preconditions.checkNotNull(resultType);
- if (limit == 0) return GraphCentricQuery.emptyQuery(resultType);
-
- //Prepare constraints
- And<TitanElement> conditions = QueryUtil.constraints2QNF(tx, constraints);
- if (conditions == null) return GraphCentricQuery.emptyQuery(resultType);
-
- //Prepare orders
- orders.makeImmutable();
- if (orders.isEmpty()) orders = OrderList.NO_ORDER;
-
- //Compile all indexes that cover at least one of the query conditions
- final Set<IndexType> indexCandidates = new HashSet<IndexType>();
- ConditionUtil.traversal(conditions,new Predicate<Condition<TitanElement>>() {
- @Override
- public boolean apply(@Nullable Condition<TitanElement> condition) {
- if (condition instanceof PredicateCondition) {
- RelationType type = ((PredicateCondition<RelationType,TitanElement>)condition).getKey();
- Preconditions.checkArgument(type!=null && type.isPropertyKey());
- Iterables.addAll(indexCandidates,Iterables.filter(((InternalRelationType) type).getKeyIndexes(), new Predicate<IndexType>() {
- @Override
- public boolean apply(@Nullable IndexType indexType) {
- return indexType.getElement()==resultType;
- }
- }));
- }
- return true;
- }
- });
-
- /*
- Determine the best join index query to answer this query:
- Iterate over all potential indexes (as compiled above) and compute a score based on how many clauses
- this index covers. The index with the highest score (as long as it covers at least one additional clause)
- is picked and added to the joint query for as long as such exist.
- */
- JointIndexQuery jointQuery = new JointIndexQuery();
- boolean isSorted = orders.isEmpty();
- Set<Condition> coveredClauses = Sets.newHashSet();
- while (true) {
- IndexType bestCandidate = null;
- double candidateScore = 0.0;
- Set<Condition> candidateSubcover = null;
- boolean candidateSupportsSort = false;
- Object candidateSubcondition = null;
-
- for (IndexType index : indexCandidates) {
- Set<Condition> subcover = Sets.newHashSet();
- Object subcondition;
- boolean supportsSort = orders.isEmpty();
- //Check that this index actually applies in case of a schema constraint
- if (index.hasSchemaTypeConstraint()) {
- TitanSchemaType type = index.getSchemaTypeConstraint();
- Map.Entry<Condition,Collection<Object>> equalCon = getEqualityConditionValues(conditions,ImplicitKey.LABEL);
- if (equalCon==null) continue;
- Collection<Object> labels = equalCon.getValue();
- assert labels.size()>=1;
- if (labels.size()>1) {
- log.warn("The query optimizer currently does not support multiple label constraints in query: {}",this);
- continue;
- }
- if (!type.getName().equals((String)Iterables.getOnlyElement(labels))) continue;
- subcover.add(equalCon.getKey());
- }
-
- if (index.isCompositeIndex()) {
- subcondition = indexCover((CompositeIndexType) index,conditions,subcover);
- } else {
- subcondition = indexCover((MixedIndexType) index,conditions,serializer,subcover);
- if (coveredClauses.isEmpty() && !supportsSort
- && indexCoversOrder((MixedIndexType)index,orders)) supportsSort=true;
- }
- if (subcondition==null) continue;
- assert !subcover.isEmpty();
- double score = 0.0;
- boolean coversAdditionalClause = false;
- for (Condition c : subcover) {
- double s = (c instanceof PredicateCondition && ((PredicateCondition)c).getPredicate()==Cmp.EQUAL)?
- EQUAL_CONDITION_SCORE:OTHER_CONDITION_SCORE;
- if (coveredClauses.contains(c)) s=s*ALREADY_MATCHED_ADJUSTOR;
- else coversAdditionalClause = true;
- score+=s;
- if (index.isCompositeIndex())
- score+=((CompositeIndexType)index).getCardinality()==Cardinality.SINGLE?
- CARDINALITY_SINGE_SCORE:CARDINALITY_OTHER_SCORE;
- }
- if (supportsSort) score+=ORDER_MATCH;
- if (coversAdditionalClause && score>candidateScore) {
- candidateScore=score;
- bestCandidate=index;
- candidateSubcover = subcover;
- candidateSubcondition = subcondition;
- candidateSupportsSort = supportsSort;
- }
- }
- if (bestCandidate!=null) {
- if (coveredClauses.isEmpty()) isSorted=candidateSupportsSort;
- coveredClauses.addAll(candidateSubcover);
-
- log.debug("Index chosen for query {} {} " , bestCandidate.isCompositeIndex() ? "COMPOSITE" : "MIXED", coveredClauses);
- if (bestCandidate.isCompositeIndex()) {
- jointQuery.add((CompositeIndexType)bestCandidate,
- serializer.getQuery((CompositeIndexType)bestCandidate,(List<Object[]>)candidateSubcondition));
- } else {
- jointQuery.add((MixedIndexType)bestCandidate,
- serializer.getQuery((MixedIndexType)bestCandidate,(Condition)candidateSubcondition,orders));
- }
- } else {
- break;
- }
- /* TODO: smarter optimization:
- - use in-memory histograms to estimate selectivity of PredicateConditions and filter out low-selectivity ones
- if they would result in an individual index call (better to filter afterwards in memory)
- - move OR's up and extend GraphCentricQuery to allow multiple JointIndexQuery for proper or'ing of queries
- */
- }
-
- BackendQueryHolder<JointIndexQuery> query;
- if (!coveredClauses.isEmpty()) {
- int indexLimit = limit == Query.NO_LIMIT ? HARD_MAX_LIMIT : limit;
- if (tx.getGraph().getConfiguration().adjustQueryLimit()) {
- indexLimit = limit == Query.NO_LIMIT ? DEFAULT_NO_LIMIT : Math.min(MAX_BASE_LIMIT, limit);
- }
- indexLimit = Math.min(HARD_MAX_LIMIT, QueryUtil.adjustLimitForTxModifications(tx, coveredClauses.size(), indexLimit));
- jointQuery.setLimit(indexLimit);
- query = new BackendQueryHolder<JointIndexQuery>(jointQuery, coveredClauses.size()==conditions.numChildren(), isSorted, null);
- } else {
- query = new BackendQueryHolder<JointIndexQuery>(new JointIndexQuery(), false, isSorted, null);
- }
-
- return new GraphCentricQuery(resultType, conditions, orders, query, limit);
- }
-
- public static final boolean indexCoversOrder(MixedIndexType index, OrderList orders) {
- for (int i = 0; i < orders.size(); i++) {
- if (!index.indexesKey(orders.getKey(i))) return false;
- }
- return true;
- }
-
- public static List<Object[]> indexCover(final CompositeIndexType index, Condition<TitanElement> condition, Set<Condition> covered) {
- assert QueryUtil.isQueryNormalForm(condition);
- assert condition instanceof And;
- if (index.getStatus()!= SchemaStatus.ENABLED) return null;
- IndexField[] fields = index.getFieldKeys();
- Object[] indexValues = new Object[fields.length];
- Set<Condition> coveredClauses = new HashSet<Condition>(fields.length);
- List<Object[]> indexCovers = new ArrayList<Object[]>(4);
-
- constructIndexCover(indexValues,0,fields,condition,indexCovers,coveredClauses);
- if (!indexCovers.isEmpty()) {
- covered.addAll(coveredClauses);
- return indexCovers;
- } else return null;
- }
-
- private static void constructIndexCover(Object[] indexValues, int position, IndexField[] fields,
- Condition<TitanElement> condition,
- List<Object[]> indexCovers, Set<Condition> coveredClauses) {
- if (position>=fields.length) {
- indexCovers.add(indexValues);
- } else {
- IndexField field = fields[position];
- Map.Entry<Condition,Collection<Object>> equalCon = getEqualityConditionValues(condition,field.getFieldKey());
- if (equalCon!=null) {
- coveredClauses.add(equalCon.getKey());
- assert equalCon.getValue().size()>0;
- for (Object value : equalCon.getValue()) {
- Object[] newValues = Arrays.copyOf(indexValues,fields.length);
- newValues[position]=value;
- constructIndexCover(newValues,position+1,fields,condition,indexCovers,coveredClauses);
- }
- } else return;
- }
-
- }
-
- private static final Map.Entry<Condition,Collection<Object>> getEqualityConditionValues(Condition<TitanElement> condition, RelationType type) {
- for (Condition c : condition.getChildren()) {
- if (c instanceof Or) {
- Map.Entry<RelationType,Collection> orEqual = QueryUtil.extractOrCondition((Or)c);
- if (orEqual!=null && orEqual.getKey().equals(type) && !orEqual.getValue().isEmpty()) {
- return new AbstractMap.SimpleImmutableEntry(c,orEqual.getValue());
- }
- } else if (c instanceof PredicateCondition) {
- PredicateCondition<RelationType, TitanRelation> atom = (PredicateCondition)c;
- if (atom.getKey().equals(type) && atom.getPredicate()==Cmp.EQUAL && atom.getValue()!=null) {
- return new AbstractMap.SimpleImmutableEntry(c,ImmutableList.of(atom.getValue()));
- }
- }
-
- }
- return null;
- }
-
- public static final Condition<TitanElement> indexCover(final MixedIndexType index, Condition<TitanElement> condition,
- final IndexSerializer indexInfo, final Set<Condition> covered) {
- assert QueryUtil.isQueryNormalForm(condition);
- assert condition instanceof And;
- And<TitanElement> subcondition = new And<TitanElement>(condition.numChildren());
- for (Condition<TitanElement> subclause : condition.getChildren()) {
- if (coversAll(index,subclause,indexInfo)) {
- subcondition.add(subclause);
- covered.add(subclause);
- }
- }
- return subcondition.isEmpty()?null:subcondition;
- }
-
- private static final boolean coversAll(final MixedIndexType index, Condition<TitanElement> condition, IndexSerializer indexInfo) {
- if (condition.getType()==Condition.Type.LITERAL) {
- if (!(condition instanceof PredicateCondition)) return false;
- PredicateCondition<RelationType, TitanElement> atom = (PredicateCondition) condition;
- if (atom.getValue()==null) return false;
-
- Preconditions.checkArgument(atom.getKey().isPropertyKey());
- PropertyKey key = (PropertyKey) atom.getKey();
- ParameterIndexField[] fields = index.getFieldKeys();
- ParameterIndexField match = null;
- for (int i = 0; i < fields.length; i++) {
- if (fields[i].getStatus()!= SchemaStatus.ENABLED) continue;
- if (fields[i].getFieldKey().equals(key)) match = fields[i];
- }
- if (match==null) return false;
- return indexInfo.supports(index,match,atom.getPredicate());
- } else {
- for (Condition<TitanElement> child : condition.getChildren()) {
- if (!coversAll(index,child,indexInfo)) return false;
- }
- return true;
- }
- }
-}