Prevent too long table names not fitting file names

The length of table names was not controlled. This is likely due to
confusion between validation methods with similar names. As result
creating tables with too long names led to the too long file name
exceptions during table creations.

This commit adds a validation of table name lengths to avoid the too
long file name errors. The validation length is based on how the table
name is used to create file/directory names, and needs to be exact to
prevent the too long file name exception, but allow all other table
names, which didn't lead to the too long file name exception. This
length limit is different from the existing name length limit of 48
characters used by common validation functions.

Thus, this commit moves out the length validation from the validation
methods into a separate length validation method, so the errors on
names are more specific. The non-length validation methods combined
into a single method, which checks for empty names and valid characters.

New constants are added for the length limits.

Table name related code are moved into methods in TableMetadata class,
so their semantics are more clear and to allow reuse, e.g., in
asserting the table name length constant.
Tests are added for the long table names and non-alphanumeric names.

Keyspace name validation function is now shared between two classes and
a unit test of it is added.

Patch by Ruslan Fomkin; reviewed by Piotr Kołaczkowski, Dmitry
Konstantinov, Maxwell Guo for CASSANDRA-20389

Author: k-rus

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0.19
+ * Prevent too long table names not fitting file names (CASSANDRA-20389)
  * Fix IndexOutOfBoundsException in sstablemetadata tool when a range tombstone is a max clustering value (CASSANDRA-20855)
  * Update Jackson to 2.19.2 (CASSANDRA-20848)
  * Update commons-lang3 to 3.18.0 (CASSANDRA-20849)

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

@@ -35,6 +35,8 @@
 import org.apache.cassandra.transport.Event.SchemaChange;
 import org.apache.cassandra.transport.messages.ResultMessage;
 
+import static org.apache.cassandra.schema.KeyspaceMetadata.validateKeyspaceName;
+
 abstract public class AlterSchemaStatement implements CQLStatement.SingleKeyspaceCqlStatement, SchemaTransformation
 {
     protected final String keyspaceName; // name of the keyspace affected by the statement
@@ -103,7 +105,7 @@ public ResultMessage execute(QueryState state, boolean locally)
         if (null != keyspace && keyspace.isVirtual())
             throw ire("Virtual keyspace '%s' is not user-modifiable", keyspaceName);
 
-        validateKeyspaceName();
+        validateKeyspaceName(keyspaceName, AlterSchemaStatement::ire);
 
         KeyspacesDiff diff = MigrationManager.announce(this, locally);
 
@@ -126,16 +128,6 @@ public ResultMessage execute(QueryState state, boolean locally)
         return new ResultMessage.SchemaChange(schemaChangeEvent(diff));
     }
 
