Do not leak 2015 synthetic memtable Epoch

patch by Berenguer Blasi; reviewed by Caleb Rackliffe for CASSANDRA-18118

build.xml

@@ -1227,6 +1227,8 @@
         <!-- disable shrinks in quicktheories CASSANDRA-15554 -->
         <jvmarg value="-DQT_SHRINKS=0"/>
 	<optjvmargs/>
+	<!-- Uncomment to debug unittest, attach debugger to port 1416 -->
+	<!--jvmarg line="-agentlib:jdwp=transport=dt_socket,address=localhost:1416,server=y,suspend=y" /-->
         <classpath>
           <pathelement path="${java.class.path}"/>
           <pathelement location="${stress.build.classes}"/>

src/java/org/apache/cassandra/db/Memtable.java

@@ -66,6 +66,7 @@ public class Memtable implements Comparable<Memtable>
     private static final Logger logger = LoggerFactory.getLogger(Memtable.class);
 
     public static final MemtablePool MEMORY_POOL = createMemtableAllocatorPool();
+    public static final long NO_MIN_TIMESTAMP = -1;
 
     private static MemtablePool createMemtableAllocatorPool()
     {
@@ -161,6 +162,16 @@ public class Memtable implements Comparable<Memtable>
         this.columnsCollector = new ColumnsCollector(metadata.partitionColumns());
     }
 
+    @VisibleForTesting
+    public Memtable(CFMetaData metadata, long minTimestamp)
+    {
+        this.initialComparator = metadata.comparator;
+        this.cfs = null;
+        this.allocator = null;
+        this.columnsCollector = new ColumnsCollector(metadata.partitionColumns());
+        this.minTimestamp = minTimestamp;
+    }
+
     public MemtableAllocator getAllocator()
     {
         return allocator;
@@ -387,9 +398,16 @@ public class Memtable implements Comparable<Memtable>
         return partitions.get(key);
     }
 
+    /**
+     * Returns the minTS if one available, otherwise NO_MIN_TIMESTAMP.
+     *
+     * EncodingStats uses a synthetic epoch TS at 2015. We don't want to leak that (CASSANDRA-18118) so we return NO_MIN_TIMESTAMP instead.
+     *
+     * @return The minTS or NO_MIN_TIMESTAMP if none available
+     */
     public long getMinTimestamp()
     {
-        return minTimestamp;
+        return minTimestamp != EncodingStats.NO_STATS.minTimestamp ? minTimestamp : NO_MIN_TIMESTAMP;
     }
 
     /**

src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java

@@ -713,7 +713,8 @@ public class SinglePartitionReadCommand extends ReadCommand
                 if (partition == null)
                     continue;
 
-                minTimestamp = Math.min(minTimestamp, memtable.getMinTimestamp());
+                if (memtable.getMinTimestamp() != Memtable.NO_MIN_TIMESTAMP)
+                    minTimestamp = Math.min(minTimestamp, memtable.getMinTimestamp());
 
                 @SuppressWarnings("resource") // 'iter' is added to iterators which is closed on exception, or through the closing of the final merged iterator
                 UnfilteredRowIterator iter = filter.getUnfilteredRowIterator(columnFilter(), partition);

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

@@ -207,7 +207,10 @@ public class CompactionController implements AutoCloseable
         }
 
         for (Memtable memtable : cfStore.getTracker().getView().getAllMemtables())
-            minTimestamp = Math.min(minTimestamp, memtable.getMinTimestamp());
+        {
+            if (memtable.getMinTimestamp() != Memtable.NO_MIN_TIMESTAMP)
+                minTimestamp = Math.min(minTimestamp, memtable.getMinTimestamp());
+        }
 
         // At this point, minTimestamp denotes the lowest timestamp of any relevant
         // SSTable or Memtable that contains a constructive value. candidates contains all the
@@ -281,11 +284,14 @@ public class CompactionController implements AutoCloseable
 
         for (Memtable memtable : memtables)
         {
-            Partition partition = memtable.getPartition(key);
-            if (partition != null)
+            if (memtable.getMinTimestamp() != Memtable.NO_MIN_TIMESTAMP)
             {
-                minTimestampSeen = Math.min(minTimestampSeen, partition.stats().minTimestamp);
-                hasTimestamp = true;
+                Partition partition = memtable.getPartition(key);
+                if (partition != null)
+                {
+                    minTimestampSeen = Math.min(minTimestampSeen, partition.stats().minTimestamp);
+                    hasTimestamp = true;
+                }
             }
         }
 

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

@@ -98,6 +98,15 @@ public class ColumnFamilyStoreTest
         Keyspace.open(KEYSPACE2).getColumnFamilyStore(CF_STANDARD1).truncateBlocking();
     }
 
+    @Test
+    public void testMemtableTimestamp() throws Throwable
+    {
+        assertEquals(Memtable.NO_MIN_TIMESTAMP,
+                     (new Memtable(Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1).metadata,
+                                   EncodingStats.NO_STATS.minTimestamp))
+                     .getMinTimestamp());
+    }
+
     @Test
     // create two sstables, and verify that we only deserialize data from the most recent one
     public void testTimeSortedQuery()