diff --git a/CHANGES.txt b/CHANGES.txt
index b1feeab60e..eb59885fdf 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,7 @@
 
 
 3.0-rc2
+ * Support empty ColumnFilter for backward compatility on empty IN (CASSANDRA-10471)
  * Remove Pig support (CASSANDRA-10542)
  * Fix LogFile throws Exception when assertion is disabled (CASSANDRA-10522)
  * Revert CASSANDRA-7486, make CMS default GC, move GC config to
diff --git a/src/java/org/apache/cassandra/cql3/ResultSet.java b/src/java/org/apache/cassandra/cql3/ResultSet.java
index 9d2fbec967..bc4daedc95 100644
--- a/src/java/org/apache/cassandra/cql3/ResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/ResultSet.java
@@ -269,6 +269,11 @@ public class ResultSet
             return names == null ? columnCount : names.size();
         }
 
+        /**
+         * Adds the specified column which will not be serialized.
+         *
+         * @param name the column
+         */
         public void addNonSerializedColumn(ColumnSpecification name)
         {
             // See comment above. Because columnCount doesn't account the newly added name, it
diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
index a6d8e93bf0..fb8d567773 100644
--- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
@@ -82,7 +82,7 @@ public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
         {
             if (cqlRows.size() != 1)
                 throw new IllegalStateException("One row required, " + cqlRows.size() + " found");
-            return new Row(cqlRows.metadata.names, cqlRows.rows.get(0));
+            return new Row(cqlRows.metadata.requestNames(), cqlRows.rows.get(0));
         }
 
         public Iterator<Row> iterator()
@@ -95,7 +95,7 @@ public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
                 {
                     if (!iter.hasNext())
                         return endOfData();
-                    return new Row(cqlRows.metadata.names, iter.next());
+                    return new Row(cqlRows.metadata.requestNames(), iter.next());
                 }
             };
         }
@@ -160,7 +160,7 @@ public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
             this.select = select;
             this.pager = pager;
             this.pageSize = pageSize;
-            this.metadata = select.getResultMetadata().names;
+            this.metadata = select.getResultMetadata().requestNames();
         }
 
         public int size()
diff --git a/src/java/org/apache/cassandra/cql3/selection/Selection.java b/src/java/org/apache/cassandra/cql3/selection/Selection.java
index bbb8b256c9..4fec2bc8eb 100644
--- a/src/java/org/apache/cassandra/cql3/selection/Selection.java
+++ b/src/java/org/apache/cassandra/cql3/selection/Selection.java
@@ -125,23 +125,6 @@ public abstract class Selection
         return false;
     }
 
-    /**
-     * Returns the index of the specified column.
-     *
-     * @param def the column definition
-     * @return the index of the specified column
-     */
-    public int indexOf(final ColumnDefinition def)
-    {
-        return Iterators.indexOf(getColumns().iterator(), new Predicate<ColumnDefinition>()
-           {
-               public boolean apply(ColumnDefinition n)
-               {
-                   return def.name.equals(n.name);
-               }
-           });
-    }
-
     public ResultSet.ResultMetadata getResultMetadata(boolean isJson)
     {
         if (!isJson)
@@ -199,6 +182,29 @@ public abstract class Selection
                : new SimpleSelection(cfm, defs, mapping, false);
     }
 
+    /**
+     * Returns the index of the specified column within the resultset
+     * @param c the column
+     * @return the index of the specified column within the resultset or -1
+     */
+    public int getResultSetIndex(ColumnDefinition c)
+    {
+        return getColumnIndex(c);
+    }
+
+    /**
+     * Returns the index of the specified column
+     * @param c the column
+     * @return the index of the specified column or -1
+     */
+    protected final int getColumnIndex(ColumnDefinition c)
+    {
+        for (int i = 0, m = columns.size(); i < m; i++)
+            if (columns.get(i).name.equals(c.name))
+                return i;
+        return -1;
+    }
+
     private static SelectionColumnMapping collectColumnMappings(CFMetaData cfm,
                                                                 List<RawSelector> rawSelectors,
                                                                 SelectorFactories factories)
@@ -497,12 +503,27 @@ public abstract class Selection
             return factories.getFunctions();
         }
 
