CNDB-12503 remove column name restrictions in SAI (#1539)

Trying to create an SAI or other index on a column, which name is either
long or has
non-alphanumeric characters, fails, while the table was successfully
created with such name. This happens if an index name is not provided.
The reason for this limitation is that the column name is used for
generating the default index name, which is used in the file names.

Fixes riptano/cndb#12503, fixes
riptano/cndb#12609

This PR removes the check for column names during an index creation.
Thus it allows to create indexes on existing columns with any names
including qualified names and long names.

1. The check was preventing qualified column names with characters,
which cannot be used in file names. This was never an issue as the index
name construction from table and column names already removes any
non-alphanumeric or non-underscore characters.
2. The check was preventing long column names. This limitation is
removed by
  - Adding calculation of maximum amount characters, which can be used in
index name, so it will allow to construct file names without exceeding
the limit of 255 chars (255 is added as a constant).
  - Then the index name is truncated. 
3. The uniqueness of index names is already fixed by existing
implementation, which adds counter prefix when needed.

This change to generated index name does not affect existing indexes as
the existing name from `IndexMetadata` is always serialized to JSON,
thus avoiding to re-generating the name.

This fix required to fix CQL tester, since it was assuming that the
restriction is in place. So column name regex pattern matcher is fixed
in `createIndex` to allow non-restricted column names. There might still
be possible corner cases on some column names used for generating index
names, which will not work with `createIndex`. In such case `execute`
should be used.

It also includes several fixes in affected files:
- Improve the name and usage of constants.
- Fix warnings with minor improvements in affected files.

Author: k-rus

src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java

@@ -180,7 +180,7 @@ public Keyspaces apply(Keyspaces schema)
             long indexesOnAllTables = StreamSupport.stream(Keyspace.all().spliterator(), false).flatMap(ks -> ks.getColumnFamilyStores().stream())
                                                    .flatMap(ks -> ks.indexManager.listIndexes().stream())
                                                    .map(i -> i.getIndexMetadata().getIndexClassName())
-                                                   .filter(otherClassName -> className.equals(otherClassName)).count();
+                                                   .filter(className::equals).count();
             guardRails.totalThreshold.guard(indexesOnAllTables + 1, indexDescription, false, state);
         }
 
