CNDB-13299 prevent too long index names (#1633)

Creating an index with too long index names leads to file errors or
incorrect results.

This PR prevents creating new indexes, i.e., not internal, with provided
index names longer than 182 characters. This constant is based on the
most restrictive requirement from different index types. It comes from
SAI by subtracting the longest file name addition from the file name
limit of 255 characters.

Author: k-rus

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

@@ -134,6 +134,15 @@ public void validate(ClientState state)
     {
         super.validate(state);
 
+        // Check the length of a valid index name.
+        // Non-valid indexes are validated in IndexMetadata#validate.
+        if (!state.isInternal
+            && SchemaConstants.isValidName(indexName, true)
+            && indexName.length() > SchemaConstants.INDEX_NAME_LENGTH)
+
+            throw ire("Index name shouldn't be more than %s characters long (got %s chars for %s)",
+                      SchemaConstants.INDEX_NAME_LENGTH, indexName.length(), indexName);
+
         // save the query state to use it for guardrails validation in #apply
         this.state = state;
     }

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

@@ -86,6 +86,14 @@ public final class SchemaConstants
      */
     public static final int NAME_LENGTH = FILENAME_LENGTH - 32 - 1;
 
+    /**
+     * Longest permissible index name, so no index can fail on file name error.
+     * It is based on the most restrictive requirement coming from SAI and calculated by
+     * {@link org.apache.cassandra.index.sai.disk.format.Version#calculateIndexNameAllowedLength}.
+     * The exact number is used here, since it will be in user's documentation.
+     */
+    public static final int INDEX_NAME_LENGTH = 182;
+
     // 59adb24e-f3cd-3e02-97f0-5b395827453f
     public static final UUID emptyVersion;
 

test/unit/org/apache/cassandra/index/IndexNameTest.java

@@ -22,10 +22,14 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import com.datastax.driver.core.exceptions.InvalidQueryException;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.index.sai.disk.format.Version;
+import org.apache.cassandra.schema.SchemaConstants;
 
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
 
 @RunWith(Parameterized.class)
 public class IndexNameTest extends CQLTester
@@ -121,4 +125,51 @@ public void failOnBadCharIndexName()
         assertThatThrownBy(() -> execute(String.format(createIndexQuery, "\"unacceptable index name\"", "%s", columnName)))
         .isInstanceOf(ConfigurationException.class);
     }
+
+    @Test
+    public void testTooLongNamesInternal() throws Throwable
+    {
+        String longName = "a".repeat(183);
+
+        createTable("CREATE TABLE %s (" +
+                    "key int PRIMARY KEY," +
+                    "value int)"
+        );
+        createIndex(String.format(createIndexQuery, longName, "%s", "value"));
+        execute(String.format("INSERT INTO %%s (\"key\", %s) VALUES (1, 1)", "value"));
+        execute(String.format("INSERT INTO %%s (\"key\", %s) VALUES (2, 2)", "value"));
+
+        beforeAndAfterFlush(() -> assertRows(execute(String.format("SELECT key, %s FROM %%s WHERE %<s = 1", "value")), row(1, 1)));
+    }
+
+    @Test
+    public void testMaxAcceptableLongNamesNewIndex() throws Throwable
+    {
+        assertEquals(182, Version.calculateIndexNameAllowedLength());
+        String longName = "a".repeat(182);
+        createTable("CREATE TABLE %s (" +
+                    "key int PRIMARY KEY," +
+                    "value int)"
+        );
+        executeNet(String.format(createIndexQuery, longName, "%s", "value"));
+
+        execute(String.format("INSERT INTO %%s (\"key\", %s) VALUES (1, 1)", "value"));
+        execute(String.format("INSERT INTO %%s (\"key\", %s) VALUES (2, 2)", "value"));
+
+        beforeAndAfterFlush(() -> assertRows(execute(String.format("SELECT key, %s FROM %%s WHERE %<s = 1", "value")), row(1, 1)));
+    }
+
+    @Test
+    public void failTooLongNamesNewIndex()
+    {
+        String longName = "a".repeat(183);
+        createTable("CREATE TABLE %s (" +
+                    "key int PRIMARY KEY," +
+                    "value int)"
+        );
+        assertThatThrownBy(() -> executeNet(String.format(createIndexQuery, longName, "%s", "value")))
+        .isInstanceOf(InvalidQueryException.class)
+        .hasMessage(String.format("Index name shouldn't be more than %s characters long (got %s chars for %s)",
+                                  SchemaConstants.INDEX_NAME_LENGTH, longName.length(), longName));
+    }
 }