+        @Override
+        public int getResultSetIndex(ColumnDefinition c)
+        {
+            int index = getColumnIndex(c);
+
+            if (index < 0)
+                return -1;
+
+            for (int i = 0, m = factories.size(); i < m; i++)
+                if (factories.get(i).isSimpleSelectorFactory(index))
+                    return i;
+
+            return -1;
+        }
+
         @Override
         public int addColumnForOrdering(ColumnDefinition c)
         {
             int index = super.addColumnForOrdering(c);
             factories.addSelectorForOrdering(c, index);
-            return index;
+            return factories.size() - 1;
         }
 
         public boolean isAggregate()
diff --git a/src/java/org/apache/cassandra/cql3/selection/Selector.java b/src/java/org/apache/cassandra/cql3/selection/Selector.java
index 8d58d12d08..4515cd0df3 100644
--- a/src/java/org/apache/cassandra/cql3/selection/Selector.java
+++ b/src/java/org/apache/cassandra/cql3/selection/Selector.java
@@ -103,6 +103,18 @@ public abstract class Selector implements AssignmentTestable
             return false;
         }
 
+        /**
+         * Checks if this factory creates <code>Selector</code>s that simply return the specified column.
+         *
+         * @param index the column index
+         * @return <code>true</code> if this factory creates <code>Selector</code>s that simply return
+         * the specified column, <code>false</code> otherwise.
+         */
+        public boolean isSimpleSelectorFactory(int index)
+        {
+            return false;
+        }
+
         /**
          * Returns the name of the column corresponding to the output value of the selector instances created by
          * this factory.
diff --git a/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java b/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java
index 5ea29577d6..fbbfbb5216 100644
--- a/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java
+++ b/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java
@@ -98,6 +98,17 @@ final class SelectorFactories implements Iterable<Selector.Factory>
         return functions;
     }
 
+    /**
+     * Returns the factory with the specified index.
+     *
+     * @param i the factory index
+     * @return the factory with the specified index
+     */
+    public Selector.Factory get(int i)
+    {
+        return factories.get(i);
+    }
+
     /**
      * Adds a new <code>Selector.Factory</code> for a column that is needed only for ORDER BY purposes.
      * @param def the column that is needed for ordering
@@ -191,4 +202,13 @@ final class SelectorFactories implements Iterable<Selector.Factory>
             }
         });
     }
+
+    /**
+     * Returns the number of factories.
+     * @return the number of factories
+     */
+    public int size()
+    {
+        return factories.size();
+    }
 }
diff --git a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java
index 2e0514a6d6..e4040fa8a6 100644
--- a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java
+++ b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java
@@ -59,6 +59,12 @@ public final class SimpleSelector extends Selector
             {
                 return new SimpleSelector(def.name.toString(), idx, def.type);
             }
+
+            @Override
+            public boolean isSimpleSelectorFactory(int index)
+            {
+                return index == idx;
+            }
         };
     }
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 67d46228c9..a0e0be8960 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -907,7 +907,7 @@ public class SelectStatement implements CQLStatement
                 final ColumnDefinition def = cfm.getColumnDefinition(column);
                 if (def == null)
                     handleUnrecognizedOrderingColumn(column);
-                int index = selection.indexOf(def);
+                int index = selection.getResultSetIndex(def);
                 if (index < 0)
                     index = selection.addColumnForOrdering(def);
                 orderingIndexes.put(def.name, index);
diff --git a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
index 98b3600e31..40097bdef1 100644
--- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java
@@ -289,7 +289,12 @@ public class ColumnFilter
         public ColumnFilter build()
         {
             boolean isFetchAll = metadata != null;
-            assert isFetchAll || selection != null;
+
+            PartitionColumns selectedColumns = selection == null ? null : selection.build();
+            // It's only ok to have selection == null in ColumnFilter if isFetchAll. So deal with the case of a "selection" builder
+            // with nothing selected (we can at least happen on some backward compatible queries - CASSANDRA-10471).
+            if (!isFetchAll && selectedColumns == null)
+                selectedColumns = PartitionColumns.NONE;
 
             SortedSetMultimap<ColumnIdentifier, ColumnSubselection> s = null;
             if (subSelections != null)
@@ -299,7 +304,7 @@ public class ColumnFilter
                     s.put(subSelection.column().name, subSelection);
             }
 
