Merge branch 'cassandra-4.1' into cassandra-5.0

* cassandra-4.1:
  Prevent too long table names not fitting file names

CHANGES.txt

@@ -4,6 +4,7 @@
 Merged from 4.1:
  * Redact security-sensitive information in system_views.settings (CASSANDRA-20856)
 Merged from 4.0:
+ * Prevent too long table names not fitting file names (CASSANDRA-20389)
  * Update Jackson to 2.19.2 (CASSANDRA-20848)
  * Update commons-lang3 to 3.18.0 (CASSANDRA-20849)
  * Add NativeTransportMaxConcurrentConnectionsPerIp to StorageProxyMBean (CASSANDRA-20642)

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

@@ -38,6 +38,8 @@ import org.apache.cassandra.transport.Dispatcher;
 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
@@ -111,7 +113,7 @@ abstract public class AlterSchemaStatement implements CQLStatement.SingleKeyspac
         if (null != keyspace && keyspace.isVirtual())
             throw ire("Virtual keyspace '%s' is not user-modifiable", keyspaceName);
 
-        validateKeyspaceName();
+        validateKeyspaceName(keyspaceName, AlterSchemaStatement::ire);
 
         SchemaTransformationResult result = Schema.instance.transform(this, locally);
 
@@ -134,16 +136,6 @@ abstract public class AlterSchemaStatement implements CQLStatement.SingleKeyspac
         return new ResultMessage.SchemaChange(schemaChangeEvent(result.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);
-        }
-    }
-
     protected void validateDefaultTimeToLive(TableParams params)
     {
         if (params.defaultTimeToLive == 0

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

@@ -60,8 +60,8 @@ public final class CreateIndexStatement extends AlterSchemaStatement
     public static final String CUSTOM_MULTIPLE_COLUMNS = "Only CUSTOM indexes support multiple columns";
     public static final String DUPLICATE_TARGET_COLUMN = "Duplicate column '%s' in index target list";
     public static final String COLUMN_DOES_NOT_EXIST = "Column '%s' doesn't exist";
-    public static final String INVALID_CUSTOM_INDEX_TARGET = "Column '%s' is longer than the permissible name length of %d characters or" +
+    public static final String INVALID_CHARS_CUSTOM_INDEX_TARGET = "Column '%s' contains non-alphanumeric-underscore characters";
-                                                             " contains non-alphanumeric-underscore characters";
+    public static final String TOO_LONG_CUSTOM_INDEX_TARGET = "Column '%s' is longer than the permissible name length of %d characters";
     public static final String COLLECTIONS_WITH_DURATIONS_NOT_SUPPORTED = "Secondary indexes are not supported on collections containing durations";
     public static final String TUPLES_WITH_DURATIONS_NOT_SUPPORTED = "Secondary indexes are not supported on tuples containing durations";
     public static final String DURATIONS_NOT_SUPPORTED = "Secondary indexes are not supported on duration columns";
@@ -206,6 +206,14 @@ public final class CreateIndexStatement extends AlterSchemaStatement
         return ImmutableSet.of();
     }
 
+    private void validateCustomIndexColumnName(String name)
+    {
+        if (!SchemaConstants.isValidCharsName(name))
+            throw ire(INVALID_CHARS_CUSTOM_INDEX_TARGET, name);
+        if (name.length() > SchemaConstants.NAME_LENGTH)
+            throw ire(TOO_LONG_CUSTOM_INDEX_TARGET, name, SchemaConstants.NAME_LENGTH);
+    }
+
     private void validateIndexTarget(TableMetadata table, IndexMetadata.Kind kind, IndexTarget target)
     {
         ColumnMetadata column = table.getColumn(target.column);
@@ -215,8 +223,9 @@ public final class CreateIndexStatement extends AlterSchemaStatement
 
         AbstractType<?> baseType = column.type.unwrap();
 
-        if ((kind == IndexMetadata.Kind.CUSTOM) && !SchemaConstants.isValidName(target.column.toString()))
+        // TODO: this check needs to be removed with CASSANDRA-20235
-            throw ire(INVALID_CUSTOM_INDEX_TARGET, target.column, SchemaConstants.NAME_LENGTH);
+        if ((kind == IndexMetadata.Kind.CUSTOM))
+            validateCustomIndexColumnName(target.column.toString());
 
         if (column.type.referencesDuration())
         {

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

@@ -120,6 +120,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.
@@ -227,14 +228,11 @@ public class Directories
         this.metadata = metadata;
         this.paths = paths;
         ImmutableMap.Builder<Path, DataDirectory> canonicalPathsBuilder = ImmutableMap.builder();
-        String tableId = metadata.id.toHexString();
+        String indexNameWithDot = metadata.getIndexNameWithDot();
-        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;
 
         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
@@ -247,7 +245,7 @@ public class Directories
         {
             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

@@ -22,7 +22,6 @@ import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import com.google.common.base.Objects;
@@ -47,6 +46,9 @@ import org.apache.cassandra.io.util.DataOutputPlus;
 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.
  */
@@ -54,10 +56,6 @@ 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();
 
     /**
@@ -111,11 +109,6 @@ public final class IndexMetadata
         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("");
@@ -128,7 +121,8 @@ public final class IndexMetadata
 
     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

@@ -36,6 +36,7 @@ import org.apache.cassandra.cql3.functions.UDAggregate;
 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.UserFunctions.FunctionsDiff;
 import org.apache.cassandra.schema.Tables.TablesDiff;
@@ -46,12 +47,30 @@ import org.apache.cassandra.service.StorageService;
 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
@@ -317,16 +336,8 @@ public final class KeyspaceMetadata implements SchemaElement
 
     public void validate()
     {
-        if (!SchemaConstants.isValidName(name))
+        validateKeyspaceName(name, ConfigurationException::new);
-        {
-            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));
-        }
-
         params.validate(name, null);
-
         tablesAndViews().forEach(TableMetadata::validate);
 
         Set<String> indexNames = new HashSet<>();

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

@@ -32,12 +32,15 @@ import org.apache.cassandra.db.Digest;
 import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.tracing.TraceKeyspace;
 
+import static org.apache.cassandra.db.Directories.TABLE_DIRECTORY_NAME_SEPARATOR;
+
 /**
  * When adding new String keyspace names here, double check if it needs to be added to PartitionDenylist.canDenylistKeyspace
  */
 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";
@@ -71,14 +74,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

@@ -52,7 +52,6 @@ import org.apache.cassandra.cql3.functions.masking.ColumnMask;
 import org.apache.cassandra.db.Clustering;
 import org.apache.cassandra.db.ClusteringComparator;
 import org.apache.cassandra.db.Columns;
-import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.RegularAndStaticColumns;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -73,7 +72,12 @@ import static com.google.common.collect.Iterables.transform;
 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
@@ -475,11 +479,9 @@ public class TableMetadata implements SchemaElement
 
     public void validate()
     {
-        if (!isNameValid(keyspace))
+        validateKeyspaceName(keyspace, this::prepareConfigurationException);
-            except("Keyspace name must not be empty, more than %s characters long, or contain non-alphanumeric-underscore characters (got \"%s\")", SchemaConstants.NAME_LENGTH, keyspace);
 
-        if (!isNameValid(name))
+        validateTableName();
-            except("Table name must not be empty, more than %s characters long, or contain non-alphanumeric-underscore characters (got \"%s\")", SchemaConstants.NAME_LENGTH, name);
 
         params.validate();
 
@@ -507,6 +509,17 @@ public class TableMetadata implements SchemaElement
         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
@@ -603,7 +616,41 @@ public class TableMetadata implements SchemaElement
     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;
     }
 
     /**
@@ -661,9 +708,14 @@ public class TableMetadata implements SchemaElement
         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
@@ -1722,4 +1774,4 @@ public class TableMetadata implements SchemaElement
 
     }
 
-}
+}

test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java

@@ -74,7 +74,6 @@ import org.apache.cassandra.inject.Injection;
 import org.apache.cassandra.inject.Injections;
 import org.apache.cassandra.inject.InvokePointBuilder;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
-import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.Throwables;
@@ -263,7 +262,7 @@ public class StorageAttachedIndexDDLTest extends SAITester
         assertThatThrownBy(() -> executeNet(String.format("CREATE INDEX ON %%s(\"%s\")" +
                                                           " USING 'sai'", invalidColumn)))
         .isInstanceOf(InvalidQueryException.class)
-        .hasMessage(String.format(CreateIndexStatement.INVALID_CUSTOM_INDEX_TARGET, invalidColumn, SchemaConstants.NAME_LENGTH));
+        .hasMessage(String.format(CreateIndexStatement.INVALID_CHARS_CUSTOM_INDEX_TARGET, invalidColumn));
     }
 
     @Test

test/unit/org/apache/cassandra/schema/CreateTableValidationTest.java

@@ -24,6 +24,10 @@ import org.apache.cassandra.exceptions.InvalidRequestException;
 
 import org.junit.Test;
 
+import static org.apache.cassandra.schema.SchemaConstants.FILENAME_LENGTH;
+import static org.apache.cassandra.schema.SchemaConstants.NAME_LENGTH;
+import static org.apache.cassandra.schema.SchemaConstants.TABLE_NAME_LENGTH;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.junit.Assert.fail;
 
@@ -100,6 +104,42 @@ public class CreateTableValidationTest extends CQLTester
                         "Missing CLUSTERING ORDER for column ck1");
     }
 
+    @Test
+    public void testCreatingTableWithLongName() throws Throwable
+    {
+        int tableIdSuffix = "-1b255f4def2540a60000000000000000".length();
+        String keyspaceName = "k".repeat(NAME_LENGTH);
+        String tableName = "t".repeat(FILENAME_LENGTH - tableIdSuffix);
+        String tooLongTableName = "l".repeat(FILENAME_LENGTH - tableIdSuffix + 1);
+
+        execute(String.format("CREATE KEYSPACE %s with replication = " +
+                              "{ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }",
+                              keyspaceName));
+
+        assertInvalidMessage(String.format("%s.%s: Table name must not be more than %d characters long (got %d characters for \"%2$s\")",
+                                           keyspaceName, tooLongTableName, TABLE_NAME_LENGTH, tooLongTableName.length()),
+                             String.format("CREATE TABLE %s.%s (" +
+                                           "key int PRIMARY KEY," +
+                                           "val int)", keyspaceName, tooLongTableName));
+
+        createTable(String.format("CREATE TABLE %s.%s (" +
+                                  "key int PRIMARY KEY," +
+                                  "val int)", keyspaceName, tableName));
+        execute(String.format("INSERT INTO %s.%s (key,val) VALUES (1,1)", keyspaceName, tableName));
+        assertThat(execute(String.format("SELECT * from %s.%s", keyspaceName, tableName))).hasSize(1);
+        flush(keyspaceName, tableName);
+        assertThat(execute(String.format("SELECT * from %s.%s", keyspaceName, tableName))).hasSize(1);
+    }
+
+    @Test
+    public void testNonAlphanummericTableName() throws Throwable
+    {
+        assertInvalidMessage(String.format("%s.d-3: Table name must not be empty or not contain non-alphanumeric-underscore characters (got \"d-3\")", KEYSPACE),
+                             String.format("CREATE TABLE %s.\"d-3\" (key int PRIMARY KEY, val int)", KEYSPACE));
+        assertInvalidMessage(String.format("%s.    : Table name must not be empty or not contain non-alphanumeric-underscore characters (got \"    \")", KEYSPACE),
+                             String.format("CREATE TABLE %s.\"    \" (key int PRIMARY KEY, val int)", KEYSPACE));
+    }
+
     private void expectedFailure(String statement, String errorMsg)
     {
 

test/unit/org/apache/cassandra/schema/IndexMetadataTest.java

@@ -25,29 +25,8 @@ import org.junit.Test;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
 
-import static org.junit.Assert.assertFalse;
+public class IndexMetadataTest
-import static org.junit.Assert.assertTrue;
+{
-
-public class IndexMetadataTest {
-    
-    @Test
-    public void testIsNameValidPositive()
-    {
-        assertTrue(IndexMetadata.isNameValid("abcdefghijklmnopqrstuvwxyz"));
-        assertTrue(IndexMetadata.isNameValid("ABCDEFGHIJKLMNOPQRSTUVWXYZ"));
-        assertTrue(IndexMetadata.isNameValid("_01234567890"));
-    }
-    
-    @Test
-    public void testIsNameValidNegative()
-    {
-        assertFalse(IndexMetadata.isNameValid(null));
-        assertFalse(IndexMetadata.isNameValid(""));
-        assertFalse(IndexMetadata.isNameValid(" "));
-        assertFalse(IndexMetadata.isNameValid("@"));
-        assertFalse(IndexMetadata.isNameValid("!"));
-    }
-    
     @Test
     public void testGetDefaultIndexName()
     {

test/unit/org/apache/cassandra/schema/MigrationManagerTest.java

@@ -24,6 +24,8 @@ import java.util.Map;
 import java.util.function.Supplier;
 
 import com.google.common.collect.ImmutableMap;
+
+import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.junit.BeforeClass;
 import org.junit.Rule;
 import org.junit.Test;
@@ -51,6 +53,8 @@ import org.apache.cassandra.utils.FBUtilities;
 import static org.apache.cassandra.Util.throwAssert;
 import static org.apache.cassandra.cql3.CQLTester.assertRows;
 import static org.apache.cassandra.cql3.CQLTester.row;
+import static org.apache.cassandra.schema.SchemaConstants.NAME_LENGTH;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -130,13 +134,21 @@ public class MigrationManagerTest
     @Test
     public void testInvalidNames()
     {
-        String[] valid = {"1", "a", "_1", "b_", "__", "1_a"};
+        String[] valid = { "1", "a", "_1", "b_", "__", "1_a" };
         for (String s : valid)
-            assertTrue(SchemaConstants.isValidName(s));
+        {
-
+            assertTrue(SchemaConstants.isValidCharsName(s));
-        String[] invalid = {"b@t", "dash-y", "", " ", "dot.s", ".hidden"};
+            KeyspaceMetadata.validateKeyspaceName(s, InvalidRequestException::new);
+        }
+        String[] invalid = { "b@t", "dash-y", "", " ", "dot.s", ".hidden" };
         for (String s : invalid)
-            assertFalse(SchemaConstants.isValidName(s));
+        {
+            assertFalse(SchemaConstants.isValidCharsName(s));
+            assertThatThrownBy(() -> KeyspaceMetadata.validateKeyspaceName(s, InvalidRequestException::new)).isInstanceOf(InvalidRequestException.class);
+        }
+
+        KeyspaceMetadata.validateKeyspaceName("k".repeat(NAME_LENGTH), InvalidRequestException::new);
+        assertThatThrownBy(() -> KeyspaceMetadata.validateKeyspaceName("k".repeat(NAME_LENGTH + 1), InvalidRequestException::new)).isInstanceOf(InvalidRequestException.class);
     }
 
     @Test
@@ -477,14 +489,22 @@ public class MigrationManagerTest
     }
 
     @Test
-    public void testValidateNullKeyspace() throws Exception
+    public void testTableMetadataBuilderWithInvalidKeyspace()
     {
-        TableMetadata.Builder builder = TableMetadata.builder(null, TABLE1).addPartitionKeyColumn("partitionKey", BytesType.instance);
+        String[][] keyspaces =
-
+        {
-        TableMetadata table1 = builder.build();
+        { null, "Keyspace name must not be empty" },
-        thrown.expect(ConfigurationException.class);
+        { "a@b", "Keyspace name must not be empty and must contain alphanumeric or underscore characters only" },
-        thrown.expectMessage(null + "." + TABLE1 + ": Keyspace name must not be empty");
+        { "k".repeat(NAME_LENGTH), String.format("Keyspace name must not be more than %d characters long", NAME_LENGTH) }
-        table1.validate();
+        };
+        for (String[] keyspace : keyspaces)
+        {
+            TableMetadata.Builder builder = TableMetadata.builder(keyspace[0], TABLE1).addPartitionKeyColumn("partitionKey", BytesType.instance);
+            TableMetadata table1 = builder.build();
+            thrown.expect(ConfigurationException.class);
+            thrown.expectMessage(String.format("%s.%s: %s", keyspace[0], TABLE1, keyspace[1]));
+            table1.validate();
+        }
     }
 
     @Test

test/unit/org/apache/cassandra/schema/ValidationTest.java

@@ -32,19 +32,19 @@ public class ValidationTest
     @Test
     public void testIsNameValidPositive()
     {
-         assertTrue(SchemaConstants.isValidName("abcdefghijklmnopqrstuvwxyz"));
+        assertTrue(SchemaConstants.isValidCharsName("abcdefghijklmnopqrstuvwxyz"));
-         assertTrue(SchemaConstants.isValidName("ABCDEFGHIJKLMNOPQRSTUVWXYZ"));
+        assertTrue(SchemaConstants.isValidCharsName("ABCDEFGHIJKLMNOPQRSTUVWXYZ"));
-         assertTrue(SchemaConstants.isValidName("_01234567890"));
+        assertTrue(SchemaConstants.isValidCharsName("_01234567890"));
     }
     
     @Test
     public void testIsNameValidNegative()
     {
-        assertFalse(SchemaConstants.isValidName(null));
+        assertFalse(SchemaConstants.isValidCharsName(null));
-        assertFalse(SchemaConstants.isValidName(""));
+        assertFalse(SchemaConstants.isValidCharsName(""));
-        assertFalse(SchemaConstants.isValidName(" "));
+        assertFalse(SchemaConstants.isValidCharsName(" "));
-        assertFalse(SchemaConstants.isValidName("@"));
+        assertFalse(SchemaConstants.isValidCharsName("@"));
-        assertFalse(SchemaConstants.isValidName("!"));
+        assertFalse(SchemaConstants.isValidCharsName("!"));
     }
 
     private static Set<String> primitiveTypes =