diff --git a/CHANGES.txt b/CHANGES.txt index b68fbe75a3..8db89944d4 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -14,17 +14,19 @@ 0.7.4 * add nodetool join command (CASSANDRA-2160) * fix secondary indexes on pre-existing or streamed data (CASSANDRA-2244) - * initialize endpoing in gossiper earlier (CASSANDRA-2228) + * initialize endpoint in gossiper earlier (CASSANDRA-2228) * add ability to write to Cassandra from Pig (CASSANDRA-1828) * add rpc_[min|max]_threads (CASSANDRA-2176) * 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 (CASSANDRA-2259) + * avoid writing empty rows when scrubbing tombstoned rows (CASSANDRA-2296) + * fix assertion error in range and index scans for CL < ALL + (CASSANDRA-2282) 0.7.3 diff --git a/NEWS.txt b/NEWS.txt index c9a4ea3352..7f3240e32c 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -1,4 +1,3 @@ -<<<<<<< .working Whatever ======== @@ -25,7 +24,19 @@ JMX - By default, JMX now listens on port 7199. -======= +0.7.4 +===== + +Upgrading +--------- + - Nothing specific to 0.7.4, but see 0.7.3 Upgrading if upgrading + from earlier than 0.7.1. + +Features +-------- + - Output to Pig is now supported as well as input + + 0.7.3 ===== diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index db31bb29e8..416eba5674 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -256,10 +256,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 diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index b8e7fbd7c9..c0f3cbbf5c 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -71,7 +71,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo } /** - * @param limit: number of rows to fetch at a time + * @param limit: number of columns to fetch in a slice */ public CassandraStorage(int limit) { diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index b9a123dfad..d979a2852d 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -114,10 +114,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 { diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index f12f5cf278..0f5f216eb9 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1236,11 +1236,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; diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java index 976745f6a5..c406f6f1b1 100644 --- a/src/java/org/apache/cassandra/db/Column.java +++ b/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() @@ -239,15 +237,5 @@ public class Column implements IColumn { return !isMarkedForDelete(); } - - public boolean isInPageCache() - { - return isInPageCache; - } - - public void setIsInPageCache(boolean isInPageCache) - { - this.isInPageCache = isInPageCache; - } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index ff6fcbf74b..4137c0def3 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -224,13 +224,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); diff --git a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java b/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java index c85588f2a6..543aed490e 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java +++ b/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,12 +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; - -import org.apache.cassandra.io.util.PageCacheInformer; -import org.apache.cassandra.utils.PageCacheMetrics; - public class ColumnFamilySerializer implements ICompactSerializer2 { @@ -74,16 +70,11 @@ public class ColumnFamilySerializer implements ICompactSerializer2 { 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); @@ -93,16 +84,8 @@ public class ColumnFamilySerializer implements ICompactSerializer2 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) @@ -139,50 +122,19 @@ public class ColumnFamilySerializer implements ICompactSerializer2 throw new UnserializableColumnFamilyException("Couldn't find cfId=" + cfId, cfId); ColumnFamily cf = ColumnFamily.create(cfId); deserializeFromSSTableNoColumns(cf, dis); - deserializeColumns(dis, cf, intern, null); - + deserializeColumns(dis, cf, intern); return cf; } - - public boolean deserializeColumns(DataInput dis, ColumnFamily cf, boolean intern, PageCacheMetrics pageCacheMetrics) throws IOException + public void deserializeColumns(DataInput dis, ColumnFamily cf, boolean intern) throws IOException { - int size = dis.readInt(); - ColumnFamilyStore interner = intern ? Table.open(CFMetaData.getCF(cf.id()).left).getColumnFamilyStore(cf.id()) : null; - boolean hasColumnsInPageCache = false; - - if (pageCacheMetrics != null && dis instanceof RandomAccessFile) - { - RandomAccessFile raf = (RandomAccessFile) dis; - - for (int i = 0; i < size; ++i) - { - long startAt = raf.getFilePointer(); - - IColumn column = cf.getColumnSerializer().deserialize(dis, interner); - - long endAt = raf.getFilePointer(); - - column.setIsInPageCache(pageCacheMetrics.isRangeInCache(startAt, endAt)); - - if(!hasColumnsInPageCache) - hasColumnsInPageCache = column.isInPageCache(); - - cf.addColumn(column); - } + for (int i = 0; i < size; ++i) + { + IColumn column = cf.getColumnSerializer().deserialize(dis, interner); + cf.addColumn(column); } - else - { - for (int i = 0; i < size; ++i) - { - IColumn column = cf.getColumnSerializer().deserialize(dis, interner); - cf.addColumn(column); - } - } - - return hasColumnsInPageCache; } public ColumnFamily deserializeFromSSTableNoColumns(ColumnFamily cf, DataInput input) throws IOException diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 60f5484321..bf3bc7166e 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -2240,11 +2240,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); } } diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 33fad4b83e..89f8118486 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/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.streaming.OperationType; @@ -391,7 +390,6 @@ public class CompactionManager implements CompactionManagerMBean assert sstables != null; Table table = cfs.table; - if (DatabaseDescriptor.isSnapshotBeforeCompaction()) table.snapshot("compact-" + cfs.columnFamily); @@ -401,11 +399,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 smallerSSTables = new ArrayList(sstables); - while (compactionFileLocation == null && smallerSSTables.size() > 1) { logger.warn("insufficient space to compact all requested files " + StringUtils.join(smallerSSTables, ", ")); @@ -417,7 +413,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, @@ -429,9 +424,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() @@ -451,13 +446,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()) @@ -487,7 +480,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(); @@ -540,9 +533,9 @@ 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; + int goodRows = 0, badRows = 0, emptyRows = 0; while (!dataFile.isEOF()) { @@ -595,9 +588,17 @@ 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); - writer.append(getCompactedRow(row, cfs, sstable.descriptor, true)); - goodRows++; + SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStart, dataSize, true); + AbstractCompactedRow compactedRow = getCompactedRow(row, cfs, sstable.descriptor, true); + if (compactedRow.isEmpty()) + { + emptyRows++; + } + else + { + writer.append(compactedRow); + goodRows++; + } if (!key.key.equals(currentIndexKey) || dataStart != dataStartFromIndex) logger.warn("Row scrubbed successfully but index file contains a different key or row size; consider rebuilding the index as described in http://www.mail-archive.com/user@cassandra.apache.org/msg03325.html"); } @@ -615,9 +616,17 @@ 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); - writer.append(getCompactedRow(row, cfs, sstable.descriptor, true)); - goodRows++; + SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStartFromIndex, dataSizeFromIndex, true); + AbstractCompactedRow compactedRow = getCompactedRow(row, cfs, sstable.descriptor, true); + if (compactedRow.isEmpty()) + { + emptyRows++; + } + else + { + writer.append(compactedRow); + goodRows++; + } } catch (Throwable th2) { @@ -630,7 +639,7 @@ public class CompactionManager implements CompactionManagerMBean } else { - logger.warn("Row is unreadable; skipping to next"); + logger.warn("Row at " + dataStart + " is unreadable; skipping to next"); if (currentIndexKey != null) dataFile.seek(nextRowPositionFromIndex); badRows++; @@ -642,14 +651,17 @@ public class CompactionManager implements CompactionManagerMBean { SSTableReader newSstable = writer.closeAndOpenReader(sstable.maxDataAge); cfs.replaceCompactedSSTables(Arrays.asList(sstable), Arrays.asList(newSstable)); - logger.info("Scrub of " + sstable + " complete: " + goodRows + " rows in new sstable"); + logger.info("Scrub of " + sstable + " complete: " + goodRows + " rows in new sstable and " + emptyRows + " empty (tombstoned) rows dropped"); if (badRows > 0) logger.warn("Unable to recover " + badRows + " rows that were skipped. You can attempt manual recovery from the pre-scrub snapshot. You can also run nodetool repair to transfer the data from a healthy replica, if any"); } else { cfs.markCompacted(Arrays.asList(sstable)); - logger.warn("No valid rows found while scrubbing " + sstable + "; it is marked for deletion now. If you want to attempt manual recovery, you can find a copy in the pre-scrub snapshot"); + if (badRows > 0) + logger.warn("No valid rows found while scrubbing " + sstable + "; it is marked for deletion now. If you want to attempt manual recovery, you can find a copy in the pre-scrub snapshot"); + else + logger.info("Scrub of " + sstable + " complete; looks like all " + emptyRows + " rows were tombstoned"); } } } @@ -700,7 +712,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++; } @@ -769,13 +781,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; } @@ -1121,8 +1133,9 @@ 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); row.echoData(out); } diff --git a/src/java/org/apache/cassandra/db/IColumn.java b/src/java/org/apache/cassandra/db/IColumn.java index f287c065cd..7e6d84faf2 100644 --- a/src/java/org/apache/cassandra/db/IColumn.java +++ b/src/java/org/apache/cassandra/db/IColumn.java @@ -57,12 +57,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); } diff --git a/src/java/org/apache/cassandra/db/SuperColumn.java b/src/java/org/apache/cassandra/db/SuperColumn.java index bdc50f45c1..c2389187f1 100644 --- a/src/java/org/apache/cassandra/db/SuperColumn.java +++ b/src/java/org/apache/cassandra/db/SuperColumn.java @@ -48,7 +48,6 @@ public class SuperColumn implements IColumn, IColumnContainer private ConcurrentSkipListMap 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) { @@ -322,16 +321,6 @@ public class SuperColumn implements IColumn, IColumnContainer { throw new UnsupportedOperationException("Super columns don't have a serialization mask"); } - - public boolean isInPageCache() - { - return isInPageCache; - } - - public void setIsInPageCache(boolean isInPageCache) - { - this.isInPageCache = isInPageCache; - } } class SuperColumnSerializer implements IColumnSerializer diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 10dd29c2ae..3c8ff2f8c4 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -174,7 +174,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 { diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java index bdb8db2790..2ae230b260 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java +++ b/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) throws IOException diff --git a/src/java/org/apache/cassandra/io/AbstractCompactedRow.java b/src/java/org/apache/cassandra/io/AbstractCompactedRow.java index 77ff784697..eaaa157219 100644 --- a/src/java/org/apache/cassandra/io/AbstractCompactedRow.java +++ b/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; - } - } diff --git a/src/java/org/apache/cassandra/io/CompactionIterator.java b/src/java/org/apache/cassandra/io/CompactionIterator.java index 6ebc5b0b3c..88271e5515 100644 --- a/src/java/org/apache/cassandra/io/CompactionIterator.java +++ b/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 rows = new ArrayList(); - 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 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"; diff --git a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java b/src/java/org/apache/cassandra/io/LazilyCompactedRow.java index 8cab293b34..d11e971303 100644 --- a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java +++ b/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,11 +94,12 @@ 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) { SSTableIdentityIterator row = rows.get(0); + assert row.dataSize > 0; out.writeLong(row.dataSize); row.echoData(out); return; @@ -106,7 +108,9 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl DataOutputBuffer clockOut = new DataOutputBuffer(); ColumnFamily.serializer().serializeCFInfo(emptyColumnFamily, clockOut); - out.writeLong(headerBuffer.getLength() + clockOut.getLength() + columnSerializedSize); + long dataSize = headerBuffer.getLength() + clockOut.getLength() + columnSerializedSize; + assert dataSize > 0; + out.writeLong(dataSize); out.write(headerBuffer.getData(), 0, headerBuffer.getLength()); out.write(clockOut.getData(), 0, clockOut.getLength()); out.writeInt(columnCount); diff --git a/src/java/org/apache/cassandra/io/PrecompactedRow.java b/src/java/org/apache/cassandra/io/PrecompactedRow.java index 5cfcf9ddae..05e3abbf8b 100644 --- a/src/java/org/apache/cassandra/io/PrecompactedRow.java +++ b/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; @@ -38,8 +39,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. @@ -95,11 +94,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; @@ -122,47 +117,13 @@ public class PrecompactedRow extends AbstractCompactedRow } } - public void write(PageCacheInformer out) throws IOException + public void write(DataOutput out) throws IOException { - out.writeLong(headerBuffer.getLength() + buffer.getLength()); + long dataSize = headerBuffer.getLength() + buffer.getLength(); + assert dataSize > 0; + out.writeLong(dataSize); out.write(headerBuffer.getData(), 0, headerBuffer.getLength()); - - List> 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 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); - } + out.write(buffer.getData(), 0, buffer.getLength()); } public void update(MessageDigest digest) diff --git a/src/java/org/apache/cassandra/io/sstable/CacheWriter.java b/src/java/org/apache/cassandra/io/sstable/CacheWriter.java index 8342d76bb9..b3ab1ce622 100644 --- a/src/java/org/apache/cassandra/io/sstable/CacheWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CacheWriter.java @@ -75,7 +75,7 @@ public class CacheWriter 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) diff --git a/src/java/org/apache/cassandra/io/sstable/IndexHelper.java b/src/java/org/apache/cassandra/io/sstable/IndexHelper.java index 2c867da1d9..59cc68dad2 100644 --- a/src/java/org/apache/cassandra/io/sstable/IndexHelper.java +++ b/src/java/org/apache/cassandra/io/sstable/IndexHelper.java @@ -110,7 +110,7 @@ public class IndexHelper { int size = file.readInt(); if (size > maxSize || size <= 0) - throw new EOFException("bloom filter claims to be longer than entire row size"); + throw new EOFException("bloom filter claims to be " + size + " bytes, longer than entire row size " + maxSize); ByteBuffer bytes = file.readBytes(size); DataInputStream stream = new DataInputStream(ByteBufferUtil.inputStream(bytes)); diff --git a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java index 7bd9eb8b5f..4f3ca43ab0 100644 --- a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java @@ -47,8 +47,7 @@ public class KeyIterator extends AbstractIterator implements Itera in = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_INDEX)), "r", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, - true, - false); + true); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/io/sstable/SSTable.java b/src/java/org/apache/cassandra/io/sstable/SSTable.java index 05d6212f0a..6b4d517622 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTable.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java @@ -174,7 +174,7 @@ public abstract class SSTable } catch (Exception e) { - if (!"snapshots".equals(name)) + if (!"snapshots".equals(name) && !"backups".equals(name)) logger.warn("Invalid file '{}' in data directory {}.", name, dir); return null; } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java index 2e05f19de9..c703053869 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java +++ b/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.PageCacheMetrics; +import org.apache.cassandra.utils.Filter; public class SSTableIdentityIterator implements Comparable, IColumnIterator { @@ -48,9 +47,6 @@ public class SSTableIdentityIterator implements Comparable= 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)); + out.write(file.readByte()); } } @@ -237,9 +173,7 @@ public class SSTableIdentityIterator implements Comparable BufferedRandomAccessFile input = new BufferedRandomAccessFile(new File(descriptor.filenameFor(Component.PRIMARY_INDEX)), "r", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, - true, - false); - + true); try { if (keyCache != null && keyCache.getCapacity() - keyCache.getSize() < keysToLoadInCache.size()) @@ -297,11 +295,9 @@ public class SSTableReader extends SSTable implements Comparable break; boolean shouldAddEntry = indexSummary.shouldAddEntry(); - ByteBuffer key = (shouldAddEntry || cacheLoading || recreatebloom) - ? ByteBufferUtil.readWithShortLength(input) - : ByteBufferUtil.skipShortLength(input); - + ? ByteBufferUtil.readWithShortLength(input) + : ByteBufferUtil.skipShortLength(input); long dataPosition = input.readLong(); if (key != null) { @@ -553,7 +549,7 @@ public class SSTableReader extends SSTable implements Comparable } /** - * 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. */ diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java index 3061728812..fa7075ba83 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java +++ b/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, Closeable @@ -41,9 +42,6 @@ public class SSTableScanner implements Iterator, Closeable private final BufferedRandomAccessFile file; private final SSTableReader sstable; - - private final PageCacheMetrics pageCacheMetrics; - private IColumnIterator row; private boolean exhausted = false; private Iterator iterator; @@ -54,25 +52,15 @@ public class SSTableScanner implements Iterator, Closeable */ SSTableScanner(SSTableReader sstable, int bufferSize, boolean skipCache) { - PageCacheMetrics pageCacheMetrics = null; - try { - File sstableFile = new File(sstable.getFilename()); - file = new BufferedRandomAccessFile(sstableFile, "r", bufferSize, skipCache, false); - - if (skipCache) - { - pageCacheMetrics = CLibrary.getCachedPages(file); - } + this.file = new BufferedRandomAccessFile(new File(sstable.getFilename()), "r", bufferSize, skipCache); } catch (IOException e) { throw new IOError(e); } - this.sstable = sstable; - this.pageCacheMetrics = pageCacheMetrics; } /** @@ -91,7 +79,6 @@ public class SSTableScanner implements Iterator, Closeable } this.sstable = sstable; this.filter = filter; - this.pageCacheMetrics = null; } public void close() throws IOException @@ -189,7 +176,8 @@ public class SSTableScanner implements Iterator, 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 { diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java index 62bfa4a2bb..dcfb07d3a9 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java @@ -58,16 +58,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(Arrays.asList(Component.DATA, Component.FILTER, Component.PRIMARY_INDEX, Component.STATS)), @@ -75,11 +75,9 @@ public class SSTableWriter extends SSTable partitioner, SSTable.defaultRowHistogram(), SSTable.defaultColumnHistogram()); - - iwriter = new IndexWriter(descriptor, partitioner, keyCount, !migratePageCache); //when we migrate pages we cache the index + iwriter = new IndexWriter(descriptor, partitioner, keyCount); 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() @@ -150,7 +148,9 @@ public class SSTableWriter extends SSTable // seek back and write the row size (not including the size Long itself) long endPosition = dataFile.getFilePointer(); dataFile.seek(sizePosition); - dataFile.writeLong(endPosition - (sizePosition + 8)); + long dataSize = endPosition - (sizePosition + 8); + assert dataSize > 0; + dataFile.writeLong(dataSize); // finally, reset for next row dataFile.seek(endPosition); afterAppend(decoratedKey, startPosition); @@ -317,7 +317,7 @@ public class SSTableWriter extends SSTable RowIndexer(Descriptor desc, CFMetaData metadata) throws IOException { - this(desc, new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "r", 8 * 1024 * 1024, true, false), metadata); + this(desc, new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "r", 8 * 1024 * 1024, true), metadata); } protected RowIndexer(Descriptor desc, BufferedRandomAccessFile dfile, CFMetaData metadata) throws IOException @@ -333,7 +333,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); return estimatedRows; } catch(IOException e) @@ -399,7 +399,7 @@ public class SSTableWriter extends SSTable { AESCommutativeRowIndexer(Descriptor desc, CFMetaData metadata) throws IOException { - super(desc, new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "rw", 8 * 1024 * 1024, true, false), metadata); + super(desc, new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "rw", 8 * 1024 * 1024, true), metadata); } @Override @@ -427,7 +427,7 @@ public class SSTableWriter extends SSTable // deserialize CF ColumnFamily cf = ColumnFamily.create(desc.ksname, desc.cfname); ColumnFamily.serializer().deserializeFromSSTableNoColumns(cf, dfile); - ColumnFamily.serializer().deserializeColumns(dfile, cf, false, null); + ColumnFamily.serializer().deserializeColumns(dfile, cf, false); rowSizes.add(dataSize); columnCounts.add(cf.getEstimatedColumnCount()); @@ -486,11 +486,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); diff --git a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java b/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java index 4157c59733..372d76ea6b 100644 --- a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java +++ b/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 BufferedRandomAccessFile 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 BufferedRandomAccessFile on the file named * name in mode mode, 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 - long startAt = 0; - - if (pageCacheMigrate) - startAt = minBufferOffset; - - CLibrary.trySkipCache(this.fd, startAt, 0); + // clear entire file from page cache + CLibrary.trySkipCache(this.fd, 0, 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 - // so we continue to clear pages we don't need from the first offset we see + // we don't know when the data reaches disk since we aren't + // 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; - - if(pageCacheMigrate) - startAt = minBufferOffset; - - CLibrary.trySkipCache(this.fd, startAt, 0); + CLibrary.trySkipCache(this.fd, 0, 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); } /** diff --git a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java b/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java index e1ec7a4284..ac8a0ed05e 100644 --- a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java +++ b/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> 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>(); - - long endAt = getCurrentPosition(); - - assert startAt <= endAt; - - pageCacheMarkers.add(new Pair((int) startAt, (int) (endAt - startAt))); - } - - /** {@InheritDoc} */ - public long getCurrentPosition() - { - return getLength(); - } - - public final List> getPageCacheMarkers() - { - return pageCacheMarkers; - } } diff --git a/src/java/org/apache/cassandra/io/util/PageCacheInformer.java b/src/java/org/apache/cassandra/io/util/PageCacheInformer.java deleted file mode 100644 index 46e57c11e6..0000000000 --- a/src/java/org/apache/cassandra/io/util/PageCacheInformer.java +++ /dev/null @@ -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(); -} diff --git a/src/java/org/apache/cassandra/utils/CLibrary.java b/src/java/org/apache/cassandra/utils/CLibrary.java index aefee7651c..3b0c7af86e 100644 --- a/src/java/org/apache/cassandra/utils/CLibrary.java +++ b/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() - { - int ps = -1; + private static native int mlockall(int flags) throws LastErrorException; + private static native int munlockall() throws LastErrorException; - if (!DatabaseDescriptor.isPageCaheMigrationEnabled()) - return null; + private static native int link(String from, String to) throws LastErrorException; - try - { - 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; - } + // 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, 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; diff --git a/src/java/org/apache/cassandra/utils/PageCacheMetrics.java b/src/java/org/apache/cassandra/utils/PageCacheMetrics.java deleted file mode 100644 index a1a73d8f36..0000000000 --- a/src/java/org/apache/cassandra/utils/PageCacheMetrics.java +++ /dev/null @@ -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; - } -} diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java index 534a3a7ba6..a0ce0f1693 100644 --- a/test/unit/org/apache/cassandra/db/ScrubTest.java +++ b/test/unit/org/apache/cassandra/db/ScrubTest.java @@ -46,6 +46,7 @@ public class ScrubTest extends CleanupHelper public String TABLE = "Keyspace1"; public String CF = "Standard1"; public String CF2 = "Super5"; + public String CF3 = "Standard2"; public String corruptSSTableName; @@ -128,6 +129,25 @@ public class ScrubTest extends CleanupHelper assertEquals(1, rows.size()); } + @Test + public void testScrubDeletedRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException + { + CompactionManager.instance.disableAutoCompaction(); + Table table = Table.open(TABLE); + ColumnFamilyStore cfs = table.getColumnFamilyStore(CF3); + + RowMutation rm; + rm = new RowMutation(TABLE, ByteBufferUtil.bytes(1)); + ColumnFamily cf = ColumnFamily.create(TABLE, CF3); + cf.delete(0, 1); // expired tombstone + rm.add(cf); + rm.applyUnsafe(); + cfs.forceBlockingFlush(); + + CompactionManager.instance.performScrub(cfs); + assert cfs.getSSTables().isEmpty(); + } + @Test public void testScrubMultiRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException { diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java index bdafef1459..7402233469 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java +++ b/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