Merge branch 'cassandra-3.11' into trunk

# Conflicts:
#	src/java/org/apache/cassandra/index/sasi/SASIIndex.java
#	test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java

CHANGES.txt

@@ -255,6 +255,7 @@
 
 
 3.11.3
+ * Validate supported column type with SASI analyzer (CASSANDRA-13669)
  * Remove BTree.Builder Recycler to reduce memory usage (CASSANDRA-13929)
  * Reduce nodetool GC thread count (CASSANDRA-14475)
  * Fix New SASI view creation during Index Redistribution (CASSANDRA-14055)

src/java/org/apache/cassandra/index/sasi/SASIIndex.java

@@ -124,6 +124,9 @@ public class SASIIndex implements Index, INotificationConsumer
         CompactionManager.instance.submitIndexBuild(new SASIIndexBuilder(baseCfs, toRebuild));
     }
 
+    /**
+     * Called via reflection at {@link IndexMetadata#validateCustomIndexOptions}
+     */
     public static Map<String, String> validateOptions(Map<String, String> options, TableMetadata metadata)
     {
         if (!(metadata.partitioner instanceof Murmur3Partitioner))
@@ -143,7 +146,7 @@ public class SASIIndex implements Index, INotificationConsumer
         if (target.left.isPartitionKey())
             throw new ConfigurationException("partition key columns are not yet supported by SASI");
 
-        IndexMode.validateAnalyzer(options);
+        IndexMode.validateAnalyzer(options, target.left);
 
         IndexMode mode = IndexMode.getMode(target.left, options);
         if (mode.mode == Mode.SPARSE)

src/java/org/apache/cassandra/index/sasi/analyzer/AbstractAnalyzer.java

@@ -42,6 +42,14 @@ public abstract class AbstractAnalyzer implements Iterator<ByteBuffer>
 
     public abstract void reset(ByteBuffer input);
 
+    /**
+     * Test whether the given validator is compatible with the underlying analyzer.
+     *
+     * @param validator
+     * @return
+     */
+    public abstract boolean isCompatibleWith(AbstractType<?> validator);
+
     /**
      * @return true if current analyzer provides text tokenization, false otherwise.
      */

src/java/org/apache/cassandra/index/sasi/analyzer/DelimiterAnalyzer.java

@@ -37,10 +37,10 @@ import org.apache.cassandra.utils.AbstractIterator;
 public class DelimiterAnalyzer extends AbstractAnalyzer
 {
 
-    private static final Map<AbstractType<?>,Charset> VALID_ANALYZABLE_TYPES = new HashMap<AbstractType<?>,Charset>()
+    private static final Map<AbstractType<?>, Charset> VALID_ANALYZABLE_TYPES = new HashMap<AbstractType<?>, Charset>()
     {{
-            put(UTF8Type.instance, StandardCharsets.UTF_8);
+        put(UTF8Type.instance, StandardCharsets.UTF_8);
-            put(AsciiType.instance, StandardCharsets.US_ASCII);
+        put(AsciiType.instance, StandardCharsets.US_ASCII);
     }};
 
     private char delimiter;
@@ -105,4 +105,10 @@ public class DelimiterAnalyzer extends AbstractAnalyzer
     {
         return true;
     }
+
+    @Override
+    public boolean isCompatibleWith(AbstractType<?> validator)
+    {
+        return VALID_ANALYZABLE_TYPES.containsKey(validator);
+    }
 }

src/java/org/apache/cassandra/index/sasi/analyzer/NoOpAnalyzer.java

@@ -51,4 +51,10 @@ public class NoOpAnalyzer extends AbstractAnalyzer
         this.input = input;
         this.hasNext = true;
     }
+
+    @Override
+    public boolean isCompatibleWith(AbstractType<?> validator)
+    {
+        return true;
+    }
 }

src/java/org/apache/cassandra/index/sasi/analyzer/NonTokenizingAnalyzer.java

@@ -123,4 +123,10 @@ public class NonTokenizingAnalyzer extends AbstractAnalyzer
             builder = builder.add("to_lower", new BasicResultFilters.LowerCase());
         return builder.build();
     }
+
+    @Override
+    public boolean isCompatibleWith(AbstractType<?> validator)
+    {
+        return VALID_ANALYZABLE_TYPES.contains(validator);
+    }
 }

src/java/org/apache/cassandra/index/sasi/analyzer/StandardAnalyzer.java

@@ -23,10 +23,13 @@ import java.io.InputStreamReader;
 import java.io.Reader;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.cassandra.index.sasi.analyzer.filter.*;
 import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.AsciiType;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -38,6 +41,15 @@ import com.carrotsearch.hppc.IntObjectOpenHashMap;
 
 public class StandardAnalyzer extends AbstractAnalyzer
 {
+
+    private static final Set<AbstractType<?>> VALID_ANALYZABLE_TYPES = new HashSet<AbstractType<?>>()
+    {
+        {
+            add(UTF8Type.instance);
+            add(AsciiType.instance);
+        }
+    };
+
     public enum TokenType
     {
         EOF(-1),
@@ -198,4 +210,10 @@ public class StandardAnalyzer extends AbstractAnalyzer
     {
         return true;
     }
+
+    @Override
+    public boolean isCompatibleWith(AbstractType<?> validator)
+    {
+        return VALID_ANALYZABLE_TYPES.contains(validator);
+    }
 }

src/java/org/apache/cassandra/index/sasi/conf/IndexMode.java

@@ -93,20 +93,36 @@ public class IndexMode
         return analyzer;
     }
 
