revert r1078803 (#1902)

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079812 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -7,7 +7,6 @@
  * add CL.TWO, CL.THREE (CASSANDRA-2013)
  * avoid exporting an un-requested row in sstable2json, when exporting 
    a key that does not exist (CASSANDRA-2168)
- * track and migrate cached pages during compaction (CASSANDRA-1902)
  * add incremental_backups option (CASSANDRA-1872)
  * add configurable row limit to Pig loadfunc (CASSANDRA-2276)
  * validate column values in batches as well as single-Column inserts

conf/cassandra.yaml

@@ -251,10 +251,6 @@ in_memory_compaction_limit_in_mb: 64
 # key caches.
 compaction_preheat_key_cache: true
 
-# When set to true, this setting lets cassandra inform the OS to pre-cache
-# popular data in newly compacted files. This requires Linux and JNA be installed.
-enable_page_cache_migration: false
-
 # Time to wait for a reply from other nodes before failing the command 
 rpc_timeout_in_ms: 10000
 

src/java/org/apache/cassandra/config/Config.java

@@ -111,10 +111,6 @@ public class Config
     public int hinted_handoff_throttle_delay_in_ms = 0;
     public boolean compaction_preheat_key_cache = true;
 
-    // make this configurable as its being releases in a maintenance release
-    // TODO: remove in 0.8
-    public Boolean enable_page_cache_migration = false;
-
     public boolean incremental_backups = false;
 
     public static enum CommitLogSync {

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1214,11 +1214,6 @@ public class    DatabaseDescriptor
             throw new ConfigurationException("memtable_flush_after_mins must be greater than 0.");
     }
 
-    public static boolean isPageCaheMigrationEnabled()
-    {
-        return conf.enable_page_cache_migration;
-    }
-
     public static boolean incrementalBackupsEnabled()
     {
         return conf.incremental_backups;

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

@@ -46,7 +46,6 @@ public class Column implements IColumn
         return new ColumnSerializer();
     }
 
-    protected boolean isInPageCache;
     protected final ByteBuffer name;
     protected final ByteBuffer value;
     protected final long timestamp;
@@ -69,7 +68,6 @@ public class Column implements IColumn
         this.name = name;
         this.value = value;
         this.timestamp = timestamp;
-        isInPageCache = false;
     }
 
     public ByteBuffer name()
@@ -234,15 +232,5 @@ public class Column implements IColumn
     {
         return !isMarkedForDelete();
     }
-
-    public boolean isInPageCache()
-    {
-        return isInPageCache;
-    }
-
-    public void setIsInPageCache(boolean isInPageCache)
-    {
-        this.isInPageCache = isInPageCache;
-    }
 }
 

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

