Merge branch 'cassandra-4.1' into cassandra-5.0

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

Author: netudima

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.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 @@ 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);
 
         SchemaTransformationResult result = Schema.instance.transform(this, locally);
 
@@ -134,16 +136,6 @@ public ResultMessage execute(QueryState state, boolean locally)
         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" +
-                                                             " 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";
@@ -206,6 +206,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);
@@ -215,8 +223,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

@@ -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 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
@@ -247,7 +245,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

@@ -22,7 +22,6 @@
 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,17 +46,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();
 
     /**
@@ -111,11 +109,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("");
@@ -128,7 +121,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

@@ -36,6 +36,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.UserFunctions.FunctionsDiff;
 import org.apache.cassandra.schema.Tables.TablesDiff;
@@ -46,12 +47,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
@@ -317,16 +336,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, 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.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.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 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 boolean dependsOn(Function function)
 
     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();
 
@@ -507,6 +509,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
@@ -603,7 +616,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;
     }
 
     /**
@@ -661,9 +708,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
@@ -1722,4 +1774,4 @@ public static ColumnMetadata getCompactValueColumn(RegularAndStaticColumns colum
 
     }
 
-}
+}