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

@@ -223,7 +223,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);
         }
 
@@ -272,10 +272,6 @@ private void validateIndexTarget(TableMetadata table, IndexMetadata.Kind kind, I
         if ((kind == IndexMetadata.Kind.CUSTOM) && !SchemaConstants.isValidName(target.column.toString()))
             throw ire(INVALID_CUSTOM_INDEX_TARGET, target.column, SchemaConstants.NAME_LENGTH);
 
-        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())
@@ -322,7 +318,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

@@ -163,7 +163,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);
     }
 
@@ -172,7 +172,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.bti.BtiFormat;
+import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -102,6 +104,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
+                       + BtiFormat.NAME.length()
+                       + tableIdLength
+                       + separatorLength * 3;
+        return addedLength;
+    }
+
     @Override
     public int hashCode()
     {
@@ -183,7 +227,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

@@ -41,7 +41,7 @@
  */
 public class Component
 {
-    public static final char separator = '-';
+    public static final char SEPARATOR = '-';
 
     /**
      * WARNING: Be careful while changing the names or string representation of the enum
@@ -230,8 +230,8 @@ public boolean isValidFor(Descriptor descriptor)
     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

@@ -46,7 +46,7 @@
 import org.apache.cassandra.utils.Pair;
 
 import static com.google.common.base.Preconditions.checkNotNull;
-import static org.apache.cassandra.io.sstable.Component.separator;
+import static org.apache.cassandra.io.sstable.Component.SEPARATOR;
 import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
 
 /**
@@ -179,7 +179,7 @@ public File tmpFileForStreaming(Component component)
 
     public String filenameFor(Component component)
     {
-        return prefix + separator + component.name();
+        return prefix + SEPARATOR + component.name();
     }
 
     public File fileFor(Component component)
@@ -213,9 +213,9 @@ public File baseFile()
 
     private void appendFileName(StringBuilder buff)
     {
-        buff.append(version).append(separator);
+        buff.append(version).append(SEPARATOR);
         buff.append(id.toString());
-        buff.append(separator).append(version.format.name());
+        buff.append(SEPARATOR).append(version.format.name());
     }
 
     public String baseFileUri()
@@ -237,7 +237,7 @@ public String relativeFilenameFor(Component component)
         }
 
         appendFileName(buff);
-        buff.append(separator).append(component.name());
+        buff.append(SEPARATOR).append(component.name());
         return buff.toString();
     }
 

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,12 +43,15 @@
 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.index.sasi.SASIIndex;
 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;
 
@@ -58,6 +64,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.
      */
@@ -109,20 +116,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

@@ -70,14 +70,21 @@ public final class SchemaConstants
     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;
+
+     /**
      * The longest permissible KS or CF name.
      *
      * Before CASSANDRA-16956, we used to care about not having the entire path longer than 255 characters because of
      * Windows support but this limit is by implementing CASSANDRA-16956 not in effect anymore.
      *
      * Note: This extended to 222 for CNDB tenant specific keyspaces.
+     * 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

@@ -275,11 +275,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()
@@ -1364,26 +1365,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

@@ -880,7 +880,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);
         }
 
@@ -1147,7 +1147,7 @@ public void testHasAvailableSpaceSumming()
 
     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)