From 767a68cd00050298abf7bbfd8b322e5663439c23 Mon Sep 17 00:00:00 2001
From: Marcus Eriksson <marcuse@apache.org>
Date: Thu, 17 Oct 2019 10:24:57 +0200
Subject: [PATCH] Ensure legacy rows have primary key livenessinfo when they
 contain illegal cells

Patch by marcuse and Sam Tunnicliffe; reviewed by Benedict Elliott Smith for CASSANDRA-15365
---
 CHANGES.txt                                   |   1 +
 .../org/apache/cassandra/db/LegacyLayout.java |  56 ++++++++++--
 ...illegal_cell_names-ka-2-CompressionInfo.db | Bin 0 -> 43 bytes
 ...cy_ka_with_illegal_cell_names-ka-2-Data.db | Bin 0 -> 59 bytes
 ...a_with_illegal_cell_names-ka-2-Digest.sha1 |   1 +
 ..._ka_with_illegal_cell_names-ka-2-Filter.db | Bin 0 -> 16 bytes
 ...y_ka_with_illegal_cell_names-ka-2-Index.db | Bin 0 -> 18 bytes
 ...with_illegal_cell_names-ka-2-Statistics.db | Bin 0 -> 4452 bytes
 ...cy_ka_with_illegal_cell_names-ka-2-TOC.txt |   8 ++
 ...legal_cell_names_2-ka-1-CompressionInfo.db | Bin 0 -> 43 bytes
 ..._ka_with_illegal_cell_names_2-ka-1-Data.db | Bin 0 -> 67 bytes
 ...with_illegal_cell_names_2-ka-1-Digest.sha1 |   1 +
 ...a_with_illegal_cell_names_2-ka-1-Filter.db | Bin 0 -> 16 bytes
 ...ka_with_illegal_cell_names_2-ka-1-Index.db | Bin 0 -> 18 bytes
 ...th_illegal_cell_names_2-ka-1-Statistics.db | Bin 0 -> 4452 bytes
 ..._with_illegal_cell_names_2-ka-1-Summary.db | Bin 0 -> 92 bytes
 ..._ka_with_illegal_cell_names_2-ka-1-TOC.txt |   8 ++
 .../io/sstable/LegacySSTableTest.java         |  80 ++++++++++++++++--
 18 files changed, 141 insertions(+), 14 deletions(-)
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-CompressionInfo.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Data.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Digest.sha1
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Filter.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Index.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Statistics.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-TOC.txt
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-CompressionInfo.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Data.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Digest.sha1
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Filter.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Index.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Statistics.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Summary.db
 create mode 100644 test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-TOC.txt

diff --git a/CHANGES.txt b/CHANGES.txt
index bd129421b8..d58a199fd5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0.20
+ * Ensure legacy rows have primary key livenessinfo when they contain illegal cells (CASSANDRA-15365)
 Merged from 2.2
  * In-JVM DTest: Set correct internode message version for upgrade test (CASSANDRA-15371)
 
diff --git a/src/java/org/apache/cassandra/db/LegacyLayout.java b/src/java/org/apache/cassandra/db/LegacyLayout.java
index 1a03c91371..6e93d0838e 100644
--- a/src/java/org/apache/cassandra/db/LegacyLayout.java
+++ b/src/java/org/apache/cassandra/db/LegacyLayout.java
@@ -1291,6 +1291,22 @@ public abstract class LegacyLayout
         private LegacyRangeTombstone rowDeletion;
         private LegacyRangeTombstone collectionDeletion;
 
+        /**
+         * Used to track if we need to add pk liveness info (row marker) when removing invalid legacy cells.
+         *
+         * In 2.1 these invalid cells existed but were not queryable, in this case specifically because they
+         * represented values for clustering key columns that were written as data cells.
+         *
+         * However, the presence (or not) of such cells on an otherwise empty CQL row (or partition) would decide
+         * if an empty result row were returned for the CQL row (or partition).  To maintain this behaviour we
+         * insert a row marker containing the liveness info of these invalid cells iff we have no other data
+         * on the row.
+         *
+         * See also CASSANDRA-15365
+         */
+        private boolean hasValidCells = false;
+        private LivenessInfo invalidLivenessInfo = null;
+
         public CellGrouper(CFMetaData metadata, SerializationHelper helper)
         {
             this(metadata, helper, false);
@@ -1317,6 +1333,8 @@ public abstract class LegacyLayout
             this.clustering = null;
             this.rowDeletion = null;
             this.collectionDeletion = null;
+            this.invalidLivenessInfo = null;
+            this.hasValidCells = false;
         }
 
         public boolean addAtom(LegacyAtom atom)
