diff --git a/CHANGES.txt b/CHANGES.txt
index da7c3e5c05..17eeb618cc 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -10,6 +10,7 @@
  * Add java.base/java.lang.reflect among opens for jvm11-client.options (CASSANDRA-19780)
 Merged from 4.1:
 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 9f5bbfe058..46a9171934 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.internal.CassandraIndex;
 import org.apache.cassandra.index.sasi.SASIIndex;
@@ -212,6 +213,8 @@ public final class CreateIndexStatement extends AlterSchemaStatement
         if (null == column)
             throw ire(COLUMN_DOES_NOT_EXIST, target.column);
 
+        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);
 
@@ -241,16 +244,16 @@ public final class CreateIndexStatement extends AlterSchemaStatement
         if (column.isPartitionKey() && table.partitionKeyColumns().size() == 1)
             throw ire(ONLY_PARTITION_KEY, column);
 
-        if (column.type.isFrozenCollection() && target.type != Type.FULL)
+        if (baseType.isFrozenCollection() && target.type != Type.FULL)
             throw ire(CREATE_ON_FROZEN_COLUMN, target.type, column, column.name.toCQLString());
 
-        if (!column.type.isFrozenCollection() && target.type == Type.FULL)
+        if (!baseType.isFrozenCollection() && target.type == Type.FULL)
             throw ire(FULL_ON_FROZEN_COLLECTIONS);
 
-        if (!column.type.isCollection() && target.type != Type.SIMPLE)
+        if (!baseType.isCollection() && target.type != Type.SIMPLE)
             throw ire(NON_COLLECTION_SIMPLE_INDEX, 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 && baseType.isMultiCell()) && (target.type == Type.KEYS || target.type == Type.KEYS_AND_VALUES))
             throw ire(CREATE_WITH_NON_MAP_TYPE, target.type, column);
 
         if (column.type.isUDT() && column.type.isMultiCell())
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 38f724968e..59a3000071 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
@@ -1714,6 +1714,15 @@ public class SecondaryIndexTest extends CQLTester
         testIndexOnRegularColumnInsertExpiringColumn(true);
     }
 
+    @Test
+    public void testFullIndexOnClusteringColumn()
+    {
+        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))");