diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
index ac775a670ae6..44dab07febc0 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
@@ -93,6 +93,15 @@ public void validate(QueryState state)
     {
         super.validate(state);
 
+        // Check the length of a valid index name.
+        // Non-valid indexes are validated in IndexMetadata#validate.
+        if (!state.getClientState().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;
     }
diff --git a/src/java/org/apache/cassandra/schema/SchemaConstants.java b/src/java/org/apache/cassandra/schema/SchemaConstants.java
index 7084b0ec8141..8fa1e76cbf20 100644
--- a/src/java/org/apache/cassandra/schema/SchemaConstants.java
+++ b/src/java/org/apache/cassandra/schema/SchemaConstants.java
@@ -76,6 +76,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;
 
diff --git a/test/unit/org/apache/cassandra/index/IndexNameTest.java b/test/unit/org/apache/cassandra/index/IndexNameTest.java
index 0bb97095aa67..6fefea75d5b3 100644
--- a/test/unit/org/apache/cassandra/index/IndexNameTest.java
+++ b/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));
+    }
 }