CNDB-14602: Fix bytes-based paging for partition deletions (#1835)

Only preserve the original data size or rows in case of purging.

Fixes DBPE-16935.

Author: adelapena

src/java/org/apache/cassandra/db/filter/DataLimits.java

@@ -628,14 +628,14 @@ public void applyToPartition(DecoratedKey partitionKey, Row staticRow)
             {
                 rowsInCurrentPartition = 0;
                 hasLiveStaticRow = !staticRow.isEmpty() && isLive(staticRow);
-                staticRowBytes = hasLiveStaticRow && bytesLimit != NO_LIMIT ? staticRow.originalDataSize() : 0;
+                staticRowBytes = hasLiveStaticRow && bytesLimit != NO_LIMIT ? staticRow.dataSizeBeforePurge() : 0;
             }
 
             @Override
             public Row applyToRow(Row row)
             {
                 if (isLive(row))
-                    incrementRowCount(bytesLimit != NO_LIMIT ? row.originalDataSize() : 0);
+                    incrementRowCount(bytesLimit != NO_LIMIT ? row.dataSizeBeforePurge() : 0);
                 return row;
             }
 
@@ -1110,7 +1110,7 @@ public void applyToPartition(DecoratedKey partitionKey, Row staticRow)
                     }
                     hasReturnedRowsFromCurrentPartition = false;
                     hasLiveStaticRow = !staticRow.isEmpty() && isLive(staticRow);
-                    staticRowBytes = hasLiveStaticRow ? staticRow.originalDataSize() : 0;
+                    staticRowBytes = hasLiveStaticRow ? staticRow.dataSizeBeforePurge() : 0;
                 }
                 currentPartitionKey = partitionKey;
                 // If we are done we need to preserve the groupInCurrentPartition and rowsCountedInCurrentPartition
@@ -1176,7 +1176,7 @@ public Row applyToRow(Row row)
                 if (isLive(row))
                 {
                     hasUnfinishedGroup = true;
-                    incrementRowCount(bytesLimit != NO_LIMIT ? row.originalDataSize() : 0);
+                    incrementRowCount(bytesLimit != NO_LIMIT ? row.dataSizeBeforePurge() : 0);
                     hasReturnedRowsFromCurrentPartition = true;
                 }
 

src/java/org/apache/cassandra/db/rows/BTreeRow.java

@@ -93,22 +93,22 @@ public class BTreeRow extends AbstractRow
     /**
      * The original data size of this row before purging it, or -1 if it hasn't been purged.
      */
-    private final int originalDataSize;
+    private final int dataSizeBeforePurge;
 
     private BTreeRow(Clustering<?> clustering,
                      LivenessInfo primaryKeyLivenessInfo,
                      Deletion deletion,
                      Object[] btree,
                      long minLocalDeletionTime,
-                     int originalDataSize)
+                     int dataSizeBeforePurge)
     {
         assert !deletion.isShadowedBy(primaryKeyLivenessInfo);
         this.clustering = clustering;
         this.primaryKeyLivenessInfo = primaryKeyLivenessInfo;
         this.deletion = deletion;
         this.btree = btree;
         this.minLocalDeletionTime = minLocalDeletionTime;
-        this.originalDataSize = originalDataSize;
+        this.dataSizeBeforePurge = dataSizeBeforePurge;
     }
 
     private BTreeRow(Clustering<?> clustering,
@@ -146,9 +146,9 @@ public static BTreeRow create(Clustering<?> clustering,
                                   Deletion deletion,
                                   Object[] btree,
                                   long minDeletionTime,
-                                  int originalDataSize)
+                                  int dataSizeBeforePurge)
     {
-        return new BTreeRow(clustering, primaryKeyLivenessInfo, deletion, btree, minDeletionTime, originalDataSize);
+        return new BTreeRow(clustering, primaryKeyLivenessInfo, deletion, btree, minDeletionTime, dataSizeBeforePurge);
     }
 
     public static BTreeRow create(Clustering<?> clustering,
@@ -522,7 +522,8 @@ public Row purge(DeletionPurger purger, long nowInSec, boolean enforceStrictLive
         if (enforceStrictLiveness && newDeletion.isLive() && newInfo.isEmpty())
             return null;
 
-        return transformAndFilter(newInfo, newDeletion, (cd) -> cd.purge(purger, nowInSec));
+        Function<ColumnData, ColumnData> columnDataPurger = (cd) -> cd.purge(purger, nowInSec);
+        return update(newInfo, newDeletion, BTree.transformAndFilter(btree, columnDataPurger), true);
     }
 
     public Row purgeDataOlderThan(long timestamp, boolean enforceStrictLiveness)
@@ -540,10 +541,10 @@ public Row purgeDataOlderThan(long timestamp, boolean enforceStrictLiveness)
     @Override
     public Row transformAndFilter(LivenessInfo info, Deletion deletion, Function<ColumnData, ColumnData> function)
     {
-        return update(info, deletion, BTree.transformAndFilter(btree, function));
+        return update(info, deletion, BTree.transformAndFilter(btree, function), false);
     }
 
-    private Row update(LivenessInfo info, Deletion deletion, Object[] newTree)
+    private Row update(LivenessInfo info, Deletion deletion, Object[] newTree, boolean preserveDataSizeBeforePurge)
     {
         if (btree == newTree && info == this.primaryKeyLivenessInfo && deletion == this.deletion)
             return this;
@@ -552,7 +553,9 @@ private Row update(LivenessInfo info, Deletion deletion, Object[] newTree)
             return null;
 
         long minDeletionTime = minDeletionTime(newTree, info, deletion.time());
-        return BTreeRow.create(clustering, info, deletion, newTree, minDeletionTime, originalDataSize());
+
+        int dataSizeBeforePurge = preserveDataSizeBeforePurge ? dataSizeBeforePurge() : -1;
+        return BTreeRow.create(clustering, info, deletion, newTree, minDeletionTime, dataSizeBeforePurge);
     }
 
     @Override
@@ -563,7 +566,7 @@ public Row transformAndFilter(Function<ColumnData, ColumnData> function)
 
     public Row transform(Function<ColumnData, ColumnData> function)
     {
-        return update(primaryKeyLivenessInfo, deletion, BTree.transform(btree, function));
+        return update(primaryKeyLivenessInfo, deletion, BTree.transform(btree, function), false);
     }
 
     @Override
@@ -595,9 +598,9 @@ public long unsharedHeapSize()
     }
 
     @Override
-    public int originalDataSize()
+    public int dataSizeBeforePurge()
     {
-        return originalDataSize >= 0 ? originalDataSize : dataSize();
+        return dataSizeBeforePurge >= 0 ? dataSizeBeforePurge : dataSize();
     }
 
     public long unsharedHeapSizeExcludingData()

src/java/org/apache/cassandra/db/rows/Row.java

@@ -331,7 +331,7 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
      *
      * @return the original data size of this row in bytes before purging
      */
-    int originalDataSize();
+    int dataSizeBeforePurge();
 
     public long unsharedHeapSizeExcludingData();
 

src/java/org/apache/cassandra/index/sai/utils/RowWithSourceTable.java

@@ -286,9 +286,9 @@ public int dataSize()
     }
 
     @Override
-    public int originalDataSize()
+    public int dataSizeBeforePurge()
     {
-        return row.originalDataSize();
+        return row.dataSizeBeforePurge();
     }
 
     @Override