Merge branch 'cassandra-5.0' into trunk

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

Author: netudima

CHANGES.txt

@@ -308,6 +308,7 @@ Merged from 4.1:
  * Enforce CQL message size limit on multiframe messages (CASSANDRA-20052)
  * Fix race condition in DecayingEstimatedHistogramReservoir during rescale (CASSANDRA-19365)
 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

@@ -46,6 +46,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
 {
     private static final Logger logger = LoggerFactory.getLogger(AlterSchemaStatement.class);
@@ -168,7 +170,8 @@ public ResultMessage execute(QueryState state)
         if (null != keyspace && keyspace.isVirtual())
             throw ire("Virtual keyspace '%s' is not user-modifiable", keyspaceName);
 
-        validateKeyspaceName();
+        validateKeyspaceName(keyspaceName, AlterSchemaStatement::ire);
+
         setExecutionTimestamp(state.getTimestamp());
         // Perform a 'dry-run' attempt to apply the transformation locally before submitting to the CMS. This can save a
         // round trip to the CMS for things syntax errors, but also fail fast for things like configuration errors.
@@ -211,16 +214,6 @@ protected ClusterMetadata commit(ClusterMetadata metadata)
         return Schema.instance.submit(this);
     }
 
-    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" +
-                                                             " contains non-alphanumeric-underscore characters";
+    public static final String INVALID_CHARS_CUSTOM_INDEX_TARGET = "Column '%s' 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";
@@ -220,6 +220,14 @@ Set<String> clientWarnings(KeyspacesDiff diff)
         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);
@@ -229,8 +237,9 @@ private void validateIndexTarget(TableMetadata table, IndexMetadata.Kind kind, I
 
         AbstractType<?> baseType = column.type.unwrap();
 
-        if ((kind == IndexMetadata.Kind.CUSTOM) && !SchemaConstants.isValidName(target.column.toString()))
-            throw ire(INVALID_CUSTOM_INDEX_TARGET, target.column, SchemaConstants.NAME_LENGTH);
+        // TODO: this check needs to be removed with CASSANDRA-20235
+        if ((kind == IndexMetadata.Kind.CUSTOM))
+            validateCustomIndexColumnName(target.column.toString());
 
         if (column.type.referencesDuration())
         {

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

@@ -117,6 +117,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.
@@ -224,14 +225,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
@@ -244,7 +242,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

@@ -28,7 +28,6 @@
 import java.util.TreeSet;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import com.google.common.base.Objects;
@@ -56,6 +55,8 @@
 
 import static org.apache.cassandra.db.TypeSizes.sizeof;
 import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
+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.
@@ -64,10 +65,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();
     public static final MetadataSerializer metadataSerializer = new MetadataSerializer();
 
@@ -117,29 +114,25 @@ public static IndexMetadata fromIndexTargets(List<IndexTarget> targets,
     {
         Map<String, String> newOptions = new HashMap<>(options);
         newOptions.put(IndexTarget.TARGET_OPTION_NAME, targets.stream()
-                                                              .map(target -> target.asCqlString())
+                                                              .map(IndexTarget::asCqlString)
                                                               .collect(Collectors.joining(", ")));
         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("");
+        return PATTERN_NON_WORD_CHAR.matcher(table + '_' + column.toString() + "_idx").replaceAll("");
     }
 
     public static String generateDefaultIndexName(String table)
     {
-        return PATTERN_NON_WORD_CHAR.matcher(table + "_" + "idx").replaceAll("");
+        return PATTERN_NON_WORD_CHAR.matcher(table + "_idx").replaceAll("");
     }
 
     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

@@ -39,6 +39,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.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
@@ -53,6 +54,7 @@
 import static com.google.common.collect.Iterables.any;
 import static java.lang.String.format;
 import static org.apache.cassandra.db.TypeSizes.sizeof;
+import static org.apache.cassandra.schema.SchemaConstants.TABLE_NAME_LENGTH;
 import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
 
 /**
@@ -62,6 +64,23 @@ public final class KeyspaceMetadata implements SchemaElement
 {
     public static final Serializer serializer = new Serializer();
 
+    /**
+     * 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
@@ -214,7 +233,7 @@ public String findAvailableIndexName(String baseName)
     }
 
     /**
-     * find an avaiable index name based on the indexes in target keyspace and indexes collections
+     * find an available index name based on the indexes in target keyspace and indexes collections
      * @param baseName the base name of index
      * @param indexes find out whether there is any conflict with baseName in the indexes
      * @param keyspaceMetadata find out whether there is any conflict with baseName in keyspaceMetadata
@@ -381,14 +400,7 @@ public String toCqlString(boolean withWarnings, boolean withInternals, boolean i
 
     public void validate(ClusterMetadata metadata)
     {
-        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, null, metadata);
         tablesAndViews().forEach(TableMetadata::validate);
 

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

@@ -33,6 +33,7 @@
 import org.apache.cassandra.service.accord.AccordKeyspace;
 import org.apache.cassandra.tracing.TraceKeyspace;
 
+import static org.apache.cassandra.db.Directories.TABLE_DIRECTORY_NAME_SEPARATOR;
 import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
 
 /**
@@ -41,6 +42,7 @@
 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";
@@ -77,14 +79,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