public class SearchCache {

    private final ClusteringComparator comparator;
    private final Cache<UUID, SearchCacheEntry> cache;

    /**
     * Constructor taking the base table metadata and the max number of cache entries.
     *
     * @param metadata the base table metadata
     * @param cacheSize the max number of cache entries
     */
    public SearchCache(CFMetaData metadata, int cacheSize) {
        this.comparator = metadata.comparator;
        this.cache = CacheBuilder.newBuilder().maximumSize(cacheSize).build();
    }

    void put(UUID key, SearchCacheEntry entry) {
        cache.put(key, entry);
    }

    /**
     * Puts a cache entry associating the specified search and {@link ReadCommand} with the specified {@link Query}.
     *
     * @param search the search
     * @param command the read command
     * @param query the cached query
     */
    public void put(String search, ReadCommand command, Query query) {
        if (command instanceof PartitionRangeReadCommand) {
            PartitionRangeReadCommand rangeCommand = (PartitionRangeReadCommand) command;
            put(UUID.randomUUID(), new SearchCacheEntry(this, search, rangeCommand, query));
        }
    }

    /**
     * Returns a {@link SearchCacheUpdater} for updating an entry associating the specified search and {@link
     * ReadCommand} with the specified {@link Query}.
     *
     * @param search the search
     * @param command the read command
     * @param query the cached query
     * @return the cache updater
     */
    public SearchCacheUpdater updater(String search, ReadCommand command, Query query) {
        return new SearchCacheUpdater(this, search, UUID.randomUUID(), command, query);
    }

    /**
     * Discards all cached entries.
     */
    public void invalidate() {
        cache.invalidateAll();
    }