-    private void validateKeyspaceName()
-    {
-        if (!SchemaConstants.isValidName(keyspaceName))
-        {
-            throw ire("Keyspace name must not be empty, more than %d characters long, " +
-                      "or contain non-alphanumeric-underscore characters (got '%s')",
-                      SchemaConstants.NAME_LENGTH, keyspaceName);
-        }
-    }
-
     private void grantPermissionsOnResource(IResource resource, AuthenticatedUser user)
     {
         try

src/java/org/apache/cassandra/db/Directories.java

@@ -89,6 +89,7 @@ public class Directories
     public static final String SNAPSHOT_SUBDIR = "snapshots";
     public static final String TMP_SUBDIR = "tmp";
     public static final String SECONDARY_INDEX_NAME_SEPARATOR = ".";
+    public static final String TABLE_DIRECTORY_NAME_SEPARATOR = "-";
 
     /**
      * The directories used to store keyspaces data.
@@ -196,14 +197,11 @@ public Directories(final TableMetadata metadata, DataDirectory[] paths)
         this.metadata = metadata;
         this.paths = paths;
         ImmutableMap.Builder<Path, DataDirectory> canonicalPathsBuilder = ImmutableMap.builder();
-        String tableId = metadata.id.toHexString();
-        int idx = metadata.name.indexOf(SECONDARY_INDEX_NAME_SEPARATOR);
-        String cfName = idx >= 0 ? metadata.name.substring(0, idx) : metadata.name;
-        String indexNameWithDot = idx >= 0 ? metadata.name.substring(idx) : null;
+        String indexNameWithDot = metadata.getIndexNameWithDot();
 
         this.dataPaths = new File[paths.length];
         // If upgraded from version less than 2.1, use existing directories
-        String oldSSTableRelativePath = join(metadata.keyspace, cfName);
+        String oldSSTableRelativePath = join(metadata.keyspace, metadata.getTableName());
         for (int i = 0; i < paths.length; ++i)
         {
             // check if old SSTable directory exists
@@ -216,7 +214,7 @@ public Directories(final TableMetadata metadata, DataDirectory[] paths)
         {
             canonicalPathsBuilder = ImmutableMap.builder();
             // use 2.1+ style
-            String newSSTableRelativePath = join(metadata.keyspace, cfName + '-' + tableId);
+            String newSSTableRelativePath = join(metadata.keyspace, metadata.getTableDirectoryName());
             for (int i = 0; i < paths.length; ++i)
             {
                 File dataPath = new File(paths[i].location, newSSTableRelativePath);

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

@@ -21,7 +21,6 @@
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.util.*;
-import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import com.google.common.base.Objects;
@@ -42,17 +41,16 @@
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.UUIDSerializer;
 
+import static org.apache.cassandra.schema.SchemaConstants.PATTERN_NON_WORD_CHAR;
+import static org.apache.cassandra.schema.SchemaConstants.isValidCharsName;
+
 /**
  * An immutable representation of secondary index metadata.
  */
 public final class IndexMetadata
 {
     private static final Logger logger = LoggerFactory.getLogger(IndexMetadata.class);
 
-    private static final Pattern PATTERN_NON_WORD_CHAR = Pattern.compile("\\W");
-    private static final Pattern PATTERN_WORD_CHARS = Pattern.compile("\\w+");
-
-
     public static final Serializer serializer = new Serializer();
 
     public enum Kind
@@ -94,11 +92,6 @@ public static IndexMetadata fromIndexTargets(List<IndexTarget> targets,
         return new IndexMetadata(name, newOptions, kind);
     }
 
-    public static boolean isNameValid(String name)
-    {
-        return name != null && !name.isEmpty() && PATTERN_WORD_CHARS.matcher(name).matches();
-    }
-
     public static String generateDefaultIndexName(String table, ColumnIdentifier column)
     {
         return PATTERN_NON_WORD_CHAR.matcher(table + "_" + column.toString() + "_idx").replaceAll("");
@@ -111,7 +104,8 @@ public static String generateDefaultIndexName(String table)
 
     public void validate(TableMetadata table)
     {
-        if (!isNameValid(name))
+        // TODO: address validating the length by CASSANDRA-20445
+        if (!isValidCharsName(name))
             throw new ConfigurationException("Illegal index name " + name);
 
         if (kind == null)

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

@@ -34,6 +34,7 @@
 import org.apache.cassandra.cql3.functions.UDFunction;
 import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.exceptions.RequestValidationException;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.schema.Functions.FunctionsDiff;
 import org.apache.cassandra.schema.Tables.TablesDiff;
@@ -44,12 +45,30 @@
 import static java.lang.String.format;
 
 import static com.google.common.collect.Iterables.any;
+import static org.apache.cassandra.schema.SchemaConstants.TABLE_NAME_LENGTH;
 
 /**
  * An immutable representation of keyspace metadata (name, params, tables, types, and functions).
  */
 public final class KeyspaceMetadata implements SchemaElement
 {
+    /**
+     * Validates the keyspace name for valid characters and correct length.
+     * Throws an exception if it's invalid.
+     *
+     * @param keyspaceName     The name of the keyspace to validate
+     * @param exceptionBuilder The exception constructor to throw if validation fails
+     */
+    public static <T extends RequestValidationException> void validateKeyspaceName(String keyspaceName, java.util.function.Function<String, T> exceptionBuilder)
+    {
+        if (!SchemaConstants.isValidCharsName(keyspaceName))
+            throw exceptionBuilder.apply(format("Keyspace name must not be empty and must contain alphanumeric or underscore characters only (got \"%s\")",
+                                                keyspaceName));
+        if (keyspaceName.length() > SchemaConstants.NAME_LENGTH)
+            throw exceptionBuilder.apply(format("Keyspace name must not be more than %d characters long (got %d characters for \"%s\")",
+                                                TABLE_NAME_LENGTH, keyspaceName.length(), keyspaceName));
+    }
+
     public enum Kind
     {
         REGULAR, VIRTUAL
@@ -301,16 +320,8 @@ public String toCqlString(boolean withInternals, boolean ifNotExists)
 
     public void validate()
     {
-        if (!SchemaConstants.isValidName(name))
-        {
-            throw new ConfigurationException(format("Keyspace name must not be empty, more than %s characters long, "
-                                                    + "or contain non-alphanumeric-underscore characters (got \"%s\")",
-                                                    SchemaConstants.NAME_LENGTH,
-                                                    name));
-        }
-
+        validateKeyspaceName(name, ConfigurationException::new);
         params.validate(name);
-
         tablesAndViews().forEach(TableMetadata::validate);
 
         Set<String> indexNames = new HashSet<>();

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

@@ -28,9 +28,12 @@
 
 import org.apache.cassandra.db.Digest;
 
+import static org.apache.cassandra.db.Directories.TABLE_DIRECTORY_NAME_SEPARATOR;
+
 public final class SchemaConstants
 {
     public static final Pattern PATTERN_WORD_CHARS = Pattern.compile("\\w+");
+    public static final Pattern PATTERN_NON_WORD_CHAR = Pattern.compile("\\W");
 
     public static final String SYSTEM_KEYSPACE_NAME = "system";
     public static final String SCHEMA_KEYSPACE_NAME = "system_schema";
@@ -58,14 +61,37 @@ public final class SchemaConstants
      */
     public static final int NAME_LENGTH = 48;
 
+    /**
+     * Longest acceptable file name. Longer names lead to too long file name error.
+     */
+    public static final int FILENAME_LENGTH = 255;
+
+    /**
+     * Length of a table uuid as a hex string.
+     */
+    public static final int TABLE_UUID_AS_HEX_LENGTH = 32;
+
+    /**
+     * Longest acceptable table name, so it can be used in a directory
+     * name constructed with a suffix of a table id and a separator.
+     */
+    public static final int TABLE_NAME_LENGTH = FILENAME_LENGTH - TABLE_UUID_AS_HEX_LENGTH - TABLE_DIRECTORY_NAME_SEPARATOR.length();
+
     // 59adb24e-f3cd-3e02-97f0-5b395827453f
     public static final UUID emptyVersion;
 
     public static final List<String> LEGACY_AUTH_TABLES = Arrays.asList("credentials", "users", "permissions");
 
-    public static boolean isValidName(String name)
+    /**
+     * Validates that a name is not empty and contains only alphanumeric characters or
+     * underscore, so it can be used in file or directory names.
+     *
+     * @param name the name to check
+     * @return whether the non-empty name contains only valid characters
+     */
+    public static boolean isValidCharsName(String name)
     {
-        return name != null && !name.isEmpty() && name.length() <= NAME_LENGTH && PATTERN_WORD_CHARS.matcher(name).matches();
+        return name != null && !name.isEmpty() && PATTERN_WORD_CHARS.matcher(name).matches();
     }
 
     static

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

@@ -48,7 +48,12 @@
 import static java.lang.String.format;
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toSet;
-import static org.apache.cassandra.schema.IndexMetadata.isNameValid;
+import static org.apache.cassandra.db.Directories.SECONDARY_INDEX_NAME_SEPARATOR;
+import static org.apache.cassandra.db.Directories.TABLE_DIRECTORY_NAME_SEPARATOR;
+import static org.apache.cassandra.schema.KeyspaceMetadata.validateKeyspaceName;
+import static org.apache.cassandra.schema.SchemaConstants.FILENAME_LENGTH;
+import static org.apache.cassandra.schema.SchemaConstants.TABLE_NAME_LENGTH;
+import static org.apache.cassandra.schema.SchemaConstants.isValidCharsName;
 
 @Unmetered
 public class TableMetadata implements SchemaElement
@@ -415,11 +420,9 @@ public boolean hasStaticColumns()
 
     public void validate()
     {
-        if (!isNameValid(keyspace))
-            except("Keyspace name must not be empty, more than %s characters long, or contain non-alphanumeric-underscore characters (got \"%s\")", SchemaConstants.NAME_LENGTH, keyspace);
+        validateKeyspaceName(keyspace, this::prepareConfigurationException);
 
-        if (!isNameValid(name))
-            except("Table name must not be empty, more than %s characters long, or contain non-alphanumeric-underscore characters (got \"%s\")", SchemaConstants.NAME_LENGTH, name);
+        validateTableName();
 
         params.validate();
 
@@ -447,6 +450,17 @@ public void validate()
         indexes.validate(this);
     }
 
+    private void validateTableName()
+    {
+        if (!isValidCharsName(name))
+            except("Table name must not be empty or not contain non-alphanumeric-underscore characters (got \"%s\")", name);
+
+        if (name.length() > TABLE_NAME_LENGTH)
+            except("Table name must not be more than %d characters long (got %d characters for \"%s\")", TABLE_NAME_LENGTH, name.length(), name);
+
+        assert getTableDirectoryName().length() <= FILENAME_LENGTH : String.format("Generated directory name for a table of %d characters doesn't fit the max filename legnth of %s. This unexpectedly wasn't prevented by check of the table name length, %d, to fit %d characters (got table name \"%s\" and generated directory name \"%s\"", getTableDirectoryName().length(), FILENAME_LENGTH, name.length(), TABLE_NAME_LENGTH, name, getTableDirectoryName());
+    }
+
     /**
      * To support backward compatibility with thrift super columns in the C* 3.0+ storage engine, we encode said super
      * columns as a CQL {@code map<blob, blob>}. To ensure the name of this map did not conflict with any other user
@@ -543,7 +557,41 @@ public ClusteringComparator partitionKeyAsClusteringComparator()
     public String indexTableName(IndexMetadata info)
     {
         // TODO simplify this when info.index_name is guaranteed to be set
-        return name + Directories.SECONDARY_INDEX_NAME_SEPARATOR + info.name;
+        return name + SECONDARY_INDEX_NAME_SEPARATOR + info.name;
+    }
+
+    /**
+     * Returns the table part of the index table name or the entire table name
+     * if not an index table.
+     * @return table name part
+     */
+    public String getTableName()
+    {
+        int idx = name.indexOf(SECONDARY_INDEX_NAME_SEPARATOR);
+        return idx >= 0 ? name.substring(0, idx) : name;
+    }
+
+    /**
+     * Generates a directory name for the table by using table part of
+     * the (index) table name and table id.
+     * @return directory name
+     */
+    public String getTableDirectoryName()
+    {
+        return getTableName() + TABLE_DIRECTORY_NAME_SEPARATOR + id.toHexString();
+    }
+
+    /**
+     * Returns the index name from the name of an index table
+     * including the dot prefixing the index name, see {@link #indexTableName}.
+     * If not an index table, returns null.
+     * @return index name prefixed with dot prefix or null
+     */
+    @Nullable
+    public String getIndexNameWithDot()
+    {
+        int idx = name.indexOf(SECONDARY_INDEX_NAME_SEPARATOR);
+        return idx >= 0 ? name.substring(idx) : null;
     }
 
     /**
@@ -601,9 +649,14 @@ public TableMetadata withUpdatedUserType(UserType udt)
         return builder.build();
     }
 
+    private ConfigurationException prepareConfigurationException(String format, Object... args)
+    {
+        return new ConfigurationException(keyspace + '.' + name + ": " + format(format, args));
+    }
+
     protected void except(String format, Object... args)
     {
-        throw new ConfigurationException(keyspace + "." + name + ": " + format(format, args));
+        throw prepareConfigurationException(format, args);
     }
 
     @Override

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

@@ -48,6 +48,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 
+import org.apache.commons.lang3.ArrayUtils;
 import org.junit.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -573,6 +574,14 @@ public void flush(String keyspace)
             store.forceBlockingFlush();
     }
 
+    public void flush(String keyspaceName, String table1, String... tables)
+    {
+        tables = ArrayUtils.add(tables, table1);
+        Keyspace keyspace = Keyspace.open(keyspaceName);
+        for (String tableName : tables)
+            keyspace.getColumnFamilyStore(tableName).forceBlockingFlush();
+    }
+
     public void disableCompaction(String keyspace)
     {
         ColumnFamilyStore store = getCurrentColumnFamilyStore(keyspace);