Merge branch 'cassandra-4.1' into cassandra-5.0

CHANGES.txt

@@ -14,6 +14,7 @@ Merged from 4.1:
  * Internode legacy SSL storage port certificate is not hot reloaded on update (CASSANDRA-18681)
  * Nodetool paxos-only repair is no longer incremental (CASSANDRA-18466)
 Merged from 4.0:
+ * Improve performance of compactions when table does not have an index (CASSANDRA-18773)
  * JMH improvements - faster build and async profiler (CASSANDRA-18871)
  * Enable 3rd party JDK installations for Debian package (CASSANDRA-18844)
 Merged from 3.11:

src/java/org/apache/cassandra/db/compaction/CompactionIterator.java

@@ -206,6 +206,18 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
     {
         return new UnfilteredPartitionIterators.MergeListener()
         {
+            private boolean rowProcessingNeeded()
+            {
+                return (type == OperationType.COMPACTION || type == OperationType.MAJOR_COMPACTION)
+                       && controller.cfs.indexManager.handles(IndexTransaction.Type.COMPACTION);
+            }
+
+            @Override
+            public boolean preserveOrder()
+            {
+                return rowProcessingNeeded();
+            }
+
             public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions)
             {
                 int merged = 0;
@@ -221,12 +233,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
 
                 CompactionIterator.this.updateCounterFor(merged);
 
-                if ( (type != OperationType.COMPACTION && type != OperationType.MAJOR_COMPACTION) 
+                if (!rowProcessingNeeded())
-                    || !controller.cfs.indexManager.handles(IndexTransaction.Type.COMPACTION) ) 
-                {
                     return null;
-                }
+
-                
                 Columns statics = Columns.NONE;
                 Columns regulars = Columns.NONE;
                 for (int i=0, isize=versions.size(); i<isize; i++)

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

@@ -45,10 +45,30 @@ public abstract class UnfilteredPartitionIterators
 
     public interface MergeListener
     {
+        /**
+         * Returns true if the merger needs to preserve the position of sources within the merge when passing data to
+         * the listener. If false, the merger can avoid creating empty sources for non-present partitions and
+         * significantly speed up processing.
+         *
+         * @return True to preserve position of source iterators.
+         */
+        public default boolean preserveOrder() { return true; }
         public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions);
         public default void close() {}
 
-        public static MergeListener NOOP = (partitionKey, versions) -> UnfilteredRowIterators.MergeListener.NOOP;
+        public static MergeListener NOOP = new MergeListener()
+        {
+            @Override
+            public boolean preserveOrder()
+            {
+                return false;
+            }
+
+            public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions)
+            {
+                return UnfilteredRowIterators.MergeListener.NOOP;
+            }
+        };
     }
 
     @SuppressWarnings("resource") // The created resources are returned right away
@@ -108,6 +128,8 @@ public abstract class UnfilteredPartitionIterators
 
         final TableMetadata metadata = iterators.get(0).metadata();
 
+        final boolean preserveOrder = listener != null && listener.preserveOrder();
+
         final MergeIterator<UnfilteredRowIterator, UnfilteredRowIterator> merged = MergeIterator.get(iterators, partitionComparator, new MergeIterator.Reducer<UnfilteredRowIterator, UnfilteredRowIterator>()
         {
             private final List<UnfilteredRowIterator> toMerge = new ArrayList<>(iterators.size());
@@ -120,9 +142,16 @@ public abstract class UnfilteredPartitionIterators
                 partitionKey = current.partitionKey();
                 isReverseOrder = current.isReverseOrder();
 
-                // Note that because the MergeListener cares about it, we want to preserve the index of the iterator.
+                if (preserveOrder)
-                // Non-present iterator will thus be set to empty in getReduced.
+                {
-                toMerge.set(idx, current);
+                    // Note that because the MergeListener cares about it, we want to preserve the index of the iterator.
+                    // Non-present iterator will thus be set to empty in getReduced.
+                    toMerge.set(idx, current);
+                }
+                else
+                {
+                    toMerge.add(current);
+                }
             }
 
             @SuppressWarnings("resource")
@@ -132,17 +161,20 @@ public abstract class UnfilteredPartitionIterators
                                                                  ? null
                                                                  : listener.getRowMergeListener(partitionKey, toMerge);
 
-                // Make a single empty iterator object to merge, we don't need toMerge.size() copiess
+                if (preserveOrder)
-                UnfilteredRowIterator empty = null;
-
-                // Replace nulls by empty iterators
-                for (int i = 0; i < toMerge.size(); i++)
                 {
-                    if (toMerge.get(i) == null)
+                    // Make a single empty iterator object to merge, we don't need toMerge.size() copiess
+                    UnfilteredRowIterator empty = null;
+
+                    // Replace nulls by empty iterators
+                    for (int i = 0; i < toMerge.size(); i++)
                     {
-                        if (null == empty)
+                        if (toMerge.get(i) == null)
-                            empty = EmptyIterators.unfilteredRow(metadata, partitionKey, isReverseOrder);
+                        {
-                        toMerge.set(i, empty);
+                            if (null == empty)
+                                empty = EmptyIterators.unfilteredRow(metadata, partitionKey, isReverseOrder);
+                            toMerge.set(i, empty);
+                        }
                     }
                 }
 
@@ -152,8 +184,11 @@ public abstract class UnfilteredPartitionIterators
             protected void onKeyChange()
             {
                 toMerge.clear();
-                for (int i = 0; i < iterators.size(); i++)
+                if (preserveOrder)
-                    toMerge.add(null);
+                {
+                    for (int i = 0; i < iterators.size(); i++)
+                        toMerge.add(null);
+                }
             }
         });