    /**
     * Gets the optional {@link SearchCacheEntry} associated to the specified search and {@link ReadCommand}.
     *
     * @param search the search
     * @param command the read command
     * @return the cache entry, maybe empty
     */
    public Optional<SearchCacheEntry> get(String search, ReadCommand command) {
        if (command instanceof PartitionRangeReadCommand) {
            PartitionRangeReadCommand rangeCommand = (PartitionRangeReadCommand) command;
            return cache.asMap().values().stream().filter(e -> e.isValid(comparator, search, rangeCommand)).findAny();
        }
        return Optional.empty();
    }

} public class SearchCacheEntry {

	private final SearchCache searchCache;
	private final String search;
	private final ReadCommand command;
	private final PartitionPosition currentPosition;
	private final Optional<Clustering> currentClustering;
	private final ScoreDoc scoreDoc;
	private final Query query;
	private final PartitionPosition startPosition;
	private final PartitionPosition stopPosition;
	private final Optional<ClusteringPrefix> startPrefix;
	private final Optional<ClusteringPrefix> stopPrefix;

	SearchCacheEntry(SearchCache searchCache, String search, PartitionRangeReadCommand command, Query query) {
		this.searchCache = searchCache;
		this.search = search;
		this.command = command;
		this.query = query;
		startPosition = command.dataRange().startKey();
		stopPosition = command.dataRange().stopKey();
		startPrefix = KeyMapper.startClusteringPrefix(command.dataRange());
		stopPrefix = KeyMapper.stopClusteringPrefix(command.dataRange());
		currentPosition = startPosition;
		currentClustering = startPrefix.isPresent() ? Optional.of(new Clustering(startPrefix.get().getRawValues()))
				: Optional.empty();
		scoreDoc = null;
	}

	SearchCacheEntry(SearchCache searchCache, String search, PartitionRangeReadCommand command,
			DecoratedKey decoratedKey, Optional<Clustering> currentClustering, ScoreDoc scoreDoc, Query query) {
		this.searchCache = searchCache;
		this.search = search;
		this.command = command;
		this.currentPosition = decoratedKey;
		this.currentClustering = currentClustering;
		this.scoreDoc = scoreDoc;
		this.query = query;
		startPosition = command.dataRange().startKey();
		stopPosition = command.dataRange().stopKey();
		startPrefix = KeyMapper.startClusteringPrefix(command.dataRange());
		stopPrefix = KeyMapper.stopClusteringPrefix(command.dataRange());
	}

	boolean isValid(ClusteringComparator comparator, String search, PartitionRangeReadCommand command) {
		if (search.equals(this.search)) {
			DataRange dataRange = command.dataRange();
			return validKey(dataRange) && validPrefix(comparator, dataRange);
		}
		return false;
	}

	private boolean validKey(DataRange dataRange) {
		PartitionPosition start = dataRange.startKey();
		if (currentPosition.compareTo(start) == 0 && startPosition.compareTo(start) <= 0) {
			PartitionPosition stop = dataRange.stopKey();
			return stopPosition.compareTo(stop) == 0;
		}
		return false;
	}

	private boolean validPrefix(ClusteringComparator comparator, DataRange dataRange) {

		// Discard start prefix
		Optional<ClusteringPrefix> start = KeyMapper.startClusteringPrefix(dataRange);
		if (start.isPresent() && startPrefix.isPresent() && comparator.compare(startPrefix.get(), start.get()) > 0) {
			return false;
		}

		// Discard null clusterings
		if (start.isPresent() != currentClustering.isPresent()) {
			return false;
		}

		// Discard clustering
		if (start.isPresent()
				&& comparator.compare(new Clustering(start.get().getRawValues()), currentClustering.get()) != 0) {
			return false;
		}

		// Discard stop prefix
		Optional<ClusteringPrefix> stop = KeyMapper.stopClusteringPrefix(dataRange);
		if (stop.isPresent() && stopPrefix.isPresent() && comparator.compare(stopPrefix.get(), stop.get()) != 0) {
			return false;
		}

		return true;
	}

	/**
	 * Returns a new {@link SearchCacheUpdater} for updating this entry.
	 *
	 * @return the cache updater
	 */
	public SearchCacheUpdater updater() {
		return searchCache.updater(search, command, query);
	}

	/**
	 * Returns the cached {@link ScoreDoc}.
	 *
	 * @return the score of the last cached position
	 */
	public ScoreDoc getScoreDoc() {
		return scoreDoc;
	}

	/**
	 * Returns the cached {@link Query}.
	 *
	 * @return the cached Lucene's query
	 */
	public Query getQuery() {
		return query;
	}
} public class SearchCacheUpdater {

    private final UUID id;
    private final SearchCache cache;
    private final String search;
    private final PartitionRangeReadCommand command;
    private final Query query;

    SearchCacheUpdater(SearchCache cache, String search, UUID id, ReadCommand command, Query query) {
        this.cache = cache;
        this.search = search;
        this.id = id;
        this.command = command instanceof PartitionRangeReadCommand ? (PartitionRangeReadCommand) command : null;
        this.query = query;
    }

    /**
     * Updates the cached entry with the specified pointer to a search result.
     *
     * @param key the row partition key
     * @param clustering the row clustering key
     * @param scoreDoc the row score for the query
     */
    public void put(DecoratedKey key, Clustering clustering, ScoreDoc scoreDoc) {
        if (command != null) {
            cache.put(id, new SearchCacheEntry(cache, search, command, key, Optional.of(clustering), scoreDoc, query));
        }
    }

    /**
     * Updates the cached entry with the specified pointer to a search result.
     *
     * @param key the row partition key
     * @param scoreDoc the row score for the query
     */
    public void put(DecoratedKey key, ScoreDoc scoreDoc) {
        if (command != null) {
            cache.put(id, new SearchCacheEntry(cache, search, command, key, Optional.empty(), scoreDoc, query));
        }
    }

} public class IndexOptions {

	public static final String REFRESH_SECONDS_OPTION = "refresh_seconds";
	public static final double DEFAULT_REFRESH_SECONDS = 60;

	public static final String RAM_BUFFER_MB_OPTION = "ram_buffer_mb";
	public static final int DEFAULT_RAM_BUFFER_MB = 64;

	public static final String MAX_MERGE_MB_OPTION = "max_merge_mb";
	public static final int DEFAULT_MAX_MERGE_MB = 5;

	public static final String MAX_CACHED_MB_OPTION = "max_cached_mb";
	public static final int DEFAULT_MAX_CACHED_MB = 30;

	public static final String INDEXING_THREADS_OPTION = "indexing_threads";
	public static final int DEFAULT_INDEXING_THREADS = 0;

	public static final String INDEXING_QUEUES_SIZE_OPTION = "indexing_queues_size";
	public static final int DEFAULT_INDEXING_QUEUES_SIZE = 50;

	public static final String SEARCH_CACHE_SIZE_OPTION = "search_cache_size";
	public static final int DEFAULT_SEARCH_CACHE_SIZE = 16;

	public static final String DIRECTORY_PATH_OPTION = "directory_path";
	public static final String INDEXES_DIR_NAME = "lucene";

	public static final String SEARCH_OPTION = "search";

	public static final String RESOURCE_TYPE_COLUMN = "resource_type_column";

	public final ResourceOptions search;

	/** The path of the directory where the index files will be stored */
	public final Path path;

	public final String resourceTypeColumn;

	/** The Lucene index searcher refresh frequency, in seconds */
	public final double refreshSeconds;

	/** The Lucene's max RAM buffer size, in MB */
	public final int ramBufferMB;

	/** The Lucene's max segments merge size size, in MB */
	public final int maxMergeMB;

	/** The Lucene's max cache size, in MB */
	public final int maxCachedMB;

	/** The number of asynchronous indexing threads */
	public final int indexingThreads;

	/** The size of the asynchronous indexing queues */
	public final int indexingQueuesSize;

	/** The max size of the search cache */
	public final int searchCacheSize;

	/** ColumnDefinition of the target column associated with the index **/
	public final ColumnDefinition targetColumn;

	/**
	 * Builds a new {@link IndexOptions} for the column family and index
	 * metadata.
	 *
	 * @param tableMetadata
	 *            the indexed table metadata
	 * @param indexMetadata
	 *            the index metadata
	 */
	public IndexOptions(CFMetaData tableMetadata, IndexMetadata indexMetadata) {
		Map<String, String> options = indexMetadata.options;
		refreshSeconds = parseRefresh(options);
		resourceTypeColumn = parseResourceTypeColumn(options);
		ramBufferMB = parseRamBufferMB(options);
		maxMergeMB = parseMaxMergeMB(options);
		maxCachedMB = parseMaxCachedMB(options);
		indexingThreads = parseIndexingThreads(options);
		indexingQueuesSize = parseIndexingQueuesSize(options);
		searchCacheSize = parseSearchCacheSize(options);
		path = parsePath(options, tableMetadata, indexMetadata);
		search = parseSearchOptions(options);

		String targetColumnName = indexMetadata.options.get("target");
		targetColumn = tableMetadata.getColumnDefinition(ByteBufferUtil.bytes(targetColumnName));

	}

	private String parseResourceTypeColumn(Map<String, String> options) {
		return options.get(RESOURCE_TYPE_COLUMN);
	}

	/**
	 * Validates the specified index options.
	 *
	 * @param options
	 *            the options to be validated
	 * @param metadata
	 *            the indexed table metadata
	 */
	public static void validateOptions(Map<String, String> options, CFMetaData metadata) {
		parseRefresh(options);
		parseRamBufferMB(options);
		parseMaxMergeMB(options);
		parseMaxCachedMB(options);
		parseIndexingThreads(options);
		parseIndexingQueuesSize(options);
		parseSearchCacheSize(options);
		parseSearchOptions(options);
		parsePath(options, metadata, null);
	}

	private static double parseRefresh(Map<String, String> options) {
		String refreshOption = options.get(REFRESH_SECONDS_OPTION);
		if (refreshOption != null) {
			double refreshSeconds;
			try {
				refreshSeconds = Double.parseDouble(refreshOption);
			} catch (NumberFormatException e) {
				throw new FhirIndexException("'%s' must be a strictly positive double", REFRESH_SECONDS_OPTION);
			}
			if (refreshSeconds <= 0) {
				throw new FhirIndexException("'%s' must be strictly positive", REFRESH_SECONDS_OPTION);
			}
			return refreshSeconds;
		} else {
			return DEFAULT_REFRESH_SECONDS;
		}
	}

	private static int parseRamBufferMB(Map<String, String> options) {
		String ramBufferSizeOption = options.get(RAM_BUFFER_MB_OPTION);
		if (ramBufferSizeOption != null) {
			int ramBufferMB;
			try {
				ramBufferMB = Integer.parseInt(ramBufferSizeOption);
			} catch (NumberFormatException e) {
				throw new FhirIndexException("'%s' must be a strictly positive integer", RAM_BUFFER_MB_OPTION);
			}
			if (ramBufferMB <= 0) {
				throw new FhirIndexException("'%s' must be strictly positive", RAM_BUFFER_MB_OPTION);
			}
			return ramBufferMB;
		} else {
			return DEFAULT_RAM_BUFFER_MB;
		}
	}

	private static int parseMaxMergeMB(Map<String, String> options) {
		String maxMergeSizeMBOption = options.get(MAX_MERGE_MB_OPTION);
		if (maxMergeSizeMBOption != null) {
			int maxMergeMB;
			try {
				maxMergeMB = Integer.parseInt(maxMergeSizeMBOption);
			} catch (NumberFormatException e) {
				throw new FhirIndexException("'%s' must be a strictly positive integer", MAX_MERGE_MB_OPTION);
			}
			if (maxMergeMB <= 0) {
				throw new FhirIndexException("'%s' must be strictly positive", MAX_MERGE_MB_OPTION);
			}
			return maxMergeMB;
		} else {
			return DEFAULT_MAX_MERGE_MB;
		}
	}

	private static int parseMaxCachedMB(Map<String, String> options) {
		String maxCachedMBOption = options.get(MAX_CACHED_MB_OPTION);
		if (maxCachedMBOption != null) {
			int maxCachedMB;
			try {
				maxCachedMB = Integer.parseInt(maxCachedMBOption);
			} catch (NumberFormatException e) {
				throw new FhirIndexException("'%s' must be a strictly positive integer", MAX_CACHED_MB_OPTION);
			}
			if (maxCachedMB <= 0) {
				throw new FhirIndexException("'%s' must be strictly positive", MAX_CACHED_MB_OPTION);
			}
			return maxCachedMB;
		} else {
			return DEFAULT_MAX_CACHED_MB;
		}
	}

	private static int parseIndexingThreads(Map<String, String> options) {
		String indexPoolNumQueuesOption = options.get(INDEXING_THREADS_OPTION);
		if (indexPoolNumQueuesOption != null) {
			try {
				return Integer.parseInt(indexPoolNumQueuesOption);
			} catch (NumberFormatException e) {
				throw new FhirIndexException("'%s' must be a positive integer", INDEXING_THREADS_OPTION);
			}
		} else {
			return DEFAULT_INDEXING_THREADS;
		}
	}

	private static int parseIndexingQueuesSize(Map<String, String> options) {
		String indexPoolQueuesSizeOption = options.get(INDEXING_QUEUES_SIZE_OPTION);
		if (indexPoolQueuesSizeOption != null) {
			int indexingQueuesSize;
			try {
				indexingQueuesSize = Integer.parseInt(indexPoolQueuesSizeOption);
			} catch (NumberFormatException e) {
				throw new FhirIndexException("'%s' must be a strictly positive integer", INDEXING_QUEUES_SIZE_OPTION);
			}
			if (indexingQueuesSize <= 0) {
				throw new FhirIndexException("'%s' must be strictly positive", INDEXING_QUEUES_SIZE_OPTION);
			}
			return indexingQueuesSize;
		} else {
			return DEFAULT_INDEXING_QUEUES_SIZE;
		}
	}

	private static int parseSearchCacheSize(Map<String, String> options) {
		String searchCacheSizeOption = options.get(SEARCH_CACHE_SIZE_OPTION);
		if (searchCacheSizeOption != null) {
			int searchCacheSize;
			try {
				searchCacheSize = Integer.parseInt(searchCacheSizeOption);
			} catch (NumberFormatException e) {
				throw new FhirIndexException("'%s' must be a positive integer", SEARCH_CACHE_SIZE_OPTION);
			}
			if (searchCacheSize < 0) {
				throw new FhirIndexException("'%s' must be positive", SEARCH_CACHE_SIZE_OPTION);
			}
			return searchCacheSize;
		} else {
			return DEFAULT_SEARCH_CACHE_SIZE;
		}
	}

	private static Path parsePath(Map<String, String> options, CFMetaData tableMetadata, IndexMetadata indexMetadata) {
		String pathOption = options.get(DIRECTORY_PATH_OPTION);
		if (pathOption != null) {
			return Paths.get(pathOption);
		} else if (indexMetadata != null) {
			Directories directories = new Directories(tableMetadata);
			String basePath = directories.getDirectoryForNewSSTables().getAbsolutePath();
			return Paths.get(basePath + File.separator + INDEXES_DIR_NAME + File.separator + indexMetadata.name);
		}
		return null;
	}

	private static ResourceOptions parseSearchOptions(Map<String, String> options) {
		String searchOption = options.get(SEARCH_OPTION);
		if (searchOption != null && !searchOption.trim().isEmpty()) {
			ResourceOptions searchOptions;
			try {
				searchOptions = JsonSerializer.fromString(searchOption, ResourceOptions.class);
				return searchOptions;
			} catch (Exception e) {
				throw new FhirIndexException(e, "'%s' is invalid : %s", SEARCH_OPTION, e.getMessage());
			}
		} else {
			throw new FhirIndexException("'%s' required", SEARCH_OPTION);
		}
	}

	@Override
	public String toString() {
		StringBuilder builder = new StringBuilder();
		builder.append("IndexOptions [path=");
		builder.append(path);
		builder.append(", resourceTypeColumn=");
		builder.append(resourceTypeColumn);
		builder.append(", refreshSeconds=");
		builder.append(refreshSeconds);
		builder.append(", ramBufferMB=");
		builder.append(ramBufferMB);
		builder.append(", maxMergeMB=");
		builder.append(maxMergeMB);
		builder.append(", maxCachedMB=");
		builder.append(maxCachedMB);
		builder.append(", indexingThreads=");
		builder.append(indexingThreads);
		builder.append(", indexingQueuesSize=");
		builder.append(indexingQueuesSize);
		builder.append(", searchCacheSize=");
		builder.append(searchCacheSize);
		builder.append(", targetColumn=");
		builder.append(targetColumn);
		builder.append("]");
		return builder.toString();
	}

} public class LuceneDocumentIterator implements CloseableIterator<Pair<Document, ScoreDoc>> {

	private static final Logger logger = LoggerFactory.getLogger(LuceneDocumentIterator.class);

	private final SearcherManager manager;
	private final Query query;
	private final Integer page;
	private final Set<String> fields;
	private final Deque<Pair<Document, ScoreDoc>> documents = new LinkedList<>();
	private Sort sort;
	private ScoreDoc after;
	private boolean mayHaveMore = true;

	/**
	 * Builds a new iterator over the {@link Document}s satisfying the specified
	 * {@link Query}.
	 *
	 * @param manager
	 *            the index searcher manager
	 * @param query
	 *            the query to be satisfied by the documents
	 * @param sort
	 *            the sort in which the documents are going to be retrieved
	 * @param after
	 *            a pointer to the start document (not included)
	 * @param limit
	 *            the max number of documents to be retrieved
	 * @param fields
	 *            the names of the fields to be loaded
	 */
	LuceneDocumentIterator(SearcherManager manager, Query query, Sort sort, ScoreDoc after, Integer limit,
			Set<String> fields) {
		this.manager = manager;
		this.query = query;
		this.sort = sort;
		this.after = after;
		this.page = limit < Integer.MAX_VALUE ? limit + 1 : limit;
		this.fields = fields;
	}

	private void fetch() {
		try {
			IndexSearcher searcher = manager.acquire();

			try {

				TimeCounter time = TimeCounter.create().start();

				// Search for top documents
				TopDocs topDocs = null;
				if (this.sort != null) {
					sort = sort.rewrite(searcher);
					topDocs = searcher.searchAfter(after, query, page, sort);
				} else {
					topDocs = searcher.searchAfter(after, query, page);
				}
				ScoreDoc[] scoreDocs = topDocs.scoreDocs;

				// Check inf mayHaveMore
				mayHaveMore = scoreDocs.length == page;

				// Collect the documents from query result
				for (ScoreDoc scoreDoc : scoreDocs) {
					Document document = searcher.doc(scoreDoc.doc, fields);
					documents.add(Pair.create(document, scoreDoc));
					after = scoreDoc;
				}

				logger.debug("Get page with {} documents in {}", scoreDocs.length, time.stop());

			} finally {
				manager.release(searcher);
			}

		} catch (Exception e) {
			e.printStackTrace();
			throw new FhirIndexException(e, "Error searching in with %s and %s", query, sort);
		}
	}

	/**
	 * Returns {@code true} if the iteration has more {@link Document}s. (In
	 * other words, returns {@code true} if {@link #next} would return an
	 * {@link Document} rather than throwing an exception.)
	 *
	 * @return {@code true} if the iteration has more {@link Document}s
	 */
	@Override
	public boolean hasNext() {
		if (needsFetch()) {
			fetch();
		}
		return !documents.isEmpty();
	}

	/**
	 * Returns if more {@link Document}s should be fetched from the Lucene
	 * index.
	 *
	 * @return {@code true} if more documents should be fetched, {@code false}
	 *         otherwise
	 */
	public boolean needsFetch() {
		return mayHaveMore && documents.isEmpty();
	}

	/**
	 * Returns the next {@link Document} in the iteration.
	 *
	 * @return the next document
	 * @throws NoSuchElementException
	 *             if the iteration has no more {@link Document}s
	 */
	@Override
	public Pair<Document, ScoreDoc> next() {
		if (hasNext()) {
			return documents.poll();
		} else {
			throw new NoSuchElementException();
		}
	}

	/** {@inheritDoc} */
	@Override
	public void close() {
	}
} public class LuceneService {
	private static final Logger logger = LoggerFactory.getLogger(LuceneService.class);

	private Path path;
	private String name;

	private Directory directory;
	private IndexWriter indexWriter;
	private SearcherManager searcherManager;
	private ControlledRealTimeReopenThread<IndexSearcher> searcherReopener;

	// Disable max boolean query clauses limit
	static {
		BooleanQuery.setMaxClauseCount(Integer.MAX_VALUE);
	}

	public LuceneService() {

	}

	/**
	 * Builds a new {@link FSIndex}.
	 *
	 * @param name
	 *            the index name
	 * @param mbeanName
	 *            the JMX MBean object name
	 * @param path
	 *            the directory path
	 * @param analyzer
	 *            the index writer analyzer
	 * @param refresh
	 *            the index reader refresh frequency in seconds
	 * @param ramBufferMB
	 *            the index writer RAM buffer size in MB
	 * @param maxMergeMB
	 *            the directory max merge size in MB
	 * @param maxCachedMB
	 *            the directory max cache size in MB
	 * @param refreshTask
	 *            action to be done during refresh
	 */
	public void init(String name, String mbeanName, Path path, Analyzer analyzer, double refresh, int ramBufferMB,
			int maxMergeMB, int maxCachedMB, Runnable refreshTask) {
		try {

			this.path = path;
			this.name = name;

			// Open or create directory
			FSDirectory fsDirectory = FSDirectory.open(path);
			this.directory = new NRTCachingDirectory(fsDirectory, maxMergeMB, maxCachedMB);

			// Setup index writer
			IndexWriterConfig indexWriterConfig = new IndexWriterConfig(analyzer);
			indexWriterConfig.setRAMBufferSizeMB(ramBufferMB);
			indexWriterConfig.setOpenMode(IndexWriterConfig.OpenMode.CREATE_OR_APPEND);
			indexWriterConfig.setUseCompoundFile(true);
			indexWriterConfig.setMergePolicy(new TieredMergePolicy());
			this.indexWriter = new IndexWriter(this.directory, indexWriterConfig);

			// Setup NRT search
			SearcherFactory searcherFactory = new SearcherFactory() {
				@Override
				public IndexSearcher newSearcher(IndexReader reader, IndexReader previousReader) {
					if (refreshTask != null) {
						refreshTask.run();
					}
					IndexSearcher searcher = new IndexSearcher(reader);
					searcher.setSimilarity(new NoIDFSimilarity());
					return searcher;
				}
			};
			TrackingIndexWriter trackingWriter = new TrackingIndexWriter(this.indexWriter);
			this.searcherManager = new SearcherManager(this.indexWriter, true, searcherFactory);
			this.searcherReopener = new ControlledRealTimeReopenThread<>(trackingWriter, this.searcherManager, refresh,
					refresh);
			this.searcherReopener.start();

			// Register JMX MBean
			// mbean = new ObjectName(mbeanName);
			// ManagementFactory.getPlatformMBeanServer().registerMBean(service,
			// this.mbean);

		} catch (Exception e) {
			throw new FhirIndexException(e, "Error while creating index %s", name);
		}
	}

	/**
	 * Upserts the specified {@link Document} by first deleting the documents
	 * containing {@code Term} and then adding the new document. The delete and
	 * then add are atomic as seen by a reader on the same index (flush may
	 * happen only after the add).
	 *
	 * @param term
	 *            the {@link Term} to identify the document(s) to be deleted
	 * @param document
	 *            the {@link Document} to be added
	 */
	public void upsert(Term term, Document document) {
		logger.debug("Indexing {} with term {} in {}", document, term, name);
		try {
			indexWriter.updateDocument(term, document);
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error indexing %s with term %s in %s", document, term, name);
		}
	}

	/**
	 * Deletes all the {@link Document}s containing the specified {@link Term}.
	 *
	 * @param term
	 *            the {@link Term} identifying the documents to be deleted
	 */
	public void delete(Term term) {
		logger.debug("Deleting {} from {}", term, name);
		try {
			indexWriter.deleteDocuments(term);
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error deleting %s from %s", term, name);
		}
	}

	/**
	 * Deletes all the {@link Document}s satisfying the specified {@link Query}.
	 *
	 * @param query
	 *            the {@link Query} identifying the documents to be deleted
	 */
	public void delete(Query query) {
		logger.debug("Deleting {} from {}", query, name);
		try {
			indexWriter.deleteDocuments(query);
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error deleting %s from %s", query, name);
		}
	}

	/**
	 * Deletes all the {@link Document}s.
	 */
	public void truncate() {
		try {
			indexWriter.deleteAll();
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error truncating %s", name);
		}
		logger.info("Truncated {}", name);
	}

	/**
	 * Commits the pending changes.
	 */

	public void commit() {
		try {
			indexWriter.commit();
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error committing %s", name);
		}
		logger.debug("Committed {}", name);
	}

	/**
	 * Commits all changes to the index, waits for pending merges to complete,
	 * and closes all associated resources.
	 */
	public void close() {
		try {
			searcherReopener.interrupt();
			searcherManager.close();
			indexWriter.close();
			directory.close();
			// ManagementFactory.getPlatformMBeanServer().unregisterMBean(mbean);
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error closing %s", name);
		}
		logger.info("Closed {}", name);
	}

	/**
	 * Closes the index and removes all its files.
	 */
	public void delete() {
		try {
			close();
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error deleting %s", name);
		} finally {
			FileUtils.deleteRecursive(path.toFile());
		}
		logger.info("Deleted {}", name);
	}

	/**
	 * Finds the top {@code count} hits for {@code query} and sorting the hits
	 * by {@code sort}.
	 *
	 * @param query
	 *            the {@link Query} to search for
	 * @param sort
	 *            the {@link Sort} to be applied
	 * @param after
	 *            the starting {@link ScoreDoc}
	 * @param count
	 *            the max number of results to be collected
	 * @param fields
	 *            the names of the fields to be loaded
	 * @return the found documents, sorted according to the supplied
	 *         {@link Sort} instance
	 */
	public LuceneDocumentIterator search(Query query, Sort sort, ScoreDoc after, Integer count, Set<String> fields) {
		logger.debug("Searching in {}\n" + "count: {}\n" + "after: {}\n" + "query: {}\n" + " sort: {}", name, count,
				after, query, sort);
		return new LuceneDocumentIterator(searcherManager, query, sort, after, count, fields);
	}

	/**
	 * Returns the total number of {@link Document}s in this index.
	 *
	 * @return the number of {@link Document}s
	 */

	public long getNumDocs() {
		logger.debug("Getting {} num docs", name);
		try {
			IndexSearcher searcher = searcherManager.acquire();
			try {
				return searcher.getIndexReader().numDocs();
			} finally {
				searcherManager.release(searcher);
			}
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error getting %s num docs", name);
		}
	}

	/**
	 * Returns the total number of deleted {@link Document}s in this index.
	 *
	 * @return the number of deleted {@link Document}s
	 */

	public long getNumDeletedDocs() {
		logger.debug("Getting %s num deleted docs", name);
		try {
			IndexSearcher searcher = searcherManager.acquire();
			try {
				return searcher.getIndexReader().numDeletedDocs();
			} finally {
				searcherManager.release(searcher);
			}
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error getting %s num docs", name);
		}
	}

	/**
	 * Optimizes the index forcing merge segments leaving the specified number
	 * of segments. This operation may block until all merging completes.
	 *
	 * @param maxNumSegments
	 *            the maximum number of segments left in the index after merging
	 *            finishes
	 * @param doWait
	 *            {@code true} if the call should block until the operation
	 *            completes
	 */

	public void forceMerge(int maxNumSegments, boolean doWait) {
		logger.info("Merging {} segments to {}", name, maxNumSegments);
		try {
			indexWriter.forceMerge(maxNumSegments, doWait);
			indexWriter.commit();
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error merging %s segments to %s", name, maxNumSegments);
		}
		logger.info("Merged {} segments to {}", name, maxNumSegments);
	}

	/**
	 * Optimizes the index forcing merge of all segments that have deleted
	 * documents. This operation may block until all merging completes.
	 *
	 * @param doWait
	 *            {@code true} if the call should block until the operation
	 *            completes
	 */
	public void forceMergeDeletes(boolean doWait) {
		logger.info("Merging {} segments with deletions", name);
		try {
			indexWriter.forceMergeDeletes(doWait);
			indexWriter.commit();
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error merging %s segments with deletion", name);
		}
		logger.info("Merged {} segments with deletions", name);
	}

	/**
	 * Refreshes the index readers.
	 */
	public void refresh() {
		logger.debug("Refreshing {} readers...", name);
		try {
			commit();
			searcherManager.maybeRefreshBlocking();
		} catch (Exception e) {
			throw new FhirIndexException(e, "Error refreshing %s readers", name);
		}
		logger.debug("Refreshed {} readers", name);
	}
} public class FhirMapper {

	private ResourceOptions searchOptions;

	private SearchParamExtractorHelper fhirExtractor;

	public FhirMapper(ResourceOptions searchOptions) {
		this.searchOptions = searchOptions;
		this.fhirExtractor = new SearchParamExtractorHelper(FhirVersionEnum.DSTU3);
	}

	/**
	 * Parses Json FHIR Resource and converts to Lucene {@link Field}s.
	 * 
	 * @param document,
	 *            the Lucene {@link Document}.
	 * @param json,
	 *            the FHIR resource as JSON format.
	 */
	public void addFields(Document document, String json) {

		IBaseResource resourceInstance = this.fhirExtractor.parseResource(json);
		String resourceName = resourceInstance.getClass().getSimpleName();

		Set<String> parameters = this.searchOptions.resources.get(resourceName);
		Set<AbstractSearchParam> values = this.fhirExtractor.extractParametersValues(resourceInstance, parameters);
		for (AbstractSearchParam entry : values) {
			doAddFields(document, entry, false);
		}

		doAddFields(document, new SearchParamString("resource_type", "", SearchParamTypes.STRING, resourceName), false);

	}

	/**
	 * Adds the specified column name and value to a Lucene {@link Document}.
	 * The added fields are indexed and sorted (if the parameter sorted is
	 * <code>true</code>).
	 *
	 * @param document
	 *            a {@link Document}
	 * @param value
	 *            the parameter to add to a {@link Document}
	 * @param sorted
	 *            sort the value or not
	 */
	private void doAddFields(Document document, AbstractSearchParam value, boolean sorted) {
		if (value != null && value.hasValue()) {
			doAddIndexedFields(document, value);

			if (sorted) {
				doAddSortedFields(document, value);
			}
		}
	}

	/**
	 * Add a {@link Field} to be sorted by.
	 * 
	 * @param document
	 * @param param
	 */
	private void doAddSortedFields(Document document, AbstractSearchParam param) {
		// BytesRef bytes = new BytesRef(value);
		// document.add(new SortedDocValuesField(value.getName(), bytes));

		// TODO: implement sort
		List<Field> fields = param.createSortedFields();
		for (Field field : fields) {
			if (field != null) {
				document.add(field);
			}
		}
	}

	/**
	 * Add an indexed {@link Field}.
	 * 
	 * @param document
	 * @param param
	 */
	private void doAddIndexedFields(Document document, AbstractSearchParam param) {
		List<Field> fields = param.createIndexedFields();
		for (Field field : fields) {
			if (field != null) {
				document.add(field);
			}
		}
	}

} public final class KeyMapper {

	public static final Logger logger = LoggerFactory.getLogger(KeyMapper.class);

	/** The Lucene field name. */
	public static final String FIELD_NAME = "_primary_key";

	/** The Lucene field type. */
	public static final FieldType FIELD_TYPE = new FieldType();

	static {
		FIELD_TYPE.setOmitNorms(true);
		FIELD_TYPE.setIndexOptions(IndexOptions.DOCS);
		FIELD_TYPE.setTokenized(false);
		FIELD_TYPE.setStored(true);
		FIELD_TYPE.setDocValuesType(DocValuesType.SORTED);
		FIELD_TYPE.freeze();
	}

	/** The indexed table metadata */
	private final CFMetaData metadata;

	/** The clustering key comparator */
	private final ClusteringComparator clusteringComparator;

	/** A composite type composed by the types of the clustering key */
	private final CompositeType clusteringType;

	/**
	 * The type of the primary key, which is composed by token, partition key
	 * and clustering key types.
	 */
	private final CompositeType type;

	/**
	 * Constructor specifying the partition and clustering key mappers.
	 *
	 * @param metadata
	 *            the indexed table metadata
	 */
	public KeyMapper(CFMetaData metadata) {
		this.metadata = metadata;
		clusteringComparator = metadata.comparator;
		clusteringType = CompositeType.getInstance(clusteringComparator.subtypes());
		type = CompositeType.getInstance(LongType.instance, metadata.getKeyValidator(), clusteringType);
	}

	/**
	 * Returns the clustering key comparator.
	 *
	 * @return the comparator
	 */
	public ClusteringComparator clusteringComparator() {
		return clusteringComparator;
	}

	/**
	 * The type of the primary key, which is composed by token, partition key
	 * and clustering key types.
	 *
	 * @return the composite type
	 */
	public CompositeType clusteringType() {
		return clusteringType;
	}

	/**
	 * Returns the {@link KeyEntry} represented by the specified Lucene
	 * {@link BytesRef}.
	 *
	 * @param bytesRef
	 *            the Lucene field binary value
	 * @return the represented key entry
	 */
	public KeyEntry entry(BytesRef bytesRef) {
		ByteBuffer bb = ByteBufferUtils.byteBuffer(bytesRef);
		ByteBuffer[] components = type.split(bb);
		return new KeyEntry(this, components);
	}

	/**
	 * Returns the {@code String} human-readable representation of the specified
	 * {@link ClusteringPrefix}.
	 *
	 * @param prefix
	 *            the clustering prefix
	 * @return a {@code String} representing {@code prefix}
	 */
	public String toString(ClusteringPrefix prefix) {
		return prefix.toString(metadata);
	}

	/**
	 * Returns the {@link ByteBuffer} representation of the primary key formed
	 * by the specified partition key and the clustering key.
	 *
	 * @param key
	 *            the partition key
	 * @param clustering
	 *            the clustering key
	 * @return the {@link ByteBuffer} representation of the primary key
	 */
	public ByteBuffer byteBuffer(DecoratedKey key, Clustering clustering) {
		return type.builder().add(TokenMapper.byteBuffer(key.getToken())).add(key.getKey()).add(byteBuffer(clustering))
				.build();
	}

	/**
	 * Returns a {@link ByteBuffer} representing the specified clustering key
	 *
	 * @param clustering
	 *            the clustering key
	 * @return the byte buffer representing {@code clustering}
	 */
	public ByteBuffer byteBuffer(Clustering clustering) {
		CompositeType.Builder builder = type.builder();
		for (ByteBuffer component : clustering.getRawValues()) {
			builder.add(component);
		}
		return builder.build();
	}

	/**
	 * Adds to the specified Lucene {@link Document} the primary key formed by
	 * the specified partition key and the clustering key.
	 *
	 * @param document
	 *            the Lucene {@link Document} in which the key is going to be
	 *            added
	 * @param key
	 *            the partition key
	 * @param clustering
	 *            the clustering key
	 */
	public void addFields(Document document, DecoratedKey key, Clustering clustering) {
		ByteBuffer bb = byteBuffer(key, clustering);
		BytesRef bytesRef = ByteBufferUtils.bytesRef(bb);
		Field field = new Field(FIELD_NAME, bytesRef, FIELD_TYPE);
		document.add(field);
	}

	/**
	 * Returns the Lucene {@link Term} representing the primary key formed by
	 * the specified partition key and the clustering key.
	 *
	 * @param key
	 *            the partition key
	 * @param clustering
	 *            the clustering key
	 * @return the Lucene {@link Term} representing the primary key
	 */
	public Term term(DecoratedKey key, Clustering clustering) {
		return new Term(FIELD_NAME, bytesRef(key, clustering));
	}

	/**
	 * Returns the {@link BytesRef} representation of the specified primary key.
	 *
	 * @param key
	 *            the partition key
	 * @param clustering
	 *            the clustering key
	 * @return the Lucene field binary value
	 */
	public BytesRef bytesRef(DecoratedKey key, Clustering clustering) {
		ByteBuffer bb = byteBuffer(key, clustering);
		return ByteBufferUtils.bytesRef(bb);
	}

	/**
	 * Returns the clustering key contained in the specified {@link Document}.
	 *
	 * @param document
	 *            a {@link Document} containing the clustering key to be get
	 * @return the clustering key contained in {@code document}
	 */
	public Clustering clustering(Document document) {
		BytesRef bytesRef = document.getBinaryValue(FIELD_NAME);
		return entry(bytesRef).getClustering();
	}

	/**
	 * Returns the start {@link ClusteringPrefix} of the first partition of the
	 * specified {@link DataRange}.
	 *
	 * @param dataRange
	 *            the data range
	 * @return the start clustering prefix of {@code dataRange}, or {@code null}
	 *         if there is no such start
	 */
	public static Optional<ClusteringPrefix> startClusteringPrefix(DataRange dataRange) {
		PartitionPosition startPosition = dataRange.startKey();
		if (startPosition instanceof DecoratedKey) {
			DecoratedKey startKey = (DecoratedKey) startPosition;
			ClusteringIndexFilter filter = dataRange.clusteringIndexFilter(startKey);
			if (filter instanceof ClusteringIndexSliceFilter) {
				ClusteringIndexSliceFilter sliceFilter = (ClusteringIndexSliceFilter) filter;
				Slices slices = sliceFilter.requestedSlices();
				return Optional.of(slices.get(0).start());
			}
		}
		return Optional.empty();
	}

	/**
	 * Returns the stop {@link ClusteringPrefix} of the last partition of the
	 * specified {@link DataRange}.
	 *
	 * @param dataRange
	 *            the data range
	 * @return the stop clustering prefix of {@code dataRange}, or {@code null}
	 *         if there is no such start
	 */
	public static Optional<ClusteringPrefix> stopClusteringPrefix(DataRange dataRange) {
		PartitionPosition stopPosition = dataRange.stopKey();
		if (stopPosition instanceof DecoratedKey) {
			DecoratedKey stopKey = (DecoratedKey) stopPosition;
			ClusteringIndexFilter filter = dataRange.clusteringIndexFilter(stopKey);
			if (filter instanceof ClusteringIndexSliceFilter) {
				ClusteringIndexSliceFilter sliceFilter = (ClusteringIndexSliceFilter) filter;
				Slices slices = sliceFilter.requestedSlices();
				return Optional.of(slices.get(slices.size() - 1).end());
			}
		}
		return Optional.empty();
	}

	/**
	 * Returns the {@link Term} representing the primary key of the specified
	 * {@link Document}.
	 *
	 * @param document
	 *            the document
	 * @return the clustering key term
	 */
	public static Term term(Document document) {
		BytesRef bytesRef = document.getBinaryValue(FIELD_NAME);
		return new Term(FIELD_NAME, bytesRef);
	}

	/**
	 * Returns a Lucene {@link SortField} to sort documents by primary key
	 * according to Cassandra's natural order.
	 *
	 * @return the sort field
	 */
	public SortField sortField() {
		return new SortField(FIELD_NAME, new FieldComparatorSource() {
			@Override
			public FieldComparator<?> newComparator(String field, int hits, int sort, boolean reversed)
					throws IOException {
				return new FieldComparator.TermValComparator(hits, field, false) {
					@Override
					public int compareValues(BytesRef val1, BytesRef val2) {
						return entry(val1).compareTo(entry(val2));
					}
				};
			}
		});
	}

} public final class PartitionMapper {

	/** The Lucene field name. */
	public static final String FIELD_NAME = "_partition_key";

	/** The Lucene field type. */
	public static final FieldType FIELD_TYPE = new FieldType();

	static {
		FIELD_TYPE.setOmitNorms(true);
		FIELD_TYPE.setIndexOptions(IndexOptions.DOCS);
		FIELD_TYPE.setTokenized(false);
		FIELD_TYPE.setStored(true);
		FIELD_TYPE.setDocValuesType(DocValuesType.SORTED);
		FIELD_TYPE.freeze();
	}

	private final CFMetaData metadata;
	private final IPartitioner partitioner;
	private final AbstractType<?> type;

	/**
	 * Constructor specifying the indexed table {@link CFMetaData}.
	 *
	 * @param metadata
	 *            the indexed table metadata
	 */
	public PartitionMapper(CFMetaData metadata) {
		this.metadata = metadata;
		partitioner = DatabaseDescriptor.getPartitioner();
		type = metadata.getKeyValidator();
	}

	/**
	 * Returns the type of the partition key.
	 *
	 * @return the key's type
	 */
	public AbstractType<?> getType() {
		return type;
	}

	/**
	 * Adds to the specified {@link Document} the {@link Field}s associated to
	 * the specified partition key.
	 *
	 * @param document
	 *            the document in which the fields are going to be added
	 * @param partitionKey
	 *            the partition key to be converted
	 */
	public void addFields(Document document, DecoratedKey partitionKey) {
		ByteBuffer bb = partitionKey.getKey();
		BytesRef bytesRef = ByteBufferUtils.bytesRef(bb);
		document.add(new Field(FIELD_NAME, bytesRef, FIELD_TYPE));
	}

	/**
	 * Returns the specified raw partition key as a Lucene {@link Term}.
	 *
	 * @param partitionKey
	 *            the raw partition key to be converted
	 * @return a Lucene {@link Term}
	 */
	public Term term(DecoratedKey partitionKey) {
		ByteBuffer bb = partitionKey.getKey();
		BytesRef bytesRef = ByteBufferUtils.bytesRef(bb);
		return new Term(FIELD_NAME, bytesRef);
	}

	/**
	 * Returns the {@link Term} representing the partition key of the specified
	 * {@link Document}.
	 *
	 * @param document
	 *            the document
	 * @return the partition key term
	 */
	public Term term(Document document) {
		BytesRef bytesRef = document.getBinaryValue(FIELD_NAME);
		return new Term(FIELD_NAME, bytesRef);
	}

	/**
	 * Returns the specified raw partition key as a Lucene {@link Query}.
	 *
	 * @param partitionKey
	 *            the raw partition key to be converted
	 * @return the specified raw partition key as a Lucene {@link Query}
	 */
	public Query query(DecoratedKey partitionKey) {
		return new TermQuery(term(partitionKey));
	}

	/**
	 * Returns the {@link DecoratedKey} contained in the specified Lucene
	 * {@link Document}.
	 *
	 * @param document
	 *            the {@link Document} containing the partition key to be get
	 * @return the {@link DecoratedKey} contained in the specified Lucene
	 *         {@link Document}
	 */
	public DecoratedKey decoratedKey(Document document) {
		BytesRef bytesRef = document.getBinaryValue(FIELD_NAME);
		ByteBuffer bb = ByteBufferUtils.byteBuffer(bytesRef);
		return decoratedKey(bb);
	}

	/**
	 * Returns the specified binary partition key as a {@link DecoratedKey}.
	 *
	 * @param partitionKey
	 *            the binary representation of a partition key
	 * @return the specified partition key as a a {@link DecoratedKey}
	 */
	public DecoratedKey decoratedKey(ByteBuffer partitionKey) {
		return partitioner.decorateKey(partitionKey);
	}

	/**
	 * Returns a Lucene {@link SortField} for sorting documents/rows according
	 * to the partition key.
	 *
	 * @return a sort field for sorting by partition key
	 */
	public SortField sortField() {
		return new SortField(FIELD_NAME, new FieldComparatorSource() {
			@Override
			public FieldComparator<?> newComparator(String field, int hits, int sort, boolean reversed)
					throws IOException {
				return new FieldComparator.TermValComparator(hits, field, false) {
					@Override
					public int compareValues(BytesRef val1, BytesRef val2) {
						ByteBuffer bb1 = ByteBufferUtils.byteBuffer(val1);
						ByteBuffer bb2 = ByteBufferUtils.byteBuffer(val2);
						return ByteBufferUtil.compareUnsigned(bb1, bb2);
					}
				};
			}
		});
	}

} public final class TokenMapper {

	/** The Lucene field name */
	static final String FIELD_NAME = "_token";

	/** The Lucene field type */
	static final FieldType FIELD_TYPE = new FieldType();

	static {
		FIELD_TYPE.setTokenized(true);
		FIELD_TYPE.setOmitNorms(true);
		FIELD_TYPE.setIndexOptions(IndexOptions.DOCS);
		FIELD_TYPE.setNumericType(FieldType.NumericType.LONG);
		FIELD_TYPE.setDocValuesType(DocValuesType.NUMERIC);
		FIELD_TYPE.freeze();
	}

	/**
	 * Adds to the specified {@link Document} the {@link Field}s associated to
	 * the token of the specified row key.
	 *
	 * @param document
	 *            a {@link Document}
	 * @param key
	 *            the raw partition key to be added
	 */
	public void addFields(Document document, DecoratedKey key) {
		Token token = key.getToken();
		Long value = value(token);
		Field field = new LongField(FIELD_NAME, value, FIELD_TYPE);
		document.add(field);
	}

	/**
	 * Returns the {code Long} value of the specified Murmur3 partitioning
	 * {@link Token}.
	 *
	 * @param token
	 *            a Murmur3 token
	 * @return the {@code token}'s {code Long} value
	 */
	public static Long value(Token token) {
		return (Long) token.getTokenValue();
	}

	/**
	 * Returns the {code ByteBuffer} value of the specified Murmur3 partitioning
	 * {@link Token}.
	 *
	 * @param token
	 *            a Murmur3 token
	 * @return the {@code token}'s {code ByteBuffer} value
	 */
	public static ByteBuffer byteBuffer(Token token) {
		return LongType.instance.decompose(value(token));
	}

	/**
	 * Returns the {@link BytesRef} indexing value of the specified Murmur3
	 * partitioning {@link Token}.
	 *
	 * @param token
	 *            a Murmur3 token
	 * @return the {@code token}'s indexing value
	 */
	private static BytesRef bytesRef(Token token) {
		Long value = value(token);
		BytesRefBuilder bytesRef = new BytesRefBuilder();
		NumericUtils.longToPrefixCoded(value, 0, bytesRef);
		return bytesRef.get();
	}

	/**
	 * Returns a Lucene {@link SortField} for sorting documents/rows according
	 * to the partitioner's order.
	 *
	 * @return a sort field for sorting by token
	 */
	public SortField sortField() {
		return new SortField(FIELD_NAME, SortField.Type.LONG);
	}

	/**
	 * Returns if the specified lower partition position must be included in a
	 * filtered range.
	 *
	 * @param position
	 *            a {@link PartitionPosition}
	 * @return {@code true} if {@code position} must be included, {@code false}
	 *         otherwise
	 */
	public boolean includeStart(PartitionPosition position) {
		return position.kind() == PartitionPosition.Kind.MIN_BOUND;
	}

	/**
	 * Returns if the specified upper partition position must be included in a
	 * filtered range.
	 *
	 * @param position
	 *            a {@link PartitionPosition}
	 * @return {@code true} if {@code position} must be included, {@code false}
	 *         otherwise
	 */
	public boolean includeStop(PartitionPosition position) {
		return position.kind() == PartitionPosition.Kind.MAX_BOUND;
	}

	/**
	 * Returns a Lucene {@link Query} to find the {@link Document}s containing
	 * the specified {@link Token}.
	 *
	 * @param token
	 *            the token
	 * @return the query to find the documents containing {@code token}
	 */
	public Query query(Token token) {
		return new TermQuery(new Term(FIELD_NAME, bytesRef(token)));
	}
} public final class ByteBufferUtils {

    /** Private constructor to hide the implicit public one. */
    private ByteBufferUtils() {
    }

    /**
     * Returns the specified {@link ByteBuffer} as a byte array.
     *
     * @param byteBuffer a {@link ByteBuffer} to be converted to a byte array
     * @return the byte array representation of the {@code byteBuffer}
     */
    public static byte[] asArray(ByteBuffer byteBuffer) {
        ByteBuffer bb = ByteBufferUtil.clone(byteBuffer);
        byte[] bytes = new byte[bb.remaining()];
        bb.get(bytes);
        return bytes;
    }

    /**
     * Returns {@code true} if the specified {@link ByteBuffer} is empty, {@code false} otherwise.
     *
     * @param byteBuffer the byte buffer
     * @return {@code true} if the specified {@link ByteBuffer} is empty, {@code false} otherwise.
     */
    public static boolean isEmpty(ByteBuffer byteBuffer) {
        return byteBuffer.remaining() == 0;
    }

    /**
     * Returns the {@link ByteBuffer}s contained in {@code byteBuffer} according to {@code type}.
     *
     * @param byteBuffer the byte buffer to be split
     * @param type the {@link AbstractType} of {@code byteBuffer}
     * @return the {@link ByteBuffer}s contained in {@code byteBuffer} according to {@code type}
     */
    public static ByteBuffer[] split(ByteBuffer byteBuffer, AbstractType<?> type) {
        if (type instanceof CompositeType) {
            return ((CompositeType) type).split(byteBuffer);
        } else {
            return new ByteBuffer[]{byteBuffer};
        }
    }

    /**
     * Returns a {@code String} representation of {@code byteBuffer} validated by {@code type}.
     *
     * @param byteBuffer the {@link ByteBuffer} to be converted to {@code String}
     * @param type {@link AbstractType} of {@code byteBuffer}
     * @return a {@code String} representation of {@code byteBuffer} validated by {@code type}
     */
    public static String toString(ByteBuffer byteBuffer, AbstractType<?> type) {
        if (type instanceof CompositeType) {
            CompositeType composite = (CompositeType) type;
            List<AbstractType<?>> types = composite.types;
            ByteBuffer[] components = composite.split(byteBuffer);
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < components.length; i++) {
                AbstractType<?> componentType = types.get(i);
                ByteBuffer component = components[i];
                sb.append(componentType.compose(component));
                if (i < types.size() - 1) {
                    sb.append(':');
                }
            }
            return sb.toString();
        } else {
            return type.compose(byteBuffer).toString();
        }
    }

    /**
     * Returns the hexadecimal {@code String} representation of the specified {@link ByteBuffer}.
     *
     * @param byteBuffer a {@link ByteBuffer}
     * @return the hexadecimal {@code String} representation of {@code byteBuffer}
     */
    public static String toHex(ByteBuffer byteBuffer) {
        return ByteBufferUtil.bytesToHex(byteBuffer);
    }

    /**
     * Returns the hexadecimal {@code String} representation of the specified {@code byte} array.
     *
     * @param bytes the {@code byte} array
     * @return The hexadecimal {@code String} representation of {@code bytes}
     */
    public static String toHex(byte[] bytes) {
        return Hex.bytesToHex(bytes);
    }

    /**
     * Returns the hexadecimal {@code String} representation of the specified {@code byte}.
     *
     * @param b the {@code byte}
     * @return the hexadecimal {@code String} representation of {@code b}
     */
    public static String toHex(byte b) {
        return Hex.bytesToHex(b);
    }

    /**
     * Returns the {@link BytesRef} representation of the specified {@link ByteBuffer}.
     *
     * @param bb the byte buffer
     * @return the {@link BytesRef} representation of the byte buffer
     */
    public static BytesRef bytesRef(ByteBuffer bb) {
        byte[] bytes = asArray(bb);
        return new BytesRef(bytes);
    }

    /**
     * Returns the {@link ByteBuffer} representation of the specified {@link BytesRef}.
     *
     * @param bytesRef the {@link BytesRef}
     * @return the {@link ByteBuffer} representation of  {@code bytesRef}
     */
    public static ByteBuffer byteBuffer(BytesRef bytesRef) {
        byte[] bytes = bytesRef.bytes;
        return ByteBuffer.wrap(bytes, bytesRef.offset, bytesRef.offset + bytesRef.length);
    }

} public class TaskQueue {

    private static final Logger logger = LoggerFactory.getLogger(TaskQueue.class);

    private BlockingExecutor[] pools;

    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    /**
     * Returns a new {@link TaskQueue}.
     *
     * @param numThreads the number of executor threads
     * @param queuesSize the max number of tasks in each thread queue before blocking
     */
    public TaskQueue(int numThreads, int queuesSize) {
        if (numThreads > 0) {
            pools = new BlockingExecutor[numThreads];
            for (int i = 0; i < numThreads; i++) {
                pools[i] = new BlockingExecutor(1,
                                                queuesSize,
                                                Long.MAX_VALUE,
                                                TimeUnit.DAYS,
                                                0,
                                                TimeUnit.NANOSECONDS,
                                                null);
                pools[i].submit(new Runnable() {
                    @Override
                    public void run() {
                        logger.debug("Task queue starts");
                    }
                });
            }
        }
    }

    /**
     * Submits a non value-returning task for asynchronous execution.
     *
     * The specified identifier is used to choose the thread executor where the task will be queued. The selection and
     * load balancing is based in the {@link #hashCode()} of this identifier.
     *
     * @param id the identifier of the task used to choose the thread executor where the task will be queued for
     * asynchronous execution
     * @param task the task to be queued for asynchronous execution
     * @return a future for the submitted task
     */
    public Future<?> submitAsynchronous(Object id, Runnable task) {
        if (pools == null) {
            task.run();
            return null;
        } else {
            lock.readLock().lock();
            try {
                int i = Math.abs(id.hashCode() % pools.length);
                return pools[i].submit(task);
            } catch (Exception e) {
                logger.error("Task queue submission failed", e);
                throw new FhirIndexException(e);
            } finally {
                lock.readLock().unlock();
            }
        }
    }

    /**
     * Submits a non value-returning task for synchronous execution. It waits for all synchronous tasks to be
     * completed.
     *
     * @param task a task to be executed synchronously
     */
    public void submitSynchronous(Runnable task) {
        if (pools == null) {
            task.run();
        } else {
            lock.writeLock().lock();
            try {
                await();
                task.run();
            } finally {
                lock.writeLock().unlock();
            }
        }
    }

    /**
     * Await for task completion.
     */
    public void await() {
        if (pools != null) {
            lock.writeLock().lock();
            try {
                Future<?>[] futures = new Future<?>[pools.length];
                for (int i = 0; i < pools.length; i++) {
                    Future<?> future = pools[i].submit(() -> {
                    });
                    futures[i] = future;
                }
                for (Future<?> future : futures) {
                    future.get();
                }
            } catch (InterruptedException e) {
                logger.error("Task queue await interrupted", e);
                throw new FhirIndexException(e);
            } catch (ExecutionException e) {
                logger.error("Task queue await failed", e);
                throw new FhirIndexException(e);
            } finally {
                lock.writeLock().unlock();
            }
        }
    }

    /**
     * Shutdowns this task.
     */
    public void shutdown() {
        if (pools != null) {
            lock.writeLock().lock();
            try {
                for (BlockingExecutor pool : pools) {
                    pool.shutdown();
                }
            } finally {
                lock.writeLock().unlock();
            }
        }
    }

}
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashSet;
import java.util.NavigableSet;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.PartitionColumns;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadOrderGroup;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.filter.RowFilter.Expression;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.index.Index.Indexer;
import org.apache.cassandra.index.Index.Searcher;
import org.apache.cassandra.index.transactions.IndexTransaction.Type;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.OpOrder.Group;
import org.apache.lucene.document.DateTools.Resolution;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.Term;
import org.apache.lucene.queryparser.classic.ParseException;
import org.apache.lucene.queryparser.classic.QueryParser;
import org.apache.lucene.search.CachingWrapperQuery;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.Sort;
import org.hl7.fhir.instance.model.api.IBaseResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ca.uhn.fhir.rest.annotation.Search;
import io.puntanegra.fhir.index.cache.SearchCache;
import io.puntanegra.fhir.index.cache.SearchCacheEntry;
import io.puntanegra.fhir.index.cache.SearchCacheUpdater;
import io.puntanegra.fhir.index.config.IndexOptions;
import io.puntanegra.fhir.index.lucene.LuceneDocumentIterator;
import io.puntanegra.fhir.index.lucene.LuceneService;
import io.puntanegra.fhir.index.mapper.FhirMapper;
import io.puntanegra.fhir.index.mapper.KeyMapper;
import io.puntanegra.fhir.index.mapper.PartitionMapper;
import io.puntanegra.fhir.index.mapper.TokenMapper;
import io.puntanegra.fhir.index.util.ByteBufferUtils;
import io.puntanegra.fhir.index.util.TaskQueue;
/*
 * Licensed to STRATIO (C) under one or more contributor license agreements.
 * See the NOTICE file distributed with this work for additional information
 * regarding copyright ownership.  The STRATIO (C) licenses this file
 * to you 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 io.puntanegra.fhir.index;




/**
 * Service that implement the logic behind the index. It reads the Cassandra's
 * index configuration and passes to {@link LuceneService} to create the index.
 * <br>
 * It also creates the mappers required to process any CRUD operation on the
 * {@link SSTable}
 * 
 * @author Jorge L. Middleton {@literal <jorge.middleton@gmail.com>}
 *
 */
