diff --git a/CHANGES.txt b/CHANGES.txt
index 667273df55..7356dae994 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
 4.1.7
 Merged from 4.0:
+ * Fix indexing of a frozen collection that is the clustering key and reversed (CASSANDRA-19889)
  * Emit error when altering a table with non-frozen UDTs with nested non-frozen collections the same way as done upon table creation (CASSANDRA-19925)
  * Safer handling of out-of-range tokens (CASSANDRA-13704)
  * Fix memory leak in BTree.FastBuilder (CASSANDRA-19785)
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 7a3a41e618..34d142ff95 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
@@ -34,6 +34,7 @@ import org.apache.cassandra.cql3.statements.schema.IndexTarget.Type;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.db.marshal.MapType;
+import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.index.sasi.SASIIndex;
 import org.apache.cassandra.schema.*;
@@ -180,6 +181,8 @@ public final class CreateIndexStatement extends AlterSchemaStatement
         if (null == column)
             throw ire("Column '%s' doesn't exist", target.column);
 
+        AbstractType<?> baseType = column.type.unwrap();
+
         if (column.type.referencesDuration())
         {
             if (column.type.isCollection())
@@ -206,17 +209,17 @@ public final class CreateIndexStatement extends AlterSchemaStatement
         if (column.isPartitionKey() && table.partitionKeyColumns().size() == 1)
             throw ire("Cannot create secondary index on the only partition key column %s", column);
 
-        if (column.type.isFrozenCollection() && target.type != Type.FULL)
+        if (baseType.isFrozenCollection() && target.type != Type.FULL)
             throw ire("Cannot create %s() index on frozen column %s. Frozen collections are immutable and must be fully " +
                       "indexed by using the 'full(%s)' modifier", target.type, column, column);
 
-        if (!column.type.isFrozenCollection() && target.type == Type.FULL)
+        if (!baseType.isFrozenCollection() && target.type == Type.FULL)
             throw ire("full() indexes can only be created on frozen collections");
 
-        if (!column.type.isCollection() && target.type != Type.SIMPLE)
+        if (!baseType.isCollection() && target.type != Type.SIMPLE)
             throw ire("Cannot create %s() index on %s. Non-collection columns only support simple indexes", target.type, column);
 
-        if (!(column.type instanceof MapType && column.type.isMultiCell()) && (target.type == Type.KEYS || target.type == Type.KEYS_AND_VALUES))
+        if (!(baseType instanceof MapType && column.type.isMultiCell()) && (target.type == Type.KEYS || target.type == Type.KEYS_AND_VALUES))
             throw ire("Cannot create index on %s of column %s with non-map type", target.type, column);
 
         if (column.type.isUDT() && column.type.isMultiCell())
diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractType.java b/src/java/org/apache/cassandra/db/marshal/AbstractType.java
index 74d4006664..883430f9c5 100644
--- a/src/java/org/apache/cassandra/db/marshal/AbstractType.java
+++ b/src/java/org/apache/cassandra/db/marshal/AbstractType.java
@@ -289,6 +289,11 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
         return false;
     }
 
+    public AbstractType<T> unwrap()
+    {
+        return isReversed() ? ((ReversedType<T>) this).baseType.unwrap() : this;
+    }
+
     public static AbstractType<?> parseDefaultParameters(AbstractType<?> baseType, TypeParser parser) throws SyntaxException
     {
         Map<String, String> parameters = parser.getKeyValueParameters();
diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
index 34b805dc9f..e990209609 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
@@ -1676,6 +1676,15 @@ public class SecondaryIndexTest extends CQLTester
         testIndexOnRegularColumnInsertExpiringColumn(true);
     }
 
+    @Test
+    public void testFullIndexOnClusteringColumn() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int,ck frozen<list<int>>,value int,PRIMARY KEY(pk, ck)) WITH CLUSTERING ORDER BY (ck DESC)");
+        createIndex("CREATE INDEX ON %s(FULL(ck));");
+        execute("INSERT INTO %s (pk,ck,value) VALUES (1,[1,2,3],4)");
+        assertRows(execute("SELECT pk FROM %S WHERE CK=[1,2,3]"), row(1));
+    }
+
     private void testIndexOnRegularColumnInsertExpiringColumn(boolean flushBeforeUpdate) throws Throwable
     {
         createTable("CREATE TABLE %s (pk int, ck int, a int, b int, PRIMARY KEY (pk, ck))");