@@ -224,10 +224,6 @@ private void validateIndexTarget(TableMetadata table, IndexMetadata.Kind kind, I
         if (null == column)
             throw ire("Column '%s' doesn't exist", target.column);
 
-        if ((kind == IndexMetadata.Kind.CUSTOM) && !SchemaConstants.isValidName(target.column.toString()))
-            throw ire("Column '%s' is longer than the permissible name length of %d characters or" +
-                      " contains non-alphanumeric-underscore characters", target.column, SchemaConstants.NAME_LENGTH);
-
         if (column.type.referencesDuration())
         {
             if (column.type.isCollection())
@@ -275,7 +271,7 @@ private String generateIndexName(KeyspaceMetadata keyspace, List<IndexTarget> ta
     {
         String baseName = targets.size() == 1
                         ? IndexMetadata.generateDefaultIndexName(tableName, targets.get(0).column)
-                        : IndexMetadata.generateDefaultIndexName(tableName);
+                        : IndexMetadata.generateDefaultIndexName(tableName, null);
         return keyspace.findAvailableIndexName(baseName);
     }
 

src/java/org/apache/cassandra/db/lifecycle/LogRecord.java

@@ -153,7 +153,7 @@ public static LogRecord makeAbort(long updateTime)
 
     public static LogRecord make(Type type, SSTable table)
     {
-        String absoluteTablePath = table.descriptor.baseFileUri() + Component.separator;
+        String absoluteTablePath = table.descriptor.baseFileUri() + Component.SEPARATOR;
         return make(type, getExistingFiles(absoluteTablePath), table.getComponentSize(), absoluteTablePath);
     }
 
@@ -162,7 +162,7 @@ public static Map<SSTable, LogRecord> make(Type type, Iterable<SSTableReader> ta
         // contains a mapping from sstable absolute path (everything up until the 'Data'/'Index'/etc part of the filename) to the sstable
         Map<String, SSTable> absolutePaths = new HashMap<>();
         for (SSTableReader table : tables)
-            absolutePaths.put(table.descriptor.baseFileUri() + Component.separator, table);
+            absolutePaths.put(table.descriptor.baseFileUri() + Component.SEPARATOR, table);
 
         // maps sstable base file name to the actual files on disk
         Map<String, List<File>> existingFiles = getExistingFiles(absolutePaths.keySet());

src/java/org/apache/cassandra/index/sai/disk/format/Version.java

@@ -36,6 +36,8 @@
 import org.apache.cassandra.index.sai.disk.v6.V6OnDiskFormat;
 import org.apache.cassandra.index.sai.disk.v7.V7OnDiskFormat;
 import org.apache.cassandra.index.sai.utils.TypeUtil;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
+import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -103,6 +105,48 @@ public static Version latest()
         return LATEST;
     }
 
+    /**
+     * Calculates the maximum allowed length for SAI index names to ensure generated filenames
+     * do not exceed the system's filename length limit (defined in {@link SchemaConstants#FILENAME_LENGTH}).
+     * This accounts for all additional components in the filename.
+     */
+    public static int calculateIndexNameAllowedLength()
+    {
+        int addedLength = getAddedLengthFromDescriptorAndVersion();
+        assert addedLength < SchemaConstants.FILENAME_LENGTH;
+        return SchemaConstants.FILENAME_LENGTH - addedLength;
+    }
+
+    /**
+     * Calculates the length of the added prefixes and suffixes from Descriptor constructor
+     * and {@link Version#stargazerFileNameFormat}.
+     *
+     * @return the length of the added prefixes and suffixes
+     */
+    private static int getAddedLengthFromDescriptorAndVersion()
+    {
+        // Prefixes and suffixes constructed by Version.stargazerFileNameFormat
+        int versionNameLength = latest().toString().length();
+        // room for up to 999 generations
+        int generationLength = 3 + SAI_SEPARATOR.length();
+        int addedLength = SAI_DESCRIPTOR.length()
+                          + versionNameLength
+                          + generationLength
+                          + IndexComponentType.PRIMARY_KEY_BLOCK_OFFSETS.representation.length()
+                          + SAI_SEPARATOR.length() * 3
+                          + EXTENSION.length();
+
+        // Prefixes from Descriptor constructor
+        int separatorLength = 1;
+        int indexVersionLength = 2;
+        int tableIdLength = 28;
+        addedLength += indexVersionLength
+                       + SSTableFormat.Type.BTI.name().length()
+                       + tableIdLength
+                       + separatorLength * 3;
+        return addedLength;
+    }
+
     @Override
     public int hashCode()
     {
@@ -184,7 +228,7 @@ default String format(IndexComponentType indexComponentType, IndexContext indexC
      */
     public static Optional<ParsedFileName> tryParseFileName(String filename)
     {
-        if (!filename.endsWith(".db"))
+        if (!filename.endsWith(EXTENSION))
             return Optional.empty();
 
         // For flexibility, we handle both "full" filename, of the form "<descriptor>-SAI+....db", or just the component

src/java/org/apache/cassandra/io/sstable/Component.java

@@ -34,7 +34,7 @@
  */
 public class Component
 {
-    public static final char separator = '-';
+    public static final char SEPARATOR = '-';
 
     final static EnumSet<Type> TYPES = EnumSet.allOf(Type.class);
 
@@ -164,8 +164,8 @@ public static Component parse(String name)
     public File getFile(String absolutePath)
     {
         File ret;
-        if (absolutePath.lastIndexOf(separator) != (absolutePath.length() - 1))
-            ret = new File(PathUtils.getPath(absolutePath + separator + name));
+        if (absolutePath.lastIndexOf(SEPARATOR) != (absolutePath.length() - 1))
+            ret = new File(PathUtils.getPath(absolutePath + SEPARATOR + name));
         else
             ret = new File(PathUtils.getPath(absolutePath + name));
 

src/java/org/apache/cassandra/io/sstable/Descriptor.java

@@ -39,7 +39,7 @@
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.UUIDGen;
 
-import static org.apache.cassandra.io.sstable.Component.separator;
+import static org.apache.cassandra.io.sstable.Component.SEPARATOR;
 
 /**
  * A SSTable is described by the keyspace and column family it contains data
@@ -108,7 +108,7 @@ public Descriptor(Version version, File directory, String ksname, String cfname,
 
         hashCode = Objects.hashCode(version, this.directory, id, ksname, cfname, formatType);
 
-        filenamePart = version.toString() + separator + id + separator + formatType.name;
+        filenamePart = version.toString() + SEPARATOR + id + SEPARATOR + formatType.name;
         String locationURI = directory.toUri().toString();
         if (!locationURI.endsWith(java.io.File.separator))
             locationURI = locationURI + java.io.File.separatorChar;
@@ -145,7 +145,7 @@ public String tmpFilenameForStreaming(Component component)
 
     public String filenameFor(Component component)
     {
-        return baseFilename() + separator + component.name();
+        return baseFilename() + SEPARATOR + component.name();
     }
 
     public File fileFor(Component component)
@@ -163,9 +163,9 @@ public String baseFilename()
 
     private void appendFileName(StringBuilder buff)
     {
-        buff.append(version).append(separator);
+        buff.append(version).append(SEPARATOR);
         buff.append(id.toString());
-        buff.append(separator).append(formatType.name);
+        buff.append(SEPARATOR).append(formatType.name);
     }
 
     public String baseFileUri()
@@ -186,7 +186,7 @@ public String relativeFilenameFor(Component component)
             buff.append(directory.name()).append(File.pathSeparator());
         }
 
-        return buff.append(filenamePart).append(separator).append(component.name()).toString();
+        return buff.append(filenamePart).append(SEPARATOR).append(component.name()).toString();
     }
 
     public SSTableFormat getFormat()

src/java/org/apache/cassandra/schema/IndexMetadata.java

@@ -20,7 +20,10 @@
 
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
@@ -40,11 +43,14 @@
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.index.internal.CassandraIndex;
 import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.index.sai.disk.format.Version;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.UUIDSerializer;
 
+import javax.annotation.Nullable;
+
 import static org.apache.cassandra.schema.SchemaConstants.PATTERN_NON_WORD_CHAR;
 import static org.apache.cassandra.schema.SchemaConstants.isValidName;
 
@@ -57,6 +63,7 @@ public final class IndexMetadata
 
     public static final Serializer serializer = new Serializer();
 
+    static final String INDEX_POSTFIX = "_idx";
     /**
      * A mapping of user-friendly index names to their fully qualified index class names.
      */
@@ -106,20 +113,55 @@ public static IndexMetadata fromIndexTargets(List<IndexTarget> targets,
         return new IndexMetadata(name, newOptions, kind);
     }
 
-    public static String generateDefaultIndexName(String table, ColumnIdentifier column)
+    /**
+     * Generates a default index name from the table and column names.
+     * Characters other than alphanumeric and underscore are removed.
+     * Long index names are truncated to fit the length allowing constructing filenames.
+     *
+     * @param table  the table name
+     * @param column the column identifier. Can be null if the index is not column specific.
+     * @return the generated index name
+     */
+    public static String generateDefaultIndexName(String table, @Nullable ColumnIdentifier column)
     {
-        return PATTERN_NON_WORD_CHAR.matcher(table + '_' + column.toString() + "_idx").replaceAll("");
+        String indexNameUncleaned = table;
+        if (column != null)
+            indexNameUncleaned += '_' + column.toString();
+        String indexNameUntrimmed = PATTERN_NON_WORD_CHAR.matcher(indexNameUncleaned).replaceAll("");
+        String indexNameTrimmed = indexNameUntrimmed
+                                  .substring(0,
+                                             Math.min(calculateGeneratedIndexNameMaxLength(),
+                                                      indexNameUntrimmed.length()));
+        return indexNameTrimmed + INDEX_POSTFIX;
     }
 
-    public static String generateDefaultIndexName(String table)
+    /**
+     * Calculates the maximum length of the generated index name to fit file names.
+     * It includes the generated suffixes in account.
+     * The calculation depends on how index implements file names construciton from index names.
+     * This needs to be addressed, see CNDB-13240.
+     *
+     * @return the allowed length of the generated index name
+     */
+    private static int calculateGeneratedIndexNameMaxLength()
     {
-        return PATTERN_NON_WORD_CHAR.matcher(table + '_' + "idx").replaceAll("");
+        // Speculative assumption that uniqueness breaker will fit into 999.
+        // The value is used for trimming the index name if needed.
+        // Introducing validation of index name length is TODO for CNDB-13198.
+        int uniquenessSuffixLength = 4;
+        int indexNameAddition = uniquenessSuffixLength + INDEX_POSTFIX.length();
+        int allowedIndexNameLength = Version.calculateIndexNameAllowedLength();
+
+        assert allowedIndexNameLength >= indexNameAddition : "cannot happen with current implementation as allowedIndexNameLength is approximately 255 - ~76. However, allowedIndexNameLength was " + allowedIndexNameLength + " and  indexNameAddition was " + indexNameAddition;
+
+        return allowedIndexNameLength - indexNameAddition;
     }
 
     public void validate(TableMetadata table)
     {
         if (!isValidName(name, true))
-            throw new ConfigurationException("Illegal index name " + name);
+            throw new ConfigurationException(String.format("Index name must not be empty, or contain non-alphanumeric-underscore characters (got \"%s\")",
+                                                           name));
 
         if (kind == null)
             throw new ConfigurationException("Index kind is null for index " + name);

src/java/org/apache/cassandra/schema/SchemaConstants.java

@@ -58,6 +58,11 @@ public final class SchemaConstants
     /* virtual keyspace names */
     public static final Set<String> VIRTUAL_KEYSPACE_NAMES = ImmutableSet.of(SCHEMA_VIRTUAL_KEYSPACE_NAME, SYSTEM_VIEWS_KEYSPACE_NAME);
 
+    /**
+     * Longest acceptable file name. Longer names lead to file write or read errors.
+     */
+    public static final int FILENAME_LENGTH = 255;
+
     /**
      * longest permissible KS or CF name.  Our main concern is that filename not be more than 255 characters;
      * the filename will contain both the KS and CF names. Since non-schema-name components only take up
@@ -66,8 +71,10 @@ public final class SchemaConstants
      *
      * Note: This extended to 222 for CNDB tenant specific keyspaces. The windows restriction is not valid here
      * because CNDB does not support windows.
+     * 222 is maximum filename length of 255 chars minus a separator char and
+     * 32 chars for table UUID.
      */
-    public static final int NAME_LENGTH = 222;
+    public static final int NAME_LENGTH = FILENAME_LENGTH - 32 - 1;
 
     // 59adb24e-f3cd-3e02-97f0-5b395827453f
     public static final UUID emptyVersion;

test/unit/org/apache/cassandra/cql3/CQLTester.java

@@ -234,11 +234,12 @@ public abstract class CQLTester
     public static final List<ProtocolVersion> PROTOCOL_VERSIONS = new ArrayList<>(ProtocolVersion.SUPPORTED.size());
 
     private static final String CREATE_INDEX_NAME_REGEX = "(\\s*(\\w*|\"\\w*\")\\s*)";
+    private static final String CREATE_INDEX_NAME_QUOTED_REGEX = "(\\s*(\\w*|\"[^\"]*\")\\s*)";
     private static final String CREATE_INDEX_REGEX = String.format("\\A\\s*CREATE(?:\\s+CUSTOM)?\\s+INDEX" +
                                                                    "(?:\\s+IF\\s+NOT\\s+EXISTS)?\\s*" +
                                                                    "%s?\\s*ON\\s+(%<s\\.)?%<s\\s*" +
-                                                                   "(\\((?:\\s*\\w+\\s*\\()?%<s\\))?",
-                                                                   CREATE_INDEX_NAME_REGEX);
+                                                                   "(\\((?:\\s*\\w+\\s*\\()?%s\\))?",
+                                                                   CREATE_INDEX_NAME_REGEX, CREATE_INDEX_NAME_QUOTED_REGEX);
     private static final Pattern CREATE_INDEX_PATTERN = Pattern.compile(CREATE_INDEX_REGEX, Pattern.CASE_INSENSITIVE);
 
     public static final NettyOptions IMMEDIATE_CONNECTION_SHUTDOWN_NETTY_OPTIONS = new NettyOptions()
@@ -1061,26 +1062,27 @@ protected static Pair<String, String> getCreateIndexName(String keyspace, String
             keyspace = parsedKeyspace;
 
         String index = matcher.group(2);
+        boolean isQuotedGeneratedIndexName = false;
         if (Strings.isNullOrEmpty(index))
         {
             String table = matcher.group(7);
             if (Strings.isNullOrEmpty(table))
                 throw new IllegalArgumentException("Table name should be specified: " + formattedQuery);
 
             String column = matcher.group(9);
+            isQuotedGeneratedIndexName = ParseUtils.isQuoted(column, '\"');
 
             String baseName = Strings.isNullOrEmpty(column)
-                              ? IndexMetadata.generateDefaultIndexName(table)
+                              ? IndexMetadata.generateDefaultIndexName(table, null)
                               : IndexMetadata.generateDefaultIndexName(table, new ColumnIdentifier(column, true));
 
             KeyspaceMetadata ks = Schema.instance.getKeyspaceMetadata(keyspace);
             assertNotNull(ks);
             index = ks.findAvailableIndexName(baseName);
         }
-
         index = ParseUtils.isQuoted(index, '\"')
                 ? ParseUtils.unDoubleQuote(index)
-                : index.toLowerCase();
+                : isQuotedGeneratedIndexName ? index : index.toLowerCase();
 
         return Pair.create(keyspace, index);
     }

test/unit/org/apache/cassandra/db/DirectoriesTest.java

@@ -744,7 +744,7 @@ private void testDirectoriesSymlinksHelper(boolean oldStyle) throws IOException
         for (TableMetadata tm : CFM)
         {
             Path keyspacedir = Files.createDirectories(ddir2.resolve(tm.keyspace));
-            String tabledir = tm.name + (oldStyle ? "" : Component.separator + tm.id.toHexString());
+            String tabledir = tm.name + (oldStyle ? "" : Component.SEPARATOR + tm.id.toHexString());
             Files.createSymbolicLink(keyspacedir.resolve(tabledir), symlinktarget);
         }
 
@@ -806,7 +806,7 @@ public void testDataDirectoriesIterator() throws IOException
 
     private String getNewFilename(TableMetadata tm, boolean oldStyle)
     {
-        return tm.keyspace + File.pathSeparator() + tm.name + (oldStyle ? "" : Component.separator + tm.id.toHexString()) + "/na-1-big-Data.db";
+        return tm.keyspace + File.pathSeparator() + tm.name + (oldStyle ? "" : Component.SEPARATOR + tm.id.toHexString()) + "/na-1-big-Data.db";
     }
 
     private List<Directories.DataDirectoryCandidate> getWriteableDirectories(DataDirectory[] dataDirectories, long writeSize)