@@ -215,13 +215,6 @@ public class ColumnFamily implements IColumnContainer, IIterableColumns
         IColumn oldColumn;
         while ((oldColumn = columns.putIfAbsent(name, column)) != null)
         {
-            // migrate any page cache info (prefer cached)
-            if (oldColumn.isInPageCache() || column.isInPageCache())
-            {
-                oldColumn.setIsInPageCache(true);
-                column.setIsInPageCache(true);
-            }
-
             if (oldColumn instanceof SuperColumn)
             {
                 ((SuperColumn) oldColumn).putColumn(column);

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

@@ -21,7 +21,9 @@ package org.apache.cassandra.db;
  */
 
 
-import java.io.*;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
 import java.util.Collection;
 
 import org.slf4j.Logger;
@@ -29,8 +31,6 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.io.ICompactSerializer2;
-import org.apache.cassandra.io.util.PageCacheInformer;
-import org.apache.cassandra.utils.PageCacheMetrics;
 
 public class ColumnFamilySerializer implements ICompactSerializer2<ColumnFamily>
 {
@@ -70,16 +70,11 @@ public class ColumnFamilySerializer implements ICompactSerializer2<ColumnFamily>
         {
             throw new RuntimeException(e);
         }
-
         serializeForSSTable(columnFamily, dos);
     }
 
     public int serializeForSSTable(ColumnFamily columnFamily, DataOutput dos)
     {
-        PageCacheInformer pci = dos instanceof PageCacheInformer
-                                ? (PageCacheInformer) dos
-                                : null;
-
         try
         {
             serializeCFInfo(columnFamily, dos);
@@ -89,16 +84,8 @@ public class ColumnFamilySerializer implements ICompactSerializer2<ColumnFamily>
             dos.writeInt(count);
             for (IColumn column : columns)
             {
-                long startAt = pci != null ? pci.getCurrentPosition() : -1;
-
                 columnFamily.getColumnSerializer().serialize(column, dos);
-
-                //Track the section of serialized data that should
-                //be included in the page cache (compaction)
-                if (column.isInPageCache() && pci != null)
-                    pci.keepCacheWindow(startAt);
             }
-
             return count;
         }
         catch (IOException e)
@@ -130,49 +117,18 @@ public class ColumnFamilySerializer implements ICompactSerializer2<ColumnFamily>
             throw new UnserializableColumnFamilyException("Couldn't find cfId=" + cfId, cfId);
         ColumnFamily cf = ColumnFamily.create(cfId);
         deserializeFromSSTableNoColumns(cf, dis);
-        deserializeColumns(dis, cf, null);
+        deserializeColumns(dis, cf);
         return cf;
     }
 
-    public boolean deserializeColumns(DataInput dis, ColumnFamily cf, PageCacheMetrics pageCacheMetrics) throws IOException
+    public void deserializeColumns(DataInput dis, ColumnFamily cf) throws IOException
     {
-
         int size = dis.readInt();
-
+        for (int i = 0; i < size; ++i)
-        boolean hasColumnsInPageCache = false;
-
-
-        if (pageCacheMetrics != null && dis instanceof RandomAccessFile)
         {
-            RandomAccessFile raf = (RandomAccessFile) dis;
+            IColumn column = cf.getColumnSerializer().deserialize(dis);
-
+            cf.addColumn(column);
-            for (int i = 0; i < size; ++i)
-            {
-                long startAt = raf.getFilePointer();
-
-                IColumn column = cf.getColumnSerializer().deserialize(dis);
-
-                long endAt = raf.getFilePointer();
-
-                column.setIsInPageCache(pageCacheMetrics.isRangeInCache(startAt, endAt));
-
-                if(!hasColumnsInPageCache)
-                    hasColumnsInPageCache = column.isInPageCache();
-
-                cf.addColumn(column);
-            }
         }
-        else
-        {
-            for (int i = 0; i < size; ++i)
-            {
-                IColumn column = cf.getColumnSerializer().deserialize(dis);
-
-                cf.addColumn(column);
-            }
-        }
-
-        return hasColumnsInPageCache;
     }
 
     public ColumnFamily deserializeFromSSTableNoColumns(ColumnFamily cf, DataInput input) throws IOException

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

@@ -32,6 +32,7 @@ import javax.management.ObjectName;
 
 import com.google.common.collect.Iterables;
 import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -334,7 +335,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         {
             public void run()
             {
-                logger.info("Creating index {}.{}", table, indexedCfMetadata.cfName);
                 try
                 {
                     forceBlockingFlush();
@@ -348,7 +348,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
                     throw new AssertionError(e);
                 }
                 buildSecondaryIndexes(getSSTables(), FBUtilities.singleton(info.name));
-                logger.info("Index {} complete", indexedCfMetadata.cfName);
                 SystemTable.setIndexBuilt(table.name, indexedCfMetadata.cfName);
             }
         };
@@ -357,7 +356,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public void buildSecondaryIndexes(Collection<SSTableReader> sstables, SortedSet<ByteBuffer> columns)
     {
-        logger.debug("Submitting index build to compactionmanager");
+        logger.info(String.format("Submitting index build of %s for data in %s",
+                                  metadata.comparator.getString(columns), StringUtils.join(sstables, ", ")));
         Table.IndexBuilder builder = table.createIndexBuilder(this, columns, new ReducingKeyIterator(sstables));
         Future future = CompactionManager.instance.submitIndexBuild(this, builder);
         try
@@ -374,6 +374,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         {
             throw new RuntimeException(e);
         }
+        logger.info("Index build of " + metadata.comparator.getString(columns) + " complete");
     }
 
     // called when dropping or renaming a CF. Performs mbean housekeeping and invalidates CFS to other operations.
@@ -684,26 +685,31 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         try
         {
             if (oldMemtable.isFrozen())
+            {
+                logger.debug("memtable is already frozen; another thread must be flushing it");
                 return null;
+            }
 
             boolean isDropped = isIndex()
                               ? DatabaseDescriptor.getCFMetaData(table.name, getParentColumnfamily()) == null
                               : DatabaseDescriptor.getCFMetaData(metadata.cfId) == null;
             if (isDropped)
-                return null; // column family was dropped. no point in flushing.
+            {
+                logger.debug("column family was dropped; no point in flushing");
+                return null;
+            }
 
             assert memtable == oldMemtable;
             memtable.freeze();
             final CommitLogSegment.CommitLogContext ctx = writeCommitLog ? CommitLog.instance.getContext() : null;
-            logger.info("switching in a fresh Memtable for " + columnFamily + " at " + ctx);
 
             // submit the memtable for any indexed sub-cfses, and our own.
             List<ColumnFamilyStore> icc = new ArrayList<ColumnFamilyStore>(indexedColumns.size());
-            icc.add(this);
+            // don't assume that this.memtable is dirty; forceFlush can bring us here during index build even if it is not
-            for (ColumnFamilyStore indexCfs : indexedColumns.values())
+            for (ColumnFamilyStore cfs : Iterables.concat(Collections.singleton(this), indexedColumns.values()))
             {
-                if (!indexCfs.memtable.isClean())
+                if (!cfs.memtable.isClean())
-                    icc.add(indexCfs);
+                    icc.add(cfs);
             }
             final CountDownLatch latch = new CountDownLatch(icc.size());
             for (ColumnFamilyStore cfs : icc)
@@ -711,6 +717,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
                 submitFlush(cfs.memtable, latch);
                 cfs.memtable = new Memtable(cfs);
             }
+            // we marked our memtable as frozen as part of the concurrency control,
+            // so even if there was nothing to flush we need to switch it out
+            if (!icc.contains(this))
+                memtable = new Memtable(this);
 
             // when all the memtables have been written, including for indexes, mark the flush in the commitlog header.
             // a second executor makes sure the onMemtableFlushes get called in the right order,
@@ -754,8 +764,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public Future<?> forceFlush()
     {
-        if (memtable.isClean())
+        // during index build, 2ary index memtables can be dirty even if parent is not.  if so,
+        // we want flushLargestMemtables to flush the 2ary index ones too.
+        boolean clean = true;
+        for (ColumnFamilyStore cfs : Iterables.concat(Collections.singleton(this), getIndexColumnFamilyStores()))
+            clean &= cfs.memtable.isClean();
+
+        if (clean)
+        {
+            logger.debug("forceFlush requested but everything is clean");
             return null;
+        }
 
         return maybeSwitchMemtable(memtable, true);
     }
@@ -1937,6 +1956,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         return indexedColumns.get(column);
     }
 
+    public Collection<ColumnFamilyStore> getIndexColumnFamilyStores()
+    {
+        return indexedColumns.values();
+    }
+
     public ColumnFamily newIndexedColumnFamily(ByteBuffer column)
     {
         return ColumnFamily.create(indexedColumns.get(column).metadata);
@@ -2161,11 +2185,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public SSTableWriter createFlushWriter(long estimatedRows) throws IOException
     {
-        return new SSTableWriter(getFlushPath(), estimatedRows, metadata, partitioner, false);
+        return new SSTableWriter(getFlushPath(), estimatedRows, metadata, partitioner);
     }
 
-    public SSTableWriter createCompactionWriter(long estimatedRows, String location, boolean migratePageCache) throws IOException
+    public SSTableWriter createCompactionWriter(long estimatedRows, String location) throws IOException
     {
-        return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner, migratePageCache);
+        return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner);
     }
 }

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

@@ -46,7 +46,6 @@ import org.apache.cassandra.io.*;
 import org.apache.cassandra.io.sstable.*;
 import org.apache.cassandra.io.util.BufferedRandomAccessFile;
 import org.apache.cassandra.io.util.FileUtils;
-import org.apache.cassandra.io.util.PageCacheInformer;
 import org.apache.cassandra.service.AntiEntropyService;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -390,7 +389,6 @@ public class CompactionManager implements CompactionManagerMBean
         assert sstables != null;
 
         Table table = cfs.table;
-
         if (DatabaseDescriptor.isSnapshotBeforeCompaction())
             table.snapshot("compact-" + cfs.columnFamily);
 
@@ -400,11 +398,9 @@ public class CompactionManager implements CompactionManagerMBean
             assert sstable.descriptor.cfname.equals(cfs.columnFamily);
 
         String compactionFileLocation = table.getDataFileLocation(cfs.getExpectedCompactedFileSize(sstables));
-
         // If the compaction file path is null that means we have no space left for this compaction.
         // try again w/o the largest one.
         List<SSTableReader> smallerSSTables = new ArrayList<SSTableReader>(sstables);
-
         while (compactionFileLocation == null && smallerSSTables.size() > 1)
         {
             logger.warn("insufficient space to compact all requested files " + StringUtils.join(smallerSSTables, ", "));
@@ -416,7 +412,6 @@ public class CompactionManager implements CompactionManagerMBean
             logger.error("insufficient space to compact even the two smallest files, aborting");
             return 0;
         }
-
         sstables = smallerSSTables;
 
         // new sstables from flush can be added during a compaction, but only the compaction can remove them,
@@ -428,9 +423,9 @@ public class CompactionManager implements CompactionManagerMBean
         long totalkeysWritten = 0;
 
         // TODO the int cast here is potentially buggy
-        int expectedBloomFilterSize = Math.max(DatabaseDescriptor.getIndexInterval(), (int) SSTableReader.getApproximateKeyCount(sstables));
+        int expectedBloomFilterSize = Math.max(DatabaseDescriptor.getIndexInterval(), (int)SSTableReader.getApproximateKeyCount(sstables));
         if (logger.isDebugEnabled())
-            logger.debug("Expected bloom filter size : " + expectedBloomFilterSize);
+          logger.debug("Expected bloom filter size : " + expectedBloomFilterSize);
 
         SSTableWriter writer;
         CompactionIterator ci = new CompactionIterator(cfs, sstables, gcBefore, major); // retain a handle so we can call close()
@@ -450,13 +445,11 @@ public class CompactionManager implements CompactionManagerMBean
                 return 0;
             }
 
