1.1.1 Changes

build.gradle

@@ -33,9 +33,7 @@ dependencies {
  compile group: 'org.apache.lucene', name: 'lucene-core', version: luceneVersion
  compile group: 'org.apache.lucene', name: 'lucene-spatial', version: luceneVersion
  compile group: 'org.apache.lucene', name: 'lucene-analyzers-common', version: luceneVersion
- compile("org.apache.lucene:lucene-queryparser:$luceneVersion") {
- exclude group: 'org.apache.lucene', module: 'lucene-sandbox'
- }
+ compile("org.apache.lucene:lucene-queryparser:$luceneVersion")
  compile("org.apache.lucene:lucene-join:$luceneVersion")
  compile('joda-time:joda-time:2.3')
  compile group: 'commons-collections', name: 'commons-collections', version: '3.2.2'
@@ -56,9 +54,9 @@ dependencies {
  compile('org.javolution:javolution:5.3.1')
 
  //testing dependencies
- testCompile group: 'junit', name: 'junit', version: '4.11'
+ testCompile group: 'junit', name: 'junit', version: '4.12'
 
- testCompile('com.datastax.cassandra:cassandra-driver-core:2.1.8') {
+ testCompile('com.datastax.cassandra:cassandra-driver-core:2.1.10.1') {
  exclude group: 'org.apache.cassandra', module: 'cassandra-all'
  exclude group: 'log4j'
  exclude group: 'io.netty'
@@ -98,6 +96,8 @@ shadowJar {
  include(dependency('org.apache.lucene:lucene-queries'))
  include(dependency('org.apache.lucene:lucene-queryparser'))
  include(dependency('org.apache.lucene:lucene-join'))
+ include(dependency('org.apache.lucene:lucene-sandbox'))
+ include(dependency('jakarta-regexp:jakarta-regexp'))
  include(dependency('joda-time:joda-time'))
  include(dependency('commons-collections:commons-collections'))
  include(dependency('org.javolution:javolution'))
@@ -109,7 +109,7 @@ task copyJarToCCM(dependsOn: shadowJar) << {
  if (project.hasProperty('cassandraVersion')) {
  copy {
  from('build/libs') {
- include('*-all.jar')
+  include('*-all.jar')
  }
  into System.getProperty('user.home') + '/.ccm/repository/' + project.property('cassandraVersion') + '/lib'
  }
@@ -126,7 +126,7 @@ task copySamples(type: Copy) {
 
 task copyJar(type: Copy) {
  dependsOn([shadowJar, copySamples])
- from ('build/libs'){
+ from('build/libs') {
  include('*-all.jar')
  }
  into 'build/dist/stargate/libs'
@@ -158,6 +158,20 @@ signing {
  sign configurations.archives
 }
 
+task test1(type: Test) {
+ systemProperty "pagesize", "5"
+}
+
+task test2(type: Test) {
+ systemProperty "pagesize", "10"
+}
+
+task test3(type: Test) {
+ systemProperty "pagesize", "1000"
+}
+
+//This can be removed once we move to Parameterized tests
+test.dependsOn(test1, test2, test3)
 
 uploadArchives {
  repositories {

docs/source/indexoptions.rst

@@ -91,12 +91,14 @@ Read lucene docs for explanation on each of them.
 
  * StandardAnalyzer
  * WhitespaceAnalyzer
+ * StopAnalyzer
  * SimpleAnalyzer
- * KeywordAnalyzer 
+ * KeywordAnalyzer
+ * CaseInsensitiveKeywordAnalyzer - Same as Lucene's KeywordAnalyzer but case-insensitive
 
 Custom Analyzers
 ^^^^^^^^^^^^^^^^
-Custom analyzers may be specified using the fully qualified class name. Lucene 5.5 custom analyzers are required.
+Custom analyzers or other Lucene Analyzers may be specified using the fully qualified class name. Lucene 5.5 custom analyzers are required.
 
 Norms 
 ^^^^^^

gradle.properties

@@ -1,3 +1,2 @@
 ossrhUsername=
 ossrhPassword=
-

src/main/java/com/tuplejump/stargate/IndexEventHandler.java

@@ -26,7 +26,7 @@ public void onEvent(IndexEntryEvent event, long sequence, boolean endOfBatch) th
  if ((sequence % numberOfConsumers) == ordinal) {
  ByteBuffer rowkeyBuffer = event.getRowKey();
  ColumnFamily columnFamily = event.getColumnFamily();
- final RowIndexSupport rowIndexSupport = indexingService.support.get(columnFamily.metadata().cfName);
+ final RowIndexSupport rowIndexSupport = indexingService.support.get(columnFamily.metadata().ksAndCFName);
  try {
  rowIndexSupport.indexRow(rowkeyBuffer, columnFamily);
  } catch (Exception e) {

src/main/java/com/tuplejump/stargate/IndexingService.java

@@ -23,6 +23,7 @@
 import com.lmax.disruptor.dsl.Disruptor;
 import com.tuplejump.stargate.cassandra.RowIndexSupport;
 import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.utils.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -40,7 +41,7 @@
 public class IndexingService {
  protected static final Logger logger = LoggerFactory.getLogger(Stargate.class);
  ExecutorService executorService;
- Map<String, RowIndexSupport> support;
+ Map<Pair<String, String>, RowIndexSupport> support;
 
  IndexEntryEvent.Factory eventFactory = new IndexEntryEvent.Factory();
  int numWorkers = Math.max(4, Runtime.getRuntime().availableProcessors());
@@ -74,7 +75,7 @@ public IndexingService(AtomicLong reads, AtomicLong writes) {
  }
 
  public void register(RowIndexSupport rowIndexSupport) {
- this.support.put(rowIndexSupport.tableMapper.cfMetaData.cfName, rowIndexSupport);
+ this.support.put(rowIndexSupport.tableMapper.cfMetaData.ksAndCFName, rowIndexSupport);
  }
 
 

src/main/java/com/tuplejump/stargate/RowIndex.java

@@ -20,7 +20,10 @@
 import com.tuplejump.stargate.cassandra.RowIndexSupport;
 import com.tuplejump.stargate.cassandra.SearchSupport;
 import com.tuplejump.stargate.cassandra.TableMapper;
-import com.tuplejump.stargate.lucene.*;
+import com.tuplejump.stargate.lucene.Constants;
+import com.tuplejump.stargate.lucene.LuceneUtils;
+import com.tuplejump.stargate.lucene.Options;
+import com.tuplejump.stargate.lucene.SearcherCallback;
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.db.ColumnFamily;
@@ -30,16 +33,15 @@
 import org.apache.cassandra.db.index.PerRowSecondaryIndex;
 import org.apache.cassandra.db.index.SecondaryIndexSearcher;
 import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.concurrent.OpOrder;
-import org.apache.commons.collections.map.LRUMap;
 import org.apache.lucene.index.Term;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -68,8 +70,6 @@ public class RowIndex extends PerRowSecondaryIndex {
  boolean nearRealTime = false;
  protected volatile long latest;
 
- public Map<String, IndexEntryCollector> collectorMap = Collections.synchronizedMap(new LRUMap(10));
-
  public TableMapper getTableMapper() {
  return tableMapper;
  }
@@ -132,6 +132,9 @@ private void waitForIndexBuilt() {
 
  @Override
  public void init() {
+ if (!(StorageService.getPartitioner() instanceof Murmur3Partitioner)) {
+ throw new RuntimeException("Stargate currently only supports Mumur3Partitioner");
+ }
  writeLock.lock();
  final Boolean isInfoLoggingEnabled = logger.isInfoEnabled();
  try {

src/main/java/com/tuplejump/stargate/Stargate.java

@@ -21,8 +21,8 @@
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.gms.*;
-import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -80,7 +80,7 @@ public void register(RowIndexSupport rowIndexSupport) {
  }
 
  public long index(ByteBuffer rowKey, ColumnFamily columnFamily) {
- final RowIndexSupport rowIndexSupport = indexingService.support.get(columnFamily.metadata().cfName);
+ final RowIndexSupport rowIndexSupport = indexingService.support.get(columnFamily.metadata().ksAndCFName);
  try {
  rowIndexSupport.indexRow(rowKey, columnFamily);
  } catch (Exception e) {
@@ -147,16 +147,16 @@ public void onRestart(InetAddress endpoint, EndpointState state) {
  public String[] allIndexes() {
  String[] allIndexes = new String[indexingService.support.size()];
  int i = 0;
- for (Map.Entry<String, RowIndexSupport> entry : indexingService.support.entrySet()) {
+ for (Map.Entry<Pair<String, String>, RowIndexSupport> entry : indexingService.support.entrySet()) {
  RowIndexSupport rowIndexSupport = entry.getValue();
- allIndexes[i++] = rowIndexSupport.indexContainer.indexName();
+ allIndexes[i++] = entry.getKey().left + "." + rowIndexSupport.indexContainer.indexName();
  }
  return allIndexes;
  }
 
  @Override
- public String[] indexShards(String indexName) {
- RowIndexSupport indexSupport = getRowIndexSupportByIndexName(indexName);
+ public String[] indexShards(String keyspaceName, String indexName) {
+ RowIndexSupport indexSupport = getRowIndexSupportByIndexName(keyspaceName, indexName);
  if (indexSupport != null && indexSupport.indexContainer instanceof PerVNodeIndexContainer) {
  PerVNodeIndexContainer indexContainer = (PerVNodeIndexContainer) indexSupport.indexContainer;
  Set<Range<Token>> indexShards = indexContainer.indexers.keySet();
@@ -171,34 +171,37 @@ public String[] indexShards(String indexName) {
  }
 
  @Override
- public String describeIndex(String indexName) throws IOException {
- RowIndexSupport indexSupport = getRowIndexSupportByIndexName(indexName);
+ public String describeIndex(String keyspaceName, String indexName) throws IOException {
+ RowIndexSupport indexSupport = getRowIndexSupportByIndexName(keyspaceName, indexName);
  if (indexSupport != null) {
  return indexSupport.getOptions().describeAsJson();
  }
  return null;
  }
 
- private RowIndexSupport getRowIndexSupportByIndexName(String indexName) {
- for (Map.Entry<String, RowIndexSupport> entry : indexingService.support.entrySet()) {
+ private RowIndexSupport getRowIndexSupportByIndexName(String keyspaceName, String indexName) {
+ for (Map.Entry<Pair<String, String>, RowIndexSupport> entry : indexingService.support.entrySet()) {
  RowIndexSupport rowIndexSupport = entry.getValue();
- if (rowIndexSupport.indexContainer.indexName().equalsIgnoreCase(indexName)) return rowIndexSupport;
+ if (entry.getKey().left.equalsIgnoreCase(keyspaceName) &&
+ rowIndexSupport.indexContainer.indexName().equalsIgnoreCase(indexName)) {
+ return rowIndexSupport;
+ }
  }
  return null;
  }
 
  @Override
- public long indexLiveSize(String indexName) {
- RowIndexSupport indexSupport = getRowIndexSupportByIndexName(indexName);
+ public long indexLiveSize(String keyspaceName, String indexName) {
+ RowIndexSupport indexSupport = getRowIndexSupportByIndexName(keyspaceName, indexName);
  if (indexSupport != null) {
  return indexSupport.indexContainer.liveSize();
  }
  return 0;
  }
 
  @Override
- public long indexSize(String indexName) {
- RowIndexSupport indexSupport = getRowIndexSupportByIndexName(indexName);
+ public long indexSize(String keyspaceName, String indexName) {
+ RowIndexSupport indexSupport = getRowIndexSupportByIndexName(keyspaceName, indexName);
  if (indexSupport != null) {
  return indexSupport.indexContainer.size();
  }

src/main/java/com/tuplejump/stargate/StargateMBean.java

@@ -27,13 +27,13 @@ public interface StargateMBean {
 
  public String[] allIndexes();
 
- public String[] indexShards(String indexName);
+ public String[] indexShards(String keyspaceName, String indexName);
 
- public String describeIndex(String indexName) throws IOException;
+ public String describeIndex(String keyspaceName, String indexName) throws IOException;
 
- public long indexSize(String indexName);
+ public long indexSize(String keyspaceName, String indexName);
 
- public long indexLiveSize(String indexName);
+ public long indexLiveSize(String keyspaceName, String indexName);
 
  public long writeGeneration();
 

src/main/java/com/tuplejump/stargate/cassandra/CassandraUtils.java

@@ -26,7 +26,9 @@
 import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.service.StorageService;
 import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper;
 import org.apache.lucene.document.FieldType;
@@ -39,6 +41,8 @@
  * Utilities to read Cassandra configuration
  */
 public class CassandraUtils {
+ public static final Token MINIMUM_TOKEN = StorageService.getPartitioner().getMinimumToken();
+ public static final Long MINIMUM_TOKEN_VALUE = (Long) MINIMUM_TOKEN.getTokenValue();
 
  public static String[] getDataDirs() throws IOException, ConfigurationException {
  return DatabaseDescriptor.getAllDataFileLocations();
@@ -54,6 +58,10 @@ public static Options getOptions(String columnName, ColumnFamilyStore baseCfs, S
  }
  }
 
+ public static boolean isMinimumToken(Token token) {
+ return MINIMUM_TOKEN.compareTo(token) == 0;
+ }
+
  public static Options getOptions(Properties mapping, ColumnFamilyStore baseCfs, String colName) {
  Map<String, NumericConfig> numericFieldOptions = new HashMap<>();
  Map<String, FieldType> fieldDocValueTypes = new TreeMap<>();

src/main/java/com/tuplejump/stargate/cassandra/ResultMapper.java

@@ -48,19 +48,21 @@ public class ResultMapper {
  public final boolean showScore;
  public final TableMapper tableMapper;
  public final SearchSupport searchSupport;
+ public final boolean reverseSort;
 
- public ResultMapper(TableMapper tableMapper, SearchSupport searchSupport, ExtendedFilter filter, IndexEntryCollector collector, boolean showScore) throws Exception {
+ public ResultMapper(TableMapper tableMapper, SearchSupport searchSupport, ExtendedFilter filter, IndexEntryCollector collector, boolean showScore, boolean reverseSort) throws Exception {
  this.tableMapper = tableMapper;
  this.searchSupport = searchSupport;
  this.filter = filter;
  this.collector = collector;
  this.limit = filter.currentLimit();
  this.showScore = showScore;
+ this.reverseSort = reverseSort;
  }
 
 
- public Map<CellName, ColumnFamily> fetchRangeSlice(Collection<IndexEntry> entries, DecoratedKey dk) {
- return getCellNameColumnFamilyMap(dk, getColumnSlices(entries));
+ public Map<CellName, ColumnFamily> fetchRangeSlice(Collection<IndexEntry> entries, DecoratedKey dk, boolean reversed) {
+ return getCellNameColumnFamilyMap(dk, getColumnSlices(entries), reversed);
  }
 
 
@@ -77,7 +79,7 @@ private ColumnSlice[] getColumnSlices(Collection<IndexEntry> entries) {
  ColumnSlice[] columnSlices = new ColumnSlice[entries.size()];
  int i = 0;
  for (IndexEntry entry : entries) {
- Composite start = tableMapper.start(entry.clusteringKey);
+ Composite start = tableMapper.start(entry.clusteringKey());
  Composite end = tableMapper.end(start);
  ColumnSlice columnSlice = new ColumnSlice(start, end);
  columnSlices[i++] = columnSlice;
@@ -86,14 +88,14 @@ private ColumnSlice[] getColumnSlices(Collection<IndexEntry> entries) {
  return columnSlices;
  }
 
- public Map<CellName, ColumnFamily> fetchPagedRangeSlice(Collection<IndexEntry> entries, DecoratedKey dk, int pageSize) {
- return getCellNameColumnFamilyMap(dk, getPagedColumnSlices(dk, entries, pageSize));
+ public Map<CellName, ColumnFamily> fetchPagedRangeSlice(Collection<IndexEntry> entries, DecoratedKey dk, int pageSize, boolean reversed) {
+ return getCellNameColumnFamilyMap(dk, getPagedColumnSlices(dk, entries, pageSize), reversed);
  }
 
  private ColumnSlice[] getPagedColumnSlices(DecoratedKey dk, Collection<IndexEntry> entries, int pageSize) {
  ArrayList<ColumnSlice> columnSlices = new ArrayList<>(Math.min(entries.size(), pageSize));
  for (IndexEntry entry : entries) {
- CellName cellName = entry.clusteringKey;
+ CellName cellName = entry.clusteringKey();
  if (!filter.columnFilter(dk.getKey()).maySelectPrefix(tableMapper.table.getComparator(), cellName.start())) {
  continue;
  }
@@ -108,8 +110,8 @@ private ColumnSlice[] getPagedColumnSlices(DecoratedKey dk, Collection<IndexEntr
  return columnSlices.toArray(new ColumnSlice[columnSlices.size()]);
  }
 
- private Map<CellName, ColumnFamily> getCellNameColumnFamilyMap(DecoratedKey dk, ColumnSlice[] columnSlices) {
- SliceQueryFilter sliceQueryFilter = new SliceQueryFilter(columnSlices, false, Integer.MAX_VALUE);
+ private Map<CellName, ColumnFamily> getCellNameColumnFamilyMap(DecoratedKey dk, ColumnSlice[] columnSlices, boolean reversed) {
+ SliceQueryFilter sliceQueryFilter = new SliceQueryFilter(columnSlices, reversed, Integer.MAX_VALUE);
  QueryFilter queryFilter = new QueryFilter(dk, tableMapper.table.name, sliceQueryFilter, filter.timestamp);
  ColumnFamily columnFamily = tableMapper.table.getColumnFamily(queryFilter);
  return tableMapper.getRows(columnFamily);