public class FhirIndexService {

	protected static final Logger logger = LoggerFactory.getLogger(FhirIndexService.class);

	public ColumnFamilyStore table;
	public IndexMetadata config;
	public CFMetaData metadata;
	public String name;

	public LuceneService lucene;
	public TaskQueue queue;
	public IndexOptions indexOptions;
	public TokenMapper tokenMapper;
	public PartitionMapper partitionMapper;
	public FhirMapper fhirMapper;
	public boolean mapsMultiCells;

	public SearchCache searchCache;
	private KeyMapper keyMapper;

	public FhirIndexService() {
	}

	public void build(ColumnFamilyStore cfStore, IndexMetadata cfg) {
		table = cfStore;
		config = cfg;
		metadata = table.metadata;

		CFMetaData metadata = table.metadata;
		name = config.name;

		// Setup cache, index and write queue
		indexOptions = new IndexOptions(metadata, config);

		String mbeanName = String.format("com.stratio.cassandra.lucene:type=Lucene,keyspace=%s,table=%s,index=%s",
				metadata.ksName, metadata.cfName, name);

		searchCache = new SearchCache(metadata, indexOptions.searchCacheSize);
		lucene = new LuceneService();
		lucene.init(name, mbeanName, indexOptions.path, indexOptions.search.defaultAnalyzer,
				indexOptions.refreshSeconds, indexOptions.ramBufferMB, indexOptions.maxMergeMB,
				indexOptions.maxCachedMB, searchCache::invalidate);
		queue = new TaskQueue(indexOptions.indexingThreads, indexOptions.indexingQueuesSize);

		// Setup mappers
		fhirMapper = new FhirMapper(indexOptions.search);
		tokenMapper = new TokenMapper();
		partitionMapper = new PartitionMapper(metadata);
		keyMapper = new KeyMapper(metadata);
	}