-            writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, ci.hasRowsInPageCache());
+            writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation);
             while (nni.hasNext())
             {
                 AbstractCompactedRow row = nni.next();
-
                 long position = writer.append(row);
-
                 totalkeysWritten++;
 
                 if (DatabaseDescriptor.getPreheatKeyCache())
@@ -486,7 +479,7 @@ public class CompactionManager implements CompactionManagerMBean
         long dTime = System.currentTimeMillis() - startTime;
         long startsize = SSTable.getTotalBytes(sstables);
         long endsize = ssTable.length();
-        double ratio = (double) endsize / (double) startsize;
+        double ratio = (double)endsize / (double)startsize;
         logger.info(String.format("Compacted to %s.  %,d to %,d (~%d%% of original) bytes for %,d keys.  Time: %,dms.",
                                   writer.getFilename(), startsize, endsize, (int) (ratio * 100), totalkeysWritten, dTime));
         return sstables.size();
@@ -539,7 +532,7 @@ public class CompactionManager implements CompactionManagerMBean
                 assert firstRowPositionFromIndex == 0 : firstRowPositionFromIndex;
             }
 
-            SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null, false);
+            SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null);
             executor.beginCompaction(cfs.columnFamily, new ScrubInfo(dataFile, sstable));
             int goodRows = 0, badRows = 0;
 
@@ -594,7 +587,7 @@ public class CompactionManager implements CompactionManagerMBean
                         throw new IOError(new IOException("Unable to read row key from data file"));
                     if (dataSize > dataFile.length())
                         throw new IOError(new IOException("Impossible row size " + dataSize));
-                    SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStart, dataSize, null, true);
+                    SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStart, dataSize, true);
                     writer.append(getCompactedRow(row, cfs, sstable.descriptor, true));
                     goodRows++;
                     if (!key.key.equals(currentIndexKey) || dataStart != dataStartFromIndex)
@@ -614,7 +607,7 @@ public class CompactionManager implements CompactionManagerMBean
                         key = SSTableReader.decodeKey(sstable.partitioner, sstable.descriptor, currentIndexKey);
                         try
                         {
-                            SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStartFromIndex, dataSizeFromIndex, null, true);
+                            SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStartFromIndex, dataSizeFromIndex, true);
                             writer.append(getCompactedRow(row, cfs, sstable.descriptor, true));
                             goodRows++;
                         }
@@ -699,7 +692,7 @@ public class CompactionManager implements CompactionManagerMBean
                     SSTableIdentityIterator row = (SSTableIdentityIterator) scanner.next();
                     if (Range.isTokenInRanges(row.getKey().token, ranges))
                     {
-                        writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer, row.hasRowsInPageCache());
+                        writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer);
                         writer.append(getCompactedRow(row, cfs, sstable.descriptor, false));
                         totalkeysWritten++;
                     }
@@ -768,13 +761,13 @@ public class CompactionManager implements CompactionManagerMBean
                : new PrecompactedRow(cfs, Arrays.asList(row), false, getDefaultGcBefore(cfs), forceDeserialize);
     }
 
-    private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer, boolean migrateCachedPages)
+    private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer)
             throws IOException
     {
         if (writer == null)
         {
             FileUtils.createDirectory(compactionFileLocation);
-            writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, migrateCachedPages);
+            writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation);
         }
         return writer;
     }
@@ -1120,7 +1113,7 @@ public class CompactionManager implements CompactionManagerMBean
             this.row = row;
         }
 
-        public void write(PageCacheInformer out) throws IOException
+        public void write(DataOutput out) throws IOException
         {
             assert row.dataSize > 0;
             out.writeLong(row.dataSize);

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

@@ -55,12 +55,4 @@ public interface IColumn
      * supercolumn deleted-at time.
      */
     boolean isLive();
-
-    /**
-     * Used to identify columns during compaction that are in the os page cache
-     * so that they can be re-cached in new SSTables
-     * @return
-     */
-    boolean isInPageCache();
-    void setIsInPageCache(boolean isInPageCache);
 }

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

@@ -18,7 +18,10 @@
 
 package org.apache.cassandra.db;
 
-import java.io.*;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.ExecutionException;

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

@@ -53,7 +53,6 @@ public class SuperColumn implements IColumn, IColumnContainer
     private ConcurrentSkipListMap<ByteBuffer, IColumn> columns_;
     private AtomicInteger localDeletionTime = new AtomicInteger(Integer.MIN_VALUE);
     private AtomicLong markedForDeleteAt = new AtomicLong(Long.MIN_VALUE);
-    private boolean isInPageCache = false;
 
     public SuperColumn(ByteBuffer name, AbstractType comparator)
     {
@@ -307,16 +306,6 @@ public class SuperColumn implements IColumn, IColumnContainer
     {
         throw new UnsupportedOperationException("This operation is unsupported on super columns.");
     }
-
-    public boolean isInPageCache()
-    {
-        return isInPageCache;
-    }
-
-    public void setIsInPageCache(boolean isInPageCache)
-    {
-        this.isInPageCache = isInPageCache;
-    }
 }
 
 class SuperColumnSerializer implements ICompactSerializer2<IColumn>

src/java/org/apache/cassandra/db/commitlog/CommitLog.java