-            return new ColumnFilter(isFetchAll, metadata, selection == null ? null : selection.build(), s);
+            return new ColumnFilter(isFetchAll, metadata, selectedColumns, s);
         }
     }
 
diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/SelectOrderByTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/SelectOrderByTest.java
index 341eed9747..73bbacaa07 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/SelectOrderByTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/SelectOrderByTest.java
@@ -338,6 +338,58 @@ public class SelectOrderByTest extends CQLTester
 
         assertRows(execute("SELECT my_id, col1 FROM %s WHERE my_id in('key1', 'key2', 'key3') ORDER BY col1"),
                    row("key1", 1), row("key3", 2), row("key2", 3));
+
+        createTable("CREATE TABLE %s (pk1 int, pk2 int, c int, v text, PRIMARY KEY ((pk1, pk2), c) )");
+        execute("INSERT INTO %s (pk1, pk2, c, v) VALUES (?, ?, ?, ?)", 1, 1, 2, "A");
+        execute("INSERT INTO %s (pk1, pk2, c, v) VALUES (?, ?, ?, ?)", 1, 2, 1, "B");
+        execute("INSERT INTO %s (pk1, pk2, c, v) VALUES (?, ?, ?, ?)", 1, 3, 3, "C");
+        execute("INSERT INTO %s (pk1, pk2, c, v) VALUES (?, ?, ?, ?)", 1, 1, 4, "D");
+
+        assertRows(execute("SELECT v, ttl(v), c FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c; ", 1, 1, 2),
+                   row("B", null, 1),
+                   row("A", null, 2),
+                   row("D", null, 4));
+
+        assertRows(execute("SELECT v, ttl(v), c as name_1 FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c; ", 1, 1, 2),
+                   row("B", null, 1),
+                   row("A", null, 2),
+                   row("D", null, 4));
+
+        assertRows(execute("SELECT v FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c; ", 1, 1, 2),
+                   row("B"),
+                   row("A"),
+                   row("D"));
+
+        assertRows(execute("SELECT v as c FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c; ", 1, 1, 2),
+                   row("B"),
+                   row("A"),
+                   row("D"));
+
+        createTable("CREATE TABLE %s (pk1 int, pk2 int, c1 int, c2 int, v text, PRIMARY KEY ((pk1, pk2), c1, c2) )");
+        execute("INSERT INTO %s (pk1, pk2, c1, c2, v) VALUES (?, ?, ?, ?, ?)", 1, 1, 4, 4, "A");
+        execute("INSERT INTO %s (pk1, pk2, c1, c2, v) VALUES (?, ?, ?, ?, ?)", 1, 2, 1, 2, "B");
+        execute("INSERT INTO %s (pk1, pk2, c1, c2, v) VALUES (?, ?, ?, ?, ?)", 1, 3, 3, 3, "C");
+        execute("INSERT INTO %s (pk1, pk2, c1, c2, v) VALUES (?, ?, ?, ?, ?)", 1, 1, 4, 1, "D");
+
+        assertRows(execute("SELECT v, ttl(v), c1, c2 FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c1, c2; ", 1, 1, 2),
+                   row("B", null, 1, 2),
+                   row("D", null, 4, 1),
+                   row("A", null, 4, 4));
+
+        assertRows(execute("SELECT v, ttl(v), c1 as name_1, c2 as name_2 FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c1, c2; ", 1, 1, 2),
+                   row("B", null, 1, 2),
+                   row("D", null, 4, 1),
+                   row("A", null, 4, 4));
+
+        assertRows(execute("SELECT v FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c1, c2; ", 1, 1, 2),
+                   row("B"),
+                   row("D"),
+                   row("A"));
+
+        assertRows(execute("SELECT v as c2 FROM %s where pk1 = ? AND pk2 IN (?, ?) ORDER BY c1, c2; ", 1, 1, 2),
+                   row("B"),
+                   row("D"),
+                   row("A"));
     }
 
     /**