	public String getName() {
		return name;
	}

	/**
	 * Commits the pending changes.
	 */
	public final void commit() {
		queue.submitSynchronous(lucene::commit);
	}

	/**
	 * Deletes all the index contents.
	 */
	public final void truncate() {
		queue.submitSynchronous(lucene::truncate);
	}

	/**
	 * Completely deletes the index.
	 */
	public void delete() {
		queue.shutdown();
		lucene.delete();
	}

	/**
	 * Upserts the specified {@link Row} to Lucene Index. <br>
	 * Based on the {@link Row} information, it creates a Lucene
	 * {@link Document} and calls the upsert on the {@link LuceneService}.
	 *
	 * @param key
	 *            the partition key
	 * @param row
	 *            the row to be upserted
	 */
	public void upsert(DecoratedKey key, Row row) {
		queue.submitAsynchronous(key, () -> document(key, row).ifPresent(document -> {
			Term term = term(key, row);
			lucene.upsert(term, document);
		}));
	}

	/**
	 * Deletes the partition identified by the specified key.
	 *
	 * @param key
	 *            the partition key
	 * @param row
	 *            the row to be deleted
	 */
	public void delete(DecoratedKey key, Row row) {
		queue.submitAsynchronous(key, () -> {
			Term term = term(key, row);
			lucene.delete(term);
		});
	}

	/**
	 * Deletes the partition identified by the specified key.
	 *
	 * @param key
	 *            the partition key
	 */
	public void delete(DecoratedKey key) {
		queue.submitAsynchronous(key, () -> {
			Term term = term(key);
			lucene.delete(term);
		});
	}

	/**
	 * This method is invoked when a CQL query is executed.
	 * 
	 * @param command
	 * @return
	 */
	// TODO: add support for ORDER BY. i.e.: ORDER BY family:asc, given:desc
	public Searcher searcher(ReadCommand command) {
		// Parse search
		String expression = expression(command);
		// Search search = SearchBuilder.fromJson(expression).build();
		Sort sort = null;// new Sort(new SortField("family",
							// SortField.Type.STRING));

		// Try luck with cache
		Optional<SearchCacheEntry> optional = searchCache.get(expression, command);
		if (optional.isPresent()) {
			logger.debug("Search cache hits");
			SearchCacheEntry entry = optional.get();
			Query query = entry.getQuery();
			ScoreDoc after = entry.getScoreDoc();
			SearchCacheUpdater cacheUpdater = entry.updater();