@@ -173,7 +173,7 @@ public class CommitLog
         for (File file : clogs)
         {
             int bufferSize = (int) Math.min(Math.max(file.length(), 1), 32 * 1024 * 1024);
-            BufferedRandomAccessFile reader = new BufferedRandomAccessFile(new File(file.getAbsolutePath()), "r", bufferSize, true, false);
+            BufferedRandomAccessFile reader = new BufferedRandomAccessFile(new File(file.getAbsolutePath()), "r", bufferSize, true);
 
             try
             {

src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java

@@ -73,7 +73,7 @@ public class CommitLogSegment
 
     private static BufferedRandomAccessFile createWriter(String file) throws IOException
     {
-        return new BufferedRandomAccessFile(new File(file), "rw", 128 * 1024, true, false);
+        return new BufferedRandomAccessFile(new File(file), "rw", 128 * 1024, true);
     }
 
     public CommitLogSegment.CommitLogContext write(RowMutation rowMutation, Object serializedRow) throws IOException

src/java/org/apache/cassandra/io/AbstractCompactedRow.java

@@ -21,11 +21,11 @@ package org.apache.cassandra.io;
  */
 
 
+import java.io.DataOutput;
 import java.io.IOException;
 import java.security.MessageDigest;
 
 import org.apache.cassandra.db.DecoratedKey;
-import org.apache.cassandra.io.util.PageCacheInformer;
 
 /**
  * a CompactedRow is an object that takes a bunch of rows (keys + columnfamilies)
@@ -35,7 +35,6 @@ import org.apache.cassandra.io.util.PageCacheInformer;
 public abstract class AbstractCompactedRow
 {
     public final DecoratedKey key;
-    protected boolean hasColumnsInPageCache = false;
 
     public AbstractCompactedRow(DecoratedKey key)
     {
@@ -45,7 +44,7 @@ public abstract class AbstractCompactedRow
     /**
      * write the row (size + column index + filter + column data, but NOT row key) to @param out
      */
-    public abstract void write(PageCacheInformer out) throws IOException;
+    public abstract void write(DataOutput out) throws IOException;
 
     /**
      * update @param digest with the data bytes of the row (not including row key or row size)
@@ -61,18 +60,4 @@ public abstract class AbstractCompactedRow
      * @return the number of columns in the row
      */
     public abstract int columnCount();
-
-    /**
-     * @return if any columns in this row are in the OS Page Cache
-     */
-    public boolean hasColumnsInPageCache()
-    {
-        return hasColumnsInPageCache;
-    }
-
-    public void setHasColumnsInPageCache(boolean hasColumnsInPageCache)
-    {
-        this.hasColumnsInPageCache = hasColumnsInPageCache;
-    }
-
 }

src/java/org/apache/cassandra/io/CompactionIterator.java

@@ -48,7 +48,6 @@ implements Closeable, ICompactionInfo
     public static final int FILE_BUFFER_SIZE = 1024 * 1024;
 
     protected final List<SSTableIdentityIterator> rows = new ArrayList<SSTableIdentityIterator>();
-
     private final ColumnFamilyStore cfs;
     private final int gcBefore;
     private final boolean major;
@@ -56,7 +55,6 @@ implements Closeable, ICompactionInfo
     private long totalBytes;
     private long bytesRead;
     private long row;
-    private boolean hasRowsInPageCache;
 
     public CompactionIterator(ColumnFamilyStore cfs, Iterable<SSTableReader> sstables, int gcBefore, boolean major) throws IOException
     {
@@ -99,9 +97,6 @@ implements Closeable, ICompactionInfo
     public void reduce(SSTableIdentityIterator current)
     {
         rows.add(current);
-
-        if(current.hasRowsInPageCache())
-            hasRowsInPageCache = true;
     }
 
     protected AbstractCompactedRow getReduced()
@@ -130,7 +125,6 @@ implements Closeable, ICompactionInfo
     protected AbstractCompactedRow getCompactedRow()
     {
         long rowSize = 0;
-
         for (SSTableIdentityIterator row : rows)
         {
             rowSize += row.dataSize;
@@ -168,11 +162,6 @@ implements Closeable, ICompactionInfo
         return bytesRead;
     }
 
-    public boolean hasRowsInPageCache()
-    {
-        return hasRowsInPageCache;
-    }
-
     public String getTaskType()
     {
         return major ? "Major" : "Minor";

src/java/org/apache/cassandra/io/LazilyCompactedRow.java

@@ -38,7 +38,8 @@ import org.apache.cassandra.db.IColumn;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
-import org.apache.cassandra.io.util.*;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.io.util.IIterableColumns;
 import org.apache.cassandra.utils.ReducingIterator;
 
 /**
@@ -93,7 +94,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
         iter = null;
     }
 
-    public void write(PageCacheInformer out) throws IOException
+    public void write(DataOutput out) throws IOException
     {
         if (rows.size() == 1 && !shouldPurge && rows.get(0).sstable.descriptor.isLatestVersion && !forceDeserialize)
         {

src/java/org/apache/cassandra/io/PrecompactedRow.java

@@ -21,6 +21,7 @@ package org.apache.cassandra.io;
  */
 
 
+import java.io.DataOutput;
 import java.io.IOError;
 import java.io.IOException;
 import java.security.MessageDigest;
@@ -37,8 +38,6 @@ import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
 import org.apache.cassandra.io.util.DataOutputBuffer;
-import org.apache.cassandra.io.util.PageCacheInformer;
-import org.apache.cassandra.utils.Pair;
 
 /**
  * PrecompactedRow merges its rows in its constructor in memory.
@@ -91,11 +90,7 @@ public class PrecompactedRow extends AbstractCompactedRow
                 {
                     cf.addAll(thisCF);
                 }
-
-                if (row.hasColumnsInPageCache())
-                    this.hasColumnsInPageCache = true;
             }
-
             ColumnFamily cfPurged = shouldPurge ? ColumnFamilyStore.removeDeleted(cf, gcBefore) : cf;
             if (cfPurged == null)
                 return;
@@ -116,47 +111,11 @@ public class PrecompactedRow extends AbstractCompactedRow
         }
     }
 
-    public void write(PageCacheInformer out) throws IOException
+    public void write(DataOutput out) throws IOException
     {
         assert buffer.getLength() > 0;
         out.writeLong(buffer.getLength());
-
+        out.write(buffer.getData(), 0, buffer.getLength());
-        List<Pair<Integer, Integer>> pageCacheMarkers = buffer.getPageCacheMarkers();
-
-        if(pageCacheMarkers == null)
-        {
-             out.write(buffer.getData(), 0, buffer.getLength());
-             return;
-        }
-
-        // Step through each page cache window and inform the
-        // output writer to respect these...
-        long startingPosition = out.getCurrentPosition();
-        int bufferOffset = 0;
-        for (Pair<Integer,Integer> window : pageCacheMarkers)
-        {
-            // write out any data before the window
-            if (window.left > (out.getCurrentPosition() - startingPosition))
-            {
-                out.write(buffer.getData(), bufferOffset, window.left - bufferOffset);
-                bufferOffset = window.left;
-            }
-
-            long startingAt = out.getCurrentPosition();
-
-            assert (bufferOffset + window.right) <= buffer.getLength() : ""+(bufferOffset + window.right)+" > "+buffer.getLength();
-
-            out.write(buffer.getData(), bufferOffset, window.right);
-            out.keepCacheWindow(startingAt);
-
-            bufferOffset += window.right;
-        }
-
-        // Write everything else
-        if (bufferOffset < buffer.getLength())
-        {
-            out.write(buffer.getData(), bufferOffset, buffer.getLength() - bufferOffset);
-        }
     }
 
     public void update(MessageDigest digest)

src/java/org/apache/cassandra/io/sstable/CacheWriter.java

@@ -75,7 +75,7 @@ public class CacheWriter<K, V> implements ICompactionInfo
         logger.debug("Saving {}", path);
         File tmpFile = File.createTempFile(path.getName(), null, path.getParentFile());
 
-        BufferedRandomAccessFile out = new BufferedRandomAccessFile(tmpFile, "rw", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, true, false);
+        BufferedRandomAccessFile out = new BufferedRandomAccessFile(tmpFile, "rw", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, true);
         try
         {
             for (K key : keys)

src/java/org/apache/cassandra/io/sstable/KeyIterator.java

@@ -47,8 +47,7 @@ public class KeyIterator extends AbstractIterator<DecoratedKey> implements Itera
             in = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_INDEX)),
                                               "r",
                                               BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE,
-                                              true,
+                                              true);
-                                              false);
         }
         catch (IOException e)
         {

src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java

@@ -35,8 +35,7 @@ import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.IColumn;
 import org.apache.cassandra.db.columniterator.IColumnIterator;
 import org.apache.cassandra.io.util.BufferedRandomAccessFile;
-import org.apache.cassandra.io.util.PageCacheInformer;
+import org.apache.cassandra.utils.Filter;
-import org.apache.cassandra.utils.PageCacheMetrics;
 
 public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterator>, IColumnIterator
 {
@@ -48,9 +47,6 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
     public final SSTableReader sstable;
     private final long dataStart;
     public final long dataSize;
-    public final PageCacheMetrics pageCacheMetrics;
-    private boolean hasRowsInPageCache;
-    private boolean hasColumnsInPageCache;
 
     private final ColumnFamily columnFamily;
     public final int columnCount;
@@ -65,13 +61,13 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
      * @param dataSize length of row data
      * @throws IOException
      */
-    public SSTableIdentityIterator(SSTableReader sstable, BufferedRandomAccessFile file, DecoratedKey key, long dataStart, long dataSize, PageCacheMetrics pageCacheMetrics)
+    public SSTableIdentityIterator(SSTableReader sstable, BufferedRandomAccessFile file, DecoratedKey key, long dataStart, long dataSize)
     throws IOException
     {
-        this(sstable, file, key, dataStart, dataSize, pageCacheMetrics, false);
+        this(sstable, file, key, dataStart, dataSize, false);
     }
 
-    public SSTableIdentityIterator(SSTableReader sstable, BufferedRandomAccessFile file, DecoratedKey key, long dataStart, long dataSize, PageCacheMetrics pageCacheMetrics, boolean deserializeRowHeader)
+    public SSTableIdentityIterator(SSTableReader sstable, BufferedRandomAccessFile file, DecoratedKey key, long dataStart, long dataSize, boolean deserializeRowHeader)
     throws IOException
     {
         this.sstable = sstable;
@@ -80,10 +76,6 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
         this.dataStart = dataStart;
         this.dataSize = dataSize;
         finishedAt = dataStart + dataSize;
-        this.pageCacheMetrics = pageCacheMetrics;
-
-        //Mark if any rows are in the pageCache
-        hasRowsInPageCache = (pageCacheMetrics != null) && pageCacheMetrics.isRangeInCache(dataStart, finishedAt);
 
         try
         {
@@ -145,19 +137,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
     {
         try
         {
-            long columnStartAt = file.getFilePointer();
+            return sstable.getColumnSerializer().deserialize(file);
-
-            IColumn col = sstable.getColumnSerializer().deserialize(file);
-
-            long columnEndAt = file.getFilePointer();
-
-            if (pageCacheMetrics != null)
-            {
-                col.setIsInPageCache(pageCacheMetrics.isRangeInCache(columnStartAt, columnEndAt));
-            }
-
-            return col;
-
         }
         catch (IOException e)
         {
@@ -180,56 +160,12 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
         return file.getPath();
     }
 
-    public void echoData(PageCacheInformer out) throws IOException
+    public void echoData(DataOutput out) throws IOException
     {
         file.seek(dataStart);
-
+        while (file.getFilePointer() < finishedAt)
-        if (pageCacheMetrics == null)
         {
-            while (file.getFilePointer() < finishedAt)
+            out.write(file.readByte());
-            {
-                out.write(file.readByte());
-            }
-        }
-        else
-        {
-            // Since this is just a big opaque block of data we
-            // Split into chunks >= pageSize
-            int chunkSize = (int) (finishedAt - dataStart) / 128;
-
-            chunkSize = chunkSize >= pageCacheMetrics.pageSize ? chunkSize : pageCacheMetrics.pageSize;
-
-            long chunkStart = 0;
-            long chunkEnd = 0;
-            boolean isChunkInPageCache = false;
-
-            while (file.getFilePointer() < finishedAt)
-            {
-
-                // Mark chunks that have cached pages
-                // So we can migrate them
-                if (file.getFilePointer() >= chunkEnd)
-                {
-                    if (isChunkInPageCache)
-                    {
-                        out.keepCacheWindow(out.getCurrentPosition() - chunkSize);
-                    }
-
-                    chunkStart = file.getFilePointer();
-                    chunkEnd = chunkStart + chunkSize;
-
-                    if(chunkEnd > finishedAt)
-                        chunkEnd = finishedAt;
-
-                    isChunkInPageCache = pageCacheMetrics.isRangeInCache(chunkStart, chunkEnd);
-                }
-
-
-                out.write(file.readByte());
-            }
-
-            if (isChunkInPageCache)
-                out.keepCacheWindow(out.getCurrentPosition() - (file.getFilePointer() - chunkStart));
         }
     }
 
@@ -237,9 +173,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
     {
         file.seek(columnPosition - 4); // seek to before column count int
         ColumnFamily cf = columnFamily.cloneMeShallow();
-
+        ColumnFamily.serializer().deserializeColumns(file, cf);
-        hasColumnsInPageCache = ColumnFamily.serializer().deserializeColumns(file, cf, pageCacheMetrics);
-
         return cf;
     }
 
@@ -259,15 +193,4 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
             throw new IOError(e);
         }
     }
-
-    public boolean hasRowsInPageCache()
-    {
-        return hasRowsInPageCache;
-    }
-
-    public boolean hasColumnsInPageCache()
-    {
-        return hasColumnsInPageCache;
-    }
-
 }

src/java/org/apache/cassandra/io/sstable/SSTableReader.java

@@ -276,9 +276,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         BufferedRandomAccessFile input = new BufferedRandomAccessFile(new File(descriptor.filenameFor(Component.PRIMARY_INDEX)),
                                                                       "r",
                                                                       BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE,
-                                                                      true,
+                                                                      true);
-                                                                      false);
-
         try
         {
             if (keyCache != null && keyCache.getCapacity() - keyCache.getSize() < keysToLoadInCache.size())
@@ -297,11 +295,9 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
                     break;
 
                 boolean shouldAddEntry = indexSummary.shouldAddEntry();
-
                 ByteBuffer key = (shouldAddEntry || cacheLoading || recreatebloom)
-                                 ? ByteBufferUtil.readWithShortLength(input)
+                             ? ByteBufferUtil.readWithShortLength(input)
-                                 : ByteBufferUtil.skipShortLength(input);
+                             : ByteBufferUtil.skipShortLength(input);
-
                 long dataPosition = input.readLong();
                 if (key != null)
                 {
@@ -553,7 +549,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
     }
 
    /**
-    * SSTableScanner that avoids polluting the page cache
+    * Direct I/O SSTableScanner
     * @param bufferSize Buffer size in bytes for this Scanner.
     * @return A Scanner for seeking over the rows of the SSTable.
     */

src/java/org/apache/cassandra/io/sstable/SSTableScanner.java

@@ -19,7 +19,10 @@
 
 package org.apache.cassandra.io.sstable;
 
-import java.io.*;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOError;
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.Iterator;
 
@@ -31,8 +34,6 @@ import org.apache.cassandra.db.columniterator.IColumnIterator;
 import org.apache.cassandra.db.filter.QueryFilter;
 import org.apache.cassandra.io.util.BufferedRandomAccessFile;
 import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.CLibrary;
-import org.apache.cassandra.utils.PageCacheMetrics;
 
 
 public class SSTableScanner implements Iterator<IColumnIterator>, Closeable
@@ -41,9 +42,6 @@ public class SSTableScanner implements Iterator<IColumnIterator>, Closeable
 
     private final BufferedRandomAccessFile file;
     private final SSTableReader sstable;
-
-    private final PageCacheMetrics pageCacheMetrics;
-
     private IColumnIterator row;
     private boolean exhausted = false;
     private Iterator<IColumnIterator> iterator;
@@ -54,25 +52,15 @@ public class SSTableScanner implements Iterator<IColumnIterator>, Closeable
      */
     SSTableScanner(SSTableReader sstable, int bufferSize, boolean skipCache)
     {
-        PageCacheMetrics pageCacheMetrics = null;
-
         try
         {
-            File sstableFile = new File(sstable.getFilename());
+            this.file = new BufferedRandomAccessFile(new File(sstable.getFilename()), "r", bufferSize, skipCache);
-            file = new BufferedRandomAccessFile(sstableFile, "r", bufferSize, skipCache, false);
-
-            if (skipCache)
-            {
-                pageCacheMetrics = CLibrary.getCachedPages(file);
-            }
         }
         catch (IOException e)
         {
             throw new IOError(e);
         }
-
         this.sstable = sstable;
-        this.pageCacheMetrics = pageCacheMetrics;
     }
 
     /**
@@ -91,7 +79,6 @@ public class SSTableScanner implements Iterator<IColumnIterator>, Closeable
         }
         this.sstable = sstable;
         this.filter = filter;
-        this.pageCacheMetrics = null;
     }
 
     public void close() throws IOException
@@ -189,7 +176,8 @@ public class SSTableScanner implements Iterator<IColumnIterator>, Closeable
 
                 if (filter == null)
                 {
-                    return row = new SSTableIdentityIterator(sstable, file, key, dataStart, dataSize, pageCacheMetrics);
+                    row = new SSTableIdentityIterator(sstable, file, key, dataStart, dataSize);
+                    return row;
                 }
                 else
                 {

src/java/org/apache/cassandra/io/sstable/SSTableWriter.java

@@ -56,16 +56,16 @@ public class SSTableWriter extends SSTable
 
     private IndexWriter iwriter;
     private SegmentedFile.Builder dbuilder;
-    public  final BufferedRandomAccessFile dataFile;
+    private final BufferedRandomAccessFile dataFile;
     private DecoratedKey lastWrittenKey;
     private FileMark dataMark;
 
     public SSTableWriter(String filename, long keyCount) throws IOException
     {
-        this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner(), false);
+        this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner());
     }
 
-    public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner, boolean migratePageCache) throws IOException
+    public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner) throws IOException
     {
         super(Descriptor.fromFilename(filename),
               new HashSet<Component>(Arrays.asList(Component.DATA, Component.FILTER, Component.PRIMARY_INDEX, Component.STATS)),
@@ -73,10 +73,9 @@ public class SSTableWriter extends SSTable
               partitioner,
               SSTable.defaultRowHistogram(),
               SSTable.defaultColumnHistogram());
-
+        iwriter = new IndexWriter(descriptor, partitioner, keyCount);
-        iwriter = new IndexWriter(descriptor, partitioner, keyCount, !migratePageCache); //when we migrate pages we cache the index
         dbuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
-        dataFile = new BufferedRandomAccessFile(new File(getFilename()), "rw", DatabaseDescriptor.getInMemoryCompactionLimit(), true, migratePageCache);
+        dataFile = new BufferedRandomAccessFile(new File(getFilename()), "rw", DatabaseDescriptor.getInMemoryCompactionLimit(), true);
     }
     
     public void mark()
@@ -255,7 +254,7 @@ public class SSTableWriter extends SSTable
             cfs = Table.open(desc.ksname).getColumnFamilyStore(desc.cfname);
             try
             {
-                dfile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "r", 8 * 1024 * 1024, true, false);
+                dfile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "r", 8 * 1024 * 1024, true);
             }
             catch (IOException e)
             {
@@ -280,7 +279,7 @@ public class SSTableWriter extends SSTable
             try
             {
                 estimatedRows = SSTable.estimateRowsFromData(desc, dfile);
-                iwriter = new IndexWriter(desc, StorageService.getPartitioner(), estimatedRows, true);
+                iwriter = new IndexWriter(desc, StorageService.getPartitioner(), estimatedRows);
             }
             catch(IOException e)
             {
@@ -367,11 +366,11 @@ public class SSTableWriter extends SSTable
         public final BloomFilter bf;
         private FileMark mark;
 
-        IndexWriter(Descriptor desc, IPartitioner part, long keyCount, boolean skipCache) throws IOException
+        IndexWriter(Descriptor desc, IPartitioner part, long keyCount) throws IOException
         {
             this.desc = desc;
             this.partitioner = part;
-            indexFile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_INDEX)), "rw", 8 * 1024 * 1024, skipCache, false);
+            indexFile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_INDEX)), "rw", 8 * 1024 * 1024, true);
             builder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
             summary = new IndexSummary(keyCount);
             bf = BloomFilter.getFilter(keyCount, 15);

src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java

@@ -26,7 +26,6 @@ import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
 
 import org.apache.cassandra.utils.CLibrary;
-import org.apache.log4j.Logger;
 
 /**
  * A <code>BufferedRandomAccessFile</code> is like a
@@ -39,9 +38,9 @@ import org.apache.log4j.Logger;
  * overridden here relies on the implementation of those methods in the
  * superclass.
  */
-public class BufferedRandomAccessFile extends RandomAccessFile implements FileDataInput, PageCacheInformer
+public class BufferedRandomAccessFile extends RandomAccessFile implements FileDataInput
 {
-    private static final Logger logger = Logger.getLogger(BufferedRandomAccessFile.class);
+    private static final long MAX_BYTES_IN_PAGE_CACHE = (long) Math.pow(2, 27); // 128mb
     
     // absolute filesystem path to the file
     private final String filePath;
@@ -73,23 +72,12 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
     // file descriptor
     private int fd;
 
-    // keep *at most* this much data in the page cache from this file at a time
-    private static final long MAX_BYTES_IN_PAGE_CACHE = (long) Math.pow(2, 27); // 128mb
-
     // skip cache - used for commit log and sstable writing w/ posix_fadvise
     private final boolean skipCache;
 
-    // used for page cache migration in compaction
-    private final boolean pageCacheMigrate;
-
-    // tracks the number of bytes read or written since last page cache flush
     private long bytesSinceCacheFlush = 0;
-
-    // tracks the lowest seen file position since the last page cache flush
-    // this is needed because the posix_fadvise call takes a offset and length
     private long minBufferOffset = Long.MAX_VALUE;
 
-
     /*
      * Open a new <code>BufferedRandomAccessFile</code> on the file named
      * <code>name</code> in mode <code>mode</code>, which should be "r" for
@@ -117,15 +105,14 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
 
     public BufferedRandomAccessFile(File file, String mode, int bufferSize) throws IOException
     {
-        this(file, mode, bufferSize, false, false);
+        this(file, mode, bufferSize, false);
     }
 
-    public BufferedRandomAccessFile(File file, String mode, int bufferSize, boolean skipCache, boolean pageCacheMigrate) throws IOException
+    public BufferedRandomAccessFile(File file, String mode, int bufferSize, boolean skipCache) throws IOException
     {
         super(file, mode);
 
         this.skipCache = skipCache;
-        this.pageCacheMigrate = pageCacheMigrate;
 
         channel = super.getChannel();
         filePath = file.getAbsolutePath();
@@ -150,15 +137,10 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
             channel.force(true); // true, because file length counts as
                                  // "meta-data"
 
-            if (skipCache && bytesSinceCacheFlush > 0)
+            if (skipCache)
             {
-                // clear remaining file from page cache
+                // clear entire file from page cache
-                long startAt = 0;
+                CLibrary.trySkipCache(this.fd, 0, 0);
-
-                if (pageCacheMigrate)
-                    startAt = minBufferOffset;
-
-                CLibrary.trySkipCache(this.fd, startAt, 0);
 
                 minBufferOffset = Long.MAX_VALUE;
                 bytesSinceCacheFlush = 0;
@@ -168,34 +150,6 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
         }
     }
 
-    /** {@InheritDoc} */
-    public long getCurrentPosition()
-    {
-        return getFilePointer();
-    }
-
-    /** @{InheritDoc} */
-    public void keepCacheWindow(long startingOffset)
-    {
-
-        if (!pageCacheMigrate)
-            return;
-
-        if (minBufferOffset < startingOffset)
-        {
-            //Flush anything before start offset
-            CLibrary.trySkipCache(this.fd, minBufferOffset, startingOffset - 1);
-
-            //jump ahead to position() so it doesn't get flushed in the range
-            minBufferOffset = bufferOffset;
-            bytesSinceCacheFlush = 0;
-        }
-
-        if(logger.isDebugEnabled())
-            logger.debug("Kept " + (getFilePointer() - startingOffset) + " bytes in page cache");
-    }
-
-
     public void flush() throws IOException
     {
         if (isDirty)
@@ -208,8 +162,10 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
             if (skipCache)
             {
 
-                // we don't know when the data reaches disk since we aren't calling flush
+                // we don't know when the data reaches disk since we aren't
-                // so we continue to clear pages we don't need from the first offset we see
+                // calling flush
+                // so we continue to clear pages we don't need from the first
+                // offset we see
                 // periodically we update this starting offset
                 bytesSinceCacheFlush += validBufferBytes;
 
@@ -218,7 +174,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
 
                 if (bytesSinceCacheFlush >= MAX_BYTES_IN_PAGE_CACHE)
                 {
-                    CLibrary.trySkipCache(this.fd, minBufferOffset, 0);
+                    CLibrary.trySkipCache(this.fd, (int) minBufferOffset, 0);
                     minBufferOffset = bufferOffset;
                     bytesSinceCacheFlush = 0;
                 }
@@ -257,7 +213,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
         bytesSinceCacheFlush += read;
         if (skipCache && bytesSinceCacheFlush >= MAX_BYTES_IN_PAGE_CACHE)
         {
-            CLibrary.trySkipCache(this.fd, minBufferOffset, 0);
+            CLibrary.trySkipCache(this.fd, (int) minBufferOffset, 0);
             bytesSinceCacheFlush = 0;
             minBufferOffset = Long.MAX_VALUE;
         }
@@ -441,12 +397,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
 
         if (skipCache && bytesSinceCacheFlush > 0)
         {
-            long startAt = 0;
+            CLibrary.trySkipCache(this.fd, 0, 0);
-
-            if(pageCacheMigrate)
-                startAt = minBufferOffset;
-
-            CLibrary.trySkipCache(this.fd, startAt, 0);
         }
 
         super.close();
@@ -492,7 +443,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa
 
     public static BufferedRandomAccessFile getUncachingReader(String filename) throws IOException
     {
-        return new BufferedRandomAccessFile(new File(filename), "r", 8 * 1024 * 1024, true, false);
+        return new BufferedRandomAccessFile(new File(filename), "r", 8 * 1024 * 1024, true);
     }
 
     /**

src/java/org/apache/cassandra/io/util/DataOutputBuffer.java

@@ -19,20 +19,14 @@
 package org.apache.cassandra.io.util;
 
 import java.io.DataOutputStream;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.cassandra.utils.Pair;
 
 
 /**
  * An implementation of the DataOutputStream interface. This class is completely thread
  * unsafe.
  */
-public final class DataOutputBuffer extends DataOutputStream implements PageCacheInformer
+public final class DataOutputBuffer extends DataOutputStream
 {
-    private List<Pair<Integer, Integer>> pageCacheMarkers;
-
     public DataOutputBuffer()
     {
         this(128);
@@ -76,32 +70,6 @@ public final class DataOutputBuffer extends DataOutputStream implements PageCach
     {
         this.written = 0;
         buffer().reset();
-        pageCacheMarkers = null;
-
         return this;
     }
-
-    /** {@InheritDoc} */
-    public void keepCacheWindow(long startAt)
-    {
-        if (pageCacheMarkers == null)
-            pageCacheMarkers = new ArrayList<Pair<Integer,Integer>>();
-
-        long endAt = getCurrentPosition();
-
-        assert startAt <= endAt;
-
-        pageCacheMarkers.add(new Pair<Integer,Integer>((int) startAt, (int) (endAt - startAt)));
-    }
-
-    /** {@InheritDoc} */
-    public long getCurrentPosition()
-    {
-        return getLength();
-    }
-
-    public final List<Pair<Integer,Integer>> getPageCacheMarkers()
-    {
-        return pageCacheMarkers;
-    }
 }

src/java/org/apache/cassandra/io/util/PageCacheInformer.java

@@ -1,35 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.io.util;
-
-import java.io.DataOutput;
-
-public interface PageCacheInformer extends DataOutput
-{
-    /**
-     * Instruct the OS to keep a portion of the page cache from startAt to current position.
-     * this call is only appropriate for writes
-     * @param startAt the start of the window
-     */
-    void keepCacheWindow(long startAt);
-
-    /**
-     * @return the current position
-     */
-    long getCurrentPosition();
-}

src/java/org/apache/cassandra/service/StorageService.java

@@ -31,6 +31,7 @@ import javax.management.ObjectName;
 
 import com.google.common.base.Charsets;
 import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Multimap;
 
 import org.apache.cassandra.db.commitlog.CommitLog;
@@ -2195,14 +2196,28 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
         ColumnFamilyStore largestByThroughput = null;
         for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
         {
-            if (largestByOps == null || cfs.getMemtableColumnsCount() > largestByOps.getMemtableColumnsCount())
+            long ops = 0;
+            long throughput = 0;
+            for (ColumnFamilyStore subordinate : Iterables.concat(Collections.singleton(cfs), cfs.getIndexColumnFamilyStores()))
+            {
+                ops += subordinate.getMemtableColumnsCount();
+                throughput = subordinate.getMemtableThroughputInMB();
+            }
+
+            if (ops > 0 && (largestByOps == null || ops > largestByOps.getMemtableColumnsCount()))
+            {
+                logger_.debug(ops + " total ops in " + cfs);
                 largestByOps = cfs;
-            if (largestByThroughput == null || cfs.getMemtableThroughputInMB() > largestByThroughput.getMemtableThroughputInMB())
+            }
+            if (throughput > 0 && (largestByThroughput == null || throughput > largestByThroughput.getMemtableThroughputInMB()))
+            {
+                logger_.debug(throughput + " total throughput in " + cfs);
                 largestByThroughput = cfs;
+            }
         }
         if (largestByOps == null)
         {
-            logger_.error("Unable to reduce heap usage since there are no column families defined");
+            logger_.info("Unable to reduce heap usage since there are no dirty column families");
             return;
         }
 

src/java/org/apache/cassandra/utils/CLibrary.java

@@ -18,14 +18,10 @@
  */
 package org.apache.cassandra.utils;
 
-import java.io.*;
+import java.io.File;
+import java.io.FileDescriptor;
+import java.io.IOException;
 import java.lang.reflect.Field;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel.MapMode;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -33,10 +29,6 @@ import org.slf4j.LoggerFactory;
 import com.sun.jna.LastErrorException;
 import com.sun.jna.Native;
 
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.io.util.BufferedRandomAccessFile;
-import org.apache.cassandra.io.util.FileUtils;
-
 public final class CLibrary
 {
     private static Logger logger = LoggerFactory.getLogger(CLibrary.class);
@@ -58,28 +50,11 @@ public final class CLibrary
     private static final int POSIX_FADV_DONTNEED   = 4; /* fadvise.h */
     private static final int POSIX_FADV_NOREUSE    = 5; /* fadvise.h */
 
-    private static native int mlockall(int flags) throws LastErrorException;
-    private static native int munlockall() throws LastErrorException;
-
-    private static native int link(String from, String to) throws LastErrorException;
-
-    // fcntl - manipulate file descriptor, `man 2 fcntl`
-    public static native int fcntl(int fd, int command, long flags) throws LastErrorException;
-
-    // fadvice
-    public static native int posix_fadvise(int fd, long offset, long len, int flag) throws LastErrorException;
-
-    public static native int mincore(ByteBuffer buf, int length, char[] vec) throws LastErrorException;
-
-    private static native int getpagesize() throws LastErrorException;
-    public  static Integer pageSize = null;
-
     static
     {
         try
         {
             Native.register("c");
-            pageSize = CLibrary.getPageSize();
         }
         catch (NoClassDefFoundError e)
         {
@@ -93,38 +68,19 @@ public final class CLibrary
         {
             logger.warn("Obsolete version of JNA present; unable to register C library. Upgrade to JNA 3.2.7 or later");
         }
-
     }
 
-    private static Integer getPageSize()
+    private static native int mlockall(int flags) throws LastErrorException;
-    {
+    private static native int munlockall() throws LastErrorException;
-        int ps = -1;
 
-        if (!DatabaseDescriptor.isPageCaheMigrationEnabled())
+    private static native int link(String from, String to) throws LastErrorException;
-            return null;
 
-        try
+    // fcntl - manipulate file descriptor, `man 2 fcntl`
-        {
+    public static native int fcntl(int fd, int command, long flags) throws LastErrorException;
-            ps = CLibrary.getpagesize();
-            assert ps >= 0; // on error a value of -1 is returned and errno is set to indicate the error.
-
-            logger.info("PageSize = " + ps);
-        }
-        catch (RuntimeException e)
-        {
-            if (!(e instanceof LastErrorException))
-                throw e;
-
-            logger.warn(String.format("getpagesize failed, errno (%d).", CLibrary.errno(e)));
-        }
-        catch (UnsatisfiedLinkError e)
-        {
-            // this will have already been logged by CLibrary, no need to repeat it
-        }
-
-        return ps > 0 ? ps : null;
-    }
 
+    // fadvice
+    public static native int posix_fadvise(int fd, int offset, int len, int flag) throws LastErrorException;
+        
     private static int errno(RuntimeException e)
     {
         assert e instanceof LastErrorException;
@@ -233,9 +189,9 @@ public final class CLibrary
         }
     }
 
-    public static void trySkipCache(int fd, long offset, long len)
+    public static void trySkipCache(int fd, int offset, int len)
     {
-        if (fd < 0 || !DatabaseDescriptor.isPageCaheMigrationEnabled())
+        if (fd < 0)
             return;
 
         try
@@ -256,63 +212,6 @@ public final class CLibrary
         }
     }
 
-    public static PageCacheMetrics getCachedPages(BufferedRandomAccessFile braf) throws IOException
-    {
-        if (pageSize == null || pageSize == 0 || !DatabaseDescriptor.isPageCaheMigrationEnabled())
-            return null;
-
-        long length = braf.length();
-
-        PageCacheMetrics pageCacheMetrics = new PageCacheMetrics(pageSize, length);
-
-        try
-        {
-            char[] pages = null;
-
-            // MMap 2G chunks
-            for (long offset = 0; offset < length; offset += Integer.MAX_VALUE)
-            {
-                long limit = (offset + Integer.MAX_VALUE) > length ? (length - offset) : Integer.MAX_VALUE;
-
-                ByteBuffer buf = braf.getChannel().map(MapMode.READ_ONLY, offset, limit);
-
-                int numPages = (int) ((limit + pageSize - 1) / pageSize);
-
-                if (pages == null || pages.length < numPages)
-                    pages = new char[numPages];
-
-                int rc = mincore(buf, (int) limit, pages);
-
-                if (rc != 0)
-                {
-                    logger.warn(String.format("mincore failed, rc (%d).", rc));
-                    break;
-                }
-
-                for (long i = 0, position = offset; position < limit; position += pageSize, i++)
-                {
-                    if ((pages[(int) i] & 1) == 1)
-                    {
-                        pageCacheMetrics.setPage(position);
-                    }
-                }
-            }
-        }
-        catch (RuntimeException e)
-        {
-            if (!(e instanceof LastErrorException))
-                throw e;
-
-            logger.warn(String.format("mincore failed, errno (%d).", CLibrary.errno(e)));
-        }
-        catch (UnsatisfiedLinkError e)
-        {
-            // this will have already been logged by CLibrary, no need to repeat it
-        }
-
-        return pageCacheMetrics;
-    }
-
     public static int tryFcntl(int fd, int command, int flags)
     {
         int result = -1;

src/java/org/apache/cassandra/utils/PageCacheMetrics.java

@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.cassandra.utils;
-
-import org.apache.cassandra.utils.obs.OpenBitSet;
-
-public class PageCacheMetrics
-{
-    private final OpenBitSet pageMap;
-    private final long fileLength;
-    public final int pageSize;
-
-    public PageCacheMetrics(int pageSize, long fileLength)
-    {
-        assert pageSize > 0;
-        assert fileLength > 0;
-
-        this.pageSize = pageSize;
-        this.fileLength = fileLength;
-
-        long numPages = (fileLength + pageSize - 1) / pageSize;
-        pageMap = new OpenBitSet(numPages);
-
-        assert isEmpty();
-    }
-
-    public boolean isEmpty()
-    {
-        return pageMap.cardinality() == 0;
-    }
-
-    public void setPage(long position)
-    {
-        pageMap.fastSet(getPageNumber(position));
-    }
-
-    private long getPageNumber(long position)
-    {
-        assert position <= fileLength;
-
-        return (position + pageSize - 1) / pageSize;
-    }
-
-    /**
-     * Returns true if any page within this range is cached
-     * @param start starting position
-     * @param end the limit of the range
-     * @return true if given range is in the cache, false otherwise
-     */
-    public boolean isRangeInCache(long start, long end)
-    {
-        long startPage = getPageNumber(start);
-        long endPage = getPageNumber(end);
-
-        assert startPage <= endPage;
-
-        for (long i = startPage; i < endPage; i++)
-        {
-            if (pageMap.fastGet(i))
-                return true;
-        }
-
-        return false;
-    }
-}

test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java

@@ -138,11 +138,11 @@ public class AntiEntropyServiceTest extends CleanupHelper
 
         // add a row with the minimum token
         validator.add(new PrecompactedRow(new DecoratedKey(min, ByteBufferUtil.bytes("nonsense!")),
-                                          new DataOutputBuffer()));
+                                       new DataOutputBuffer()));
 
         // and a row after it
         validator.add(new PrecompactedRow(new DecoratedKey(mid, ByteBufferUtil.bytes("inconceivable!")),
-                                          new DataOutputBuffer()));
+                                       new DataOutputBuffer()));
         validator.complete();
 
         // confirm that the tree was validated