Report deleted rows in rowIterator and fix columns

Author: blambov

src/java/org/apache/cassandra/db/partitions/TrieBackedPartition.java

@@ -269,21 +269,35 @@ public static TrieBackedPartition create(DecoratedKey partitionKey,
                : new WithEnsureOnHeap(partitionKey, columnMetadata, encodingStats, rowCountIncludingStatic, tombstoneCount, trie, metadata, ensureOnHeap);
     }
 
-    class RowIterator extends TrieEntriesIterator<Object, Row>
+    class RowIterator extends TrieEntriesIterator.WithNullFiltering<Object, Row>
     {
         public RowIterator(DeletionAwareTrie<Object, TrieTombstoneMarker> trie, Direction direction)
         {
-            super(trie.contentOnlyTrie(), direction, RowData.class::isInstance);
+            super(trie.mergedTrie(TrieBackedPartition::combineDataAndDeletion), direction);
         }
 
         @Override
         protected Row mapContent(Object content, byte[] bytes, int byteLength)
         {
-            var rd = (RowData) content;
-            return toRow(rd,
-                         metadata.comparator.clusteringFromByteComparable(
-                             ByteBufferAccessor.instance,
-                             ByteComparable.preencoded(BYTE_COMPARABLE_VERSION, bytes, 0, byteLength)));
+            if (content instanceof RowData)
+                return toRow((RowData) content,
+                             getClustering(bytes, byteLength));
+            if (content instanceof Row)
+            {
+                BTreeRow row = (BTreeRow) content;
+                return BTreeRow.create(getClustering(bytes, byteLength),
+                                       row.primaryKeyLivenessInfo(),
+                                       row.deletion(),
+                                       row.getBTree(),
+                                       row.getMinLocalDeletionTime());
+            }
+
+            TrieTombstoneMarker marker = (TrieTombstoneMarker) content;
+            if (marker.hasPointData())
+                return BTreeRow.emptyDeletedRow(getClustering(bytes, byteLength),
+                                                Row.Deletion.regular(marker.deletionTime()));
+            else
+                return null;
         }
     }
 
@@ -513,6 +527,14 @@ public static Object combineDataAndDeletion(Object data, TrieTombstoneMarker del
         return rowData.toRow(Clustering.EMPTY, deletion.deletionTime());
     }
 
+    private Clustering<?> getClustering(byte[] bytes, int byteLength)
+    {
+        return metadata.comparator.clusteringFromByteComparable(ByteBufferAccessor.instance,
+                                                                ByteComparable.preencoded(BYTE_COMPARABLE_VERSION,
+                                                                                          bytes, 0, byteLength),
+                                                                BYTE_COMPARABLE_VERSION);
+    }
+
     /// Implementation of [UnfilteredRowIterator] for this partition.
     ///
     /// Currently, this implementation is pretty involved because it has to revert the transformations done to row and
@@ -573,14 +595,6 @@ protected Unfiltered mapContent(Object content, byte[] bytes, int byteLength)
                     partitionLevelDeletion);
         }
 
-        private Clustering<?> getClustering(byte[] bytes, int byteLength)
-        {
-            return metadata.comparator.clusteringFromByteComparable(ByteBufferAccessor.instance,
-                                                                    ByteComparable.preencoded(BYTE_COMPARABLE_VERSION,
-                                                                                              bytes, 0, byteLength),
-                                                                    BYTE_COMPARABLE_VERSION);
-        }
-
         @Override
         public DeletionTime partitionLevelDeletion()
         {
@@ -608,7 +622,7 @@ public boolean isReverseOrder()
         @Override
         public RegularAndStaticColumns columns()
         {
-            return columns;
+            return selection.fetchedColumns();
         }
 
         @Override

test/unit/org/apache/cassandra/db/RowUpdateBuilder.java

@@ -72,6 +72,17 @@ public RowUpdateBuilder(TableMetadata metadata, int localDeletionTime, long time
         this.updateBuilder.nowInSec(localDeletionTime);
     }
 
+    public RowUpdateBuilder(TableMetadata metadata, DeletionTime partitionDeletion, int nowInSec, long timestamp, Object partitionKey)
+    {
+        this(PartitionUpdate.simpleBuilder(metadata, partitionKey));
+
+        this.updateBuilder.timestamp(partitionDeletion.markedForDeleteAt());
+        this.updateBuilder.nowInSec(partitionDeletion.localDeletionTime());
+        this.updateBuilder.delete();
+        this.updateBuilder.timestamp(timestamp);
+        this.updateBuilder.nowInSec(nowInSec);
+    }
+
     public RowUpdateBuilder timestamp(long ts)
     {
         updateBuilder.timestamp(ts);

test/unit/org/apache/cassandra/service/reads/DataResolverTest.java

@@ -658,10 +658,9 @@ public void testRepairRangeTombstoneWithPartitionDeletion2()
 
         // 1st "stream": a partition deletion and a range tombstone
         RangeTombstone rt1 = tombstone("0", true , "9", true, 11, nowInSec);
-        PartitionUpdate upd1 = new RowUpdateBuilder(cfm, nowInSec, 1L, dk)
+        PartitionUpdate upd1 = new RowUpdateBuilder(cfm, new DeletionTime(10, nowInSec), nowInSec, 1L, dk)
                                .addRangeTombstone(rt1)
                                .buildUpdate();
-        ((MutableDeletionInfo)upd1.deletionInfo()).add(new DeletionTime(10, nowInSec));
         UnfilteredPartitionIterator iter1 = iter(upd1);
 
         // 2nd "stream": a range tombstone that is covered by the other stream rt