-    public static void validateAnalyzer(Map<String, String> indexOptions) throws ConfigurationException
+    public static void validateAnalyzer(Map<String, String> indexOptions, ColumnMetadata cd) throws ConfigurationException
     {
         // validate that a valid analyzer class was provided if specified
         if (indexOptions.containsKey(INDEX_ANALYZER_CLASS_OPTION))
         {
+            Class<?> analyzerClass;
             try
             {
-                Class.forName(indexOptions.get(INDEX_ANALYZER_CLASS_OPTION));
+                analyzerClass = Class.forName(indexOptions.get(INDEX_ANALYZER_CLASS_OPTION));
             }
             catch (ClassNotFoundException e)
             {
                 throw new ConfigurationException(String.format("Invalid analyzer class option specified [%s]",
                                                                indexOptions.get(INDEX_ANALYZER_CLASS_OPTION)));
             }
+
+            AbstractAnalyzer analyzer;
+            try
+            {
+                analyzer = (AbstractAnalyzer) analyzerClass.newInstance();
+                if (!analyzer.isCompatibleWith(cd.type))
+                    throw new ConfigurationException(String.format("%s does not support type %s",
+                                                                   analyzerClass.getSimpleName(),
+                                                                   cd.type.asCQL3Type()));
+            }
+            catch (InstantiationException | IllegalAccessException e)
+            {
+                throw new ConfigurationException(String.format("Unable to initialize analyzer class option specified [%s]",
+                                                               analyzerClass.getSimpleName()));
+            }
         }
     }
 

test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java

@@ -31,6 +31,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.cql3.CQLTester;
@@ -39,6 +40,7 @@ import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.Term;
@@ -57,6 +59,11 @@ import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.index.sasi.analyzer.AbstractAnalyzer;
+import org.apache.cassandra.index.sasi.analyzer.DelimiterAnalyzer;
+import org.apache.cassandra.index.sasi.analyzer.NoOpAnalyzer;
+import org.apache.cassandra.index.sasi.analyzer.NonTokenizingAnalyzer;
+import org.apache.cassandra.index.sasi.analyzer.StandardAnalyzer;
 import org.apache.cassandra.index.sasi.conf.ColumnIndex;
 import org.apache.cassandra.index.sasi.disk.OnDiskIndexBuilder;
 import org.apache.cassandra.index.sasi.exceptions.TimeQuotaExceededException;
@@ -2409,6 +2416,76 @@ public class SASIIndexTest
         Assert.assertEquals(index.searchMemtable(expression).getCount(), 0);
     }
 
+    @Test
+    public void testAnalyzerValidation()
+    {
+        final String TABLE_NAME = "analyzer_validation";
+        QueryProcessor.executeOnceInternal(String.format("CREATE TABLE %s.%s (" +
+                                                         "  pk text PRIMARY KEY, " +
+                                                         "  ascii_v ascii, " +
+                                                         "  bigint_v bigint, " +
+                                                         "  blob_v blob, " +
+                                                         "  boolean_v boolean, " +
+                                                         "  date_v date, " +
+                                                         "  decimal_v decimal, " +
+                                                         "  double_v double, " +
+                                                         "  float_v float, " +
+                                                         "  inet_v inet, " +
+                                                         "  int_v int, " +
+                                                         "  smallint_v smallint, " +
+                                                         "  text_v text, " +
+                                                         "  time_v time, " +
+                                                         "  timestamp_v timestamp, " +
+                                                         "  timeuuid_v timeuuid, " +
+                                                         "  tinyint_v tinyint, " +
+                                                         "  uuid_v uuid, " +
+                                                         "  varchar_v varchar, " +
+                                                         "  varint_v varint" +
+                                                         ");",
+                                                         KS_NAME,
+                                                         TABLE_NAME));
+
+        Columns regulars = Schema.instance.getTableMetadata(KS_NAME, TABLE_NAME).regularColumns();
+        List<String> allColumns = regulars.stream().map(ColumnMetadata::toString).collect(Collectors.toList());
+        List<String> textColumns = Arrays.asList("text_v", "ascii_v", "varchar_v");
+
+        new HashMap<Class<? extends AbstractAnalyzer>, List<String>>()
+        {{
+            put(StandardAnalyzer.class, textColumns);
+            put(NonTokenizingAnalyzer.class, textColumns);
+            put(DelimiterAnalyzer.class, textColumns);
+            put(NoOpAnalyzer.class, allColumns);
+        }}
+        .forEach((analyzer, supportedColumns) -> {
+            for (String column : allColumns)
+            {
+                String query = String.format("CREATE CUSTOM INDEX ON %s.%s(%s) " +
+                                             "USING 'org.apache.cassandra.index.sasi.SASIIndex' " +
+                                             "WITH OPTIONS = {'analyzer_class': '%s', 'mode':'PREFIX'};",
+                                             KS_NAME, TABLE_NAME, column, analyzer.getName());
+
+                if (supportedColumns.contains(column))
+                {
+                    QueryProcessor.executeOnceInternal(query);
+                }
+                else
+                {
+                    try
+                    {
+                        QueryProcessor.executeOnceInternal(query);
+                        Assert.fail("Expected ConfigurationException");
+                    }
+                    catch (ConfigurationException e)
+                    {
+                        // expected
+                        Assert.assertTrue("Unexpected error message " + e.getMessage(),
+                                          e.getMessage().contains("does not support type"));
+                    }
+                }
+            }
+        });
+    }
+
     private static ColumnFamilyStore loadData(Map<String, Pair<String, Integer>> data, boolean forceFlush)
     {
         return loadData(data, System.currentTimeMillis(), forceFlush);