@@ -1326,7 +1344,7 @@ public abstract class LegacyLayout
                  : addRangeTombstone(atom.asRangeTombstone());
         }
 
-        public boolean addCell(LegacyCell cell)
+        private boolean addCell(LegacyCell cell)
         {
             if (clustering == null)
             {
@@ -1359,21 +1377,38 @@ public abstract class LegacyLayout
                     builder.addRowDeletion(Row.Deletion.regular(new DeletionTime(cell.timestamp, cell.localDeletionTime)));
                 else
                     builder.addPrimaryKeyLivenessInfo(LivenessInfo.create(cell.timestamp, FAKE_TTL, cell.localDeletionTime));
+                hasValidCells = true;
+            }
+            else if (column.isPrimaryKeyColumn() && metadata.isCQLTable())
+            {
+                // SSTables generated offline and side-loaded may include invalid cells which have the column name
+                // of a primary key column. So that we don't fail when encountering these cells, we treat them the
+                // same way as 2.1 did, namely we include their clusterings in the new CQL row, but drop the invalid
+                // column part of the cell
+                noSpamLogger.warn("Illegal cell name for CQL3 table {}.{}. {} is defined as a primary key column",
+                                  metadata.ksName, metadata.cfName, column.name);
+
+                if (invalidLivenessInfo != null)
+                {
+                    // when we have several invalid cells we follow the logic in LivenessInfo#supersedes when picking the PKLI to keep:
+                    LivenessInfo newInvalidLiveness = LivenessInfo.create(cell.timestamp, cell.isTombstone() ? FAKE_TTL : cell.ttl, cell.localDeletionTime);
+                    if (newInvalidLiveness.supersedes(invalidLivenessInfo))
+                        invalidLivenessInfo = newInvalidLiveness;
+                }
+                else
+                {
+                    invalidLivenessInfo = LivenessInfo.create(cell.timestamp, cell.isTombstone() ? FAKE_TTL : cell.ttl, cell.localDeletionTime);
+                }
+                return true;
             }
             else
             {
                 if (collectionDeletion != null && collectionDeletion.start.collectionName.name.equals(column.name) && collectionDeletion.deletionTime.deletes(cell.timestamp))
                     return true;
 
-                if (column.isPrimaryKeyColumn() && metadata.isCQLTable())
-                {
-                    noSpamLogger.warn("Illegal cell name for CQL3 table {}.{}. {} is defined as a primary key column",
-                                      metadata.ksName, metadata.cfName, column.name);
-                    return true;
-                }
-
                 if (helper.includes(column))
                 {
+                    hasValidCells = true;
                     CellPath path = null;
                     if (column.isComplex())
                     {
@@ -1422,6 +1457,7 @@ public abstract class LegacyLayout
                     {
                         builder.addRowDeletion(Row.Deletion.regular(tombstone.deletionTime));
                         rowDeletion = tombstone;
+                        hasValidCells = true;
                     }
                     return true;
                 }
@@ -1434,6 +1470,7 @@ public abstract class LegacyLayout
             builder.newRow(clustering);
             builder.addRowDeletion(Row.Deletion.regular(tombstone.deletionTime));
             rowDeletion = tombstone;
+            hasValidCells = true;
 
             return true;
         }
@@ -1464,6 +1501,7 @@ public abstract class LegacyLayout
             builder.addComplexDeletion(tombstone.start.collectionName, tombstone.deletionTime);
             if (rowDeletion == null || tombstone.deletionTime.supersedes(rowDeletion.deletionTime))
                 collectionDeletion = tombstone;
+            hasValidCells = true;
 
             return true;
         }
@@ -1488,6 +1526,8 @@ public abstract class LegacyLayout
 
         public Row getRow()
         {
+            if (!hasValidCells && invalidLivenessInfo != null)
+                builder.addPrimaryKeyLivenessInfo(invalidLivenessInfo);
             return builder.build();
         }
     }
diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-CompressionInfo.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-CompressionInfo.db
new file mode 100644
index 0000000000000000000000000000000000000000..26a0dbeab4c4b867266df98a4fb516e81fea424f
GIT binary patch
literal 43
fcmZSJ^@%cZ&d)6<N-ZwVFJb@zMlcB?%#nEjvSS5K

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Data.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Data.db
new file mode 100644
index 0000000000000000000000000000000000000000..c805f7d0b9115235993b58b112cec0fb7fe5dc4a
GIT binary patch
literal 59
zcmXqKU|=}Rzyc(h>;M1%-@w4g(7?yQn8?7ul*r1UP>?Ok@PU~zcU7=yHBb#(Mruw@
Mz5xS6LM+dI0G7uNM*si-

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Digest.sha1 b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Digest.sha1
new file mode 100644
index 0000000000..0c696fbd10
--- /dev/null
+++ b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Digest.sha1
@@ -0,0 +1 @@
+2529627719
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Filter.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Filter.db
new file mode 100644
index 0000000000000000000000000000000000000000..55433281172fcd95413597915e3189f0b8ed77bc
GIT binary patch
literal 16
XcmZQzU|?lnU|?io5MXFvVNd`70`>sC

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Index.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Index.db
new file mode 100644
index 0000000000000000000000000000000000000000..fbdd95081da260bedef7a940475304f25f656ffb
GIT binary patch
literal 18
QcmZQzVPIfj24Xk>004ae2mk;8

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Statistics.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-Statistics.db
new file mode 100644
index 0000000000000000000000000000000000000000..0e471d4f026f9a051b78f5040a7e061ff32d9a6a
GIT binary patch
literal 4452
zcmeI$`%6<%902fp?>6Ui>3pS?lxCL1l*~-9RyLW7Ky#V)BCl@v03XazW`<GL4-#fk
znc$-bNSGvL<rDj%(%wFZtVl6f7Dmk?%`#kfZr6Rk=Rb&c;BwCAe9v~zIrp5q!zhXp
zP?ofq%V_RO<}QZ2vZ=6qU8amJU^B9`vJAGckj>NRSeYhEFN-~-*CyyRTAfyt$m;aj
z`t1BXtxjH6{9bv(>oE6Wb8<DAOtWr&G~Iu))01=6T&?9M*5X`dz9~%Ro5RF+kgpT+
z1#2!M>_Avd*paX^VOPQu!XAXZ2rneOgm4hyP{M1h=I14bh{qG&NjQ^m3E}gETL=$f
z)5P<j*Aa0&;cjfkm+&!c0rCBU8XOmz7bHlc2mr3=umM|4jMs4n$DMnzU5(faPGCzf
zVf*aH_8-Jv8iu{19$S`*9aW1R+l8Izjh*I)UAP^)oW*V+)^D!@$BhBVbig<4)E(Hh
z{Q9(1)5cdQ&ZzlyYpG_QU<*=Tp!Px`UjM>@XDBXuRDtX;(TMD*HX%FBYLQ)L&F>yt
zl4$UH^iX`V1-X(}yASfh5r5<*LrKU%efW8YPUGhtQF#xw-!S?CdFy7}zg_+Y6wmo0
zMLwF)j9mR|J#yOvJU_#2Xg+CY_et2EmMc#omj+%#etJI~nh9w|?^ou6<Ry#?{EN`~
z5VY@ef^lJy(?)2~-XSzzhy3g!7#Fu0rl6e^aX+A40y@xsa!Zx=!??RZ)dTGr6^z!i
zujJQ77+-Y#%sBM2weh#0LtdxEK!^2b-GGixJ5mR&cpIJqom8J237rvqcN6r1-)>{j
z6%QMg(AU+~fzZ#Z%1%R1-5HDnrdPV9w?k)a*#h@J-I&-4_djhqCxQE)(HcfN!RMLL
zrj5nG0zGXw3oKmTCsjdDPnW>wBl=l+9{zraS9y+Y2JIcIZWiALc22n5qXTx`U-1==
zU-HId9FE^(AihNk+IzF*1@QO4@0z?r2I2va$E5IiE&r}mm4o=|WMeQq4@5?M?va4F
zeCT7xLEvpQ%}uaBdyLVMa6RWGDH<w4`?B=5-cjJXcf(x`z};7_HpBiJ?uQOW+2XYH
z__IH;z^a_dnPWCnh1N3v4kmNHDZ>}6`Tu(6;->-aajw#S=Hh2&F1Me#&~#AtGZ&f<
z`<csr<^l`iUpjMf|5?l>OZVyAMs+0je}*M5<s$Nl`GINXOSS?-(JaNVELSsM{0jn<
BPe=d&

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-TOC.txt b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-TOC.txt
new file mode 100644
index 0000000000..1222811e02
--- /dev/null
+++ b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names/legacy_tables-legacy_ka_with_illegal_cell_names-ka-2-TOC.txt
@@ -0,0 +1,8 @@
+Digest.sha1
+Data.db
+Filter.db
+Summary.db
+Index.db
+TOC.txt
+CompressionInfo.db
+Statistics.db
diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-CompressionInfo.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-CompressionInfo.db
new file mode 100644
index 0000000000000000000000000000000000000000..908f3b1f9b529e4fe3a20b93ed256f0534e25ee4
GIT binary patch
literal 43
fcmZSJ^@%cZ&d)6<N-ZwVFJb@zMlcB?0+4wCv#14U

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Data.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Data.db
new file mode 100644
index 0000000000000000000000000000000000000000..33b88a0329afc66aa3d543ada042dac70341c22a
GIT binary patch
literal 67
zcmWGwU|=}Rzyc(h>;M1%-@w4gFqwyeF_D3RDKSxifsu(pFolglj8B@ukf|VB7%0rJ
T;RH}MOKxgv0RzKiNA3>*#}*C0

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Digest.sha1 b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Digest.sha1
new file mode 100644
index 0000000000..20deb5b232
--- /dev/null
+++ b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Digest.sha1
@@ -0,0 +1 @@
+3340111295
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Filter.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Filter.db
new file mode 100644
index 0000000000000000000000000000000000000000..55433281172fcd95413597915e3189f0b8ed77bc
GIT binary patch
literal 16
XcmZQzU|?lnU|?io5MXFvVNd`70`>sC

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Index.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Index.db
new file mode 100644
index 0000000000000000000000000000000000000000..fbdd95081da260bedef7a940475304f25f656ffb
GIT binary patch
literal 18
QcmZQzVPIfj24Xk>004ae2mk;8

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Statistics.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Statistics.db
new file mode 100644
index 0000000000000000000000000000000000000000..f83575cfa835a348fd652254c1f377b4aaca19c2
GIT binary patch
literal 4452
zcmeI$`AZa0902g0*<EipU2jV%k1UB*GBZKl)D<-XP4nm?M|bgnRMw@;48yD+B+Mc+
z!J`92RFtyv#D1{U%JLwxBE?{{GHNT*G{bFHciQjw{RRE74~BW4_kFYT=FN9z9?RqL
z_`J2Uf!$H;mc(u}yJhqI@|2kpCZEa3Qb;nGf&wO2u4E+gER`hUq)HK|lq-}9c|4<3
zWvjCDaurHxS<$=bHv3cT!Mb9l)oNF)T_|eoe6Jnr>ebe@ljy6fJKafP(VZL?Zh%~y
zfUD^DLc%75O$nP3wjgXpSVY*CusvaC!fu2;3HuV>PB?^cjNaV5B@^*X!o`HE2zL;k
z#HNV*LG2{sD#HEPv=iY{Y(DY({8}6r=n@hXVE_PIH3`KwC9c<O8OJS#u&p%M>&{_|
zuVXtN#CDy)-sp$DxdB^}iXBvk9nps!?|_}=f?co=yPUypB=UDihT|G{WXk<JcItlY
zI-H-B<|T^LiCkW;HPcV9`KixQe}Mq!zhHa>#f8Ha$R^*Kk<Ajd$mS~wWXl!ZyT?#O
za-5I0vX6!!TeI(WM0TEbMRuD^K=vHP_uF?7-|v9R2dIDO%zNZL;dp!pT-7L^^F@q&
zCaxX1W@#64*F!u%Q(b62DGQr<*q@R{UqCMLXhMEE;15mvbfWhweO2@V#`&&=C_nh!
zN6cYdP-q?oEj&DluGb_lyAZ}ryVMKN=CY`t(3b8!=y|eE6_3HV4L^1e+Ahc&<+GD$
z=^Bi$Z@DxFy=h0x9cZ6dDIw5)V_9v`!D*-Kp=EFUQ=k(Xjs-$zc;DX*ef+oeEOf=A
z=4j}a#2OFiXVqmFp%?B=L;+J<tkb)pGa@75^G`L$cf#kN(q0k4=bu)nr+dNi^h|45
z5inmxsV@Tyyhg>b(2I-3@csyYRaU|8hv`<k*>KR`tonA*U0{p28-q$<tD_ZP;q{AN
z+s?u3w;hk^5QF{>OnE;19=J3~dn6$4{&-dl@0Zt)=-6@)-<G8DhU<aAfX{;>5SLDV
z=s5wrx3;|%j^~gjI1uJ@Zi1|_67(-i?;4r`u75k#*9hEy^Hw_?uX@0DBFGS@#AiqT
zL;%O;%rBQ3%q!5>@jKYC2IJakaEz{MTyx<G;nrMSYcf~S?~Q9NG$F<{7n%;9am|J1
z!?@-$uDL)${7-8x_CJf&0){8o*N02ki1bsyYGK>BcF>jm<Ah>(G{dlM|EhlhcYsZ4

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Summary.db b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-Summary.db
new file mode 100644
index 0000000000000000000000000000000000000000..9b9000596da2a87ab0262fc4e1893b7161b9f76d
GIT binary patch
literal 92
zcmZQzU}#`qU|<Ad5D)<31|Vi+0TMvK457dzNC?D-&@8#Ri3LC&6Ic#G2r+<V_^!D+
F0RW*b1_b~B

literal 0
HcmV?d00001

diff --git a/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-TOC.txt b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-TOC.txt
new file mode 100644
index 0000000000..8d621be658
--- /dev/null
+++ b/test/data/legacy-sstables/ka/legacy_tables/legacy_ka_with_illegal_cell_names_2/legacy_tables-legacy_ka_with_illegal_cell_names_2-ka-1-TOC.txt
@@ -0,0 +1,8 @@
+Data.db
+TOC.txt
+Filter.db
+Summary.db
+CompressionInfo.db
+Statistics.db
+Digest.sha1
+Index.db
diff --git a/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java b/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java
index ecb812567d..42e8ff6c27 100644
--- a/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java
@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
 
+import com.google.common.collect.Iterables;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -41,13 +42,17 @@ import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.LivenessInfo;
 import org.apache.cassandra.db.SinglePartitionSliceCommandTest;
 import org.apache.cassandra.db.compaction.Verifier;
+import org.apache.cassandra.db.lifecycle.SSTableSet;
 import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.db.marshal.SetType;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.db.rows.RangeTombstoneMarker;
+import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.db.rows.Unfiltered;
+import org.apache.cassandra.db.rows.UnfilteredRowIterator;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
@@ -62,11 +67,11 @@ import org.apache.cassandra.streaming.StreamPlan;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Pair;
 
 import static org.apache.cassandra.cql3.CQLTester.assertRows;
 import static org.apache.cassandra.cql3.CQLTester.row;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 
 /**
  * Tests backwards compatibility for SSTables
@@ -428,10 +433,15 @@ public class LegacySSTableTest
          *            ["b:bb:pk","00000002",1555000750634000],
          *            ["b:bb:v1","bbb",1555000750634000]]}
          * ]
+         * and an extra sstable with only the invalid cell name
+         * [
+         * {"key": "3",
+         *  "cells": [["a:aa:pk","68656c6c6f30",1570466358949]]}
+         * ]
          *
          */
-
-        QueryProcessor.executeInternal("CREATE TABLE legacy_tables.legacy_ka_with_illegal_cell_names (" +
+        String table = "legacy_ka_with_illegal_cell_names";
+        QueryProcessor.executeInternal("CREATE TABLE legacy_tables." + table + " (" +
                                        " pk int," +
                                        " c1 text," +
                                        " c2 text," +
@@ -439,10 +449,68 @@ public class LegacySSTableTest
                                        " PRIMARY KEY(pk, c1, c2))");
         loadLegacyTable("legacy_%s_with_illegal_cell_names%s", "ka", "");
         UntypedResultSet results =
-            QueryProcessor.executeOnceInternal(
-                String.format("SELECT * FROM legacy_tables.legacy_ka_with_illegal_cell_names"));
+            QueryProcessor.executeOnceInternal("SELECT * FROM legacy_tables."+table);
 
-        assertRows(results, row(1, "a", "aa", "aaa"), row(2, "b", "bb", "bbb"));
+        assertRows(results, row(1, "a", "aa", "aaa"), row(2, "b", "bb", "bbb"), row (3, "a", "aa", null));
+        Keyspace.open("legacy_tables").getColumnFamilyStore(table).forceMajorCompaction();
+    }
+
+    @Test
+    public void testReadingLegacyTablesWithIllegalCellNamesPKLI() throws Exception {
+        /**
+         *
+         * Makes sure we grab the correct PKLI when we have illegal columns
+         *
+         * sstable looks like this:
+         * [
+         * {"key": "3",
+         *  "cells": [["a:aa:","",100],
+         *            ["a:aa:pk","6d656570",200]]}
+         * ]
+         */
+        /*
+        this generates the stable on 2.1:
+        CFMetaData metadata = CFMetaData.compile("create table legacy_tables.legacy_ka_with_illegal_cell_names_2 (pk int, c1 text, c2 text, v1 text, primary key (pk, c1, c2))", "legacy_tables");
+        try (SSTableSimpleUnsortedWriter writer = new SSTableSimpleUnsortedWriter(new File("/tmp/sstable21"),
+                                                                                  metadata,
+                                                                                  new ByteOrderedPartitioner(),
+                                                                                  10))
+        {
+            writer.newRow(bytes(3));
+            writer.addColumn(new BufferCell(Util.cellname("a", "aa", ""), bytes(""), 100));
+            writer.addColumn(new BufferCell(Util.cellname("a", "aa", "pk"), bytes("meep"), 200));
+        }
+        */
+        String table = "legacy_ka_with_illegal_cell_names_2";
+        QueryProcessor.executeInternal("CREATE TABLE legacy_tables." + table + " (" +
+                                       " pk int," +
+                                       " c1 text," +
+                                       " c2 text," +
+                                       " v1 text," +
+                                       " PRIMARY KEY(pk, c1, c2))");
+        loadLegacyTable("legacy_%s_with_illegal_cell_names_2%s", "ka", "");
+        ColumnFamilyStore cfs = Keyspace.open("legacy_tables").getColumnFamilyStore(table);
+        assertEquals(1, Iterables.size(cfs.getSSTables(SSTableSet.CANONICAL)));
+        cfs.forceMajorCompaction();
+        assertEquals(1, Iterables.size(cfs.getSSTables(SSTableSet.CANONICAL)));
+        SSTableReader sstable = Iterables.getFirst(cfs.getSSTables(SSTableSet.CANONICAL), null);
+        LivenessInfo livenessInfo = null;
+        try (ISSTableScanner scanner = sstable.getScanner())
+        {
+            while (scanner.hasNext())
+            {
+                try (UnfilteredRowIterator iter = scanner.next())
+                {
+                    while (iter.hasNext())
+                    {
+                        Unfiltered uf = iter.next();
+                        livenessInfo = ((Row)uf).primaryKeyLivenessInfo();
+                    }
+                }
+            }
+        }
+        assertNotNull(livenessInfo);
+        assertEquals(100, livenessInfo.timestamp());
     }
 
     @Test