Author: kturner
Date: Tue Apr 24 15:33:06 2012
New Revision: 1329792
URL: http://svn.apache.org/viewvc?rev=1329792&view=rev
Log:
ACCUMULO-550 Added code to buffer rfile index entries so that multiple index
blocks are written contiguously (merged from trunk)
Modified:
accumulo/branches/1.4/ (props changed)
accumulo/branches/1.4/src/ (props changed)
accumulo/branches/1.4/src/core/ (props changed)
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/MultiLevelIndex.java
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java
accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/file/rfile/MultiLevelIndexTest.java
accumulo/branches/1.4/src/server/ (props changed)
Propchange: accumulo/branches/1.4/
------------------------------------------------------------------------------
Merged /accumulo/trunk:r1329425
Propchange: accumulo/branches/1.4/src/
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Tue Apr 24 15:33:06 2012
@@ -0,0 +1,7 @@
+/accumulo/branches/1.3/src:1309369,1328076
+/accumulo/trunk:1329425
+/accumulo/trunk/src:1329425
+/incubator/accumulo/branches/1.3/src:1190280,1190413,1190420,1190427,1190500,1195622,1195625,1195629,1195635,1196044,1196054,1196057,1196071-1196072,1196106,1197066,1198935,1199383,1203683,1204625,1205547,1205880,1206169,1208031,1209124,1209526,1209532,1209539,1209541,1209587,1209657,1210518,1210571,1210596,1210598,1213424,1214320,1225006,1227215,1227231,1227611,1228195,1230180,1230736,1231043,1236873,1245632
+/incubator/accumulo/branches/1.3.5rc/src:1209938
+/incubator/accumulo/branches/1.4.0rc/src:1304025,1305326
+/incubator/accumulo/trunk/src:1205476,1205570,1208726,1222413,1222719,1222725,1222733-1222734,1296160-1296495
Propchange: accumulo/branches/1.4/src/core/
------------------------------------------------------------------------------
Merged /accumulo/trunk/src/core:r1329425
Merged /accumulo/trunk/core:r1329425
Modified:
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/MultiLevelIndex.java
URL:
http://svn.apache.org/viewvc/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/MultiLevelIndex.java?rev=1329792&r1=1329791&r2=1329792&view=diff
==============================================================================
---
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/MultiLevelIndex.java
(original)
+++
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/MultiLevelIndex.java
Tue Apr 24 15:33:06 2012
@@ -346,6 +346,60 @@ public class MultiLevelIndex {
}
+ /**
+ * this class buffers writes to the index so that chunks of index blocks are
contiguous in the file instead of having index blocks sprinkled throughout the
+ * file making scans of the entire index slow.
+ */
+ public static class BufferedWriter {
+
+ private Writer writer;
+ private DataOutputStream buffer;
+ private int buffered;
+ private ByteArrayOutputStream baos;
+
+ public BufferedWriter(Writer writer) {
+ this.writer = writer;
+ baos = new ByteArrayOutputStream(1 << 20);
+ buffer = new DataOutputStream(baos);
+ buffered = 0;
+ }
+
+ private void flush() throws IOException {
+ buffer.close();
+
+ DataInputStream dis = new DataInputStream(new
ByteArrayInputStream(baos.toByteArray()));
+
+ IndexEntry ie = new IndexEntry(true);
+ for (int i = 0; i < buffered; i++) {
+ ie.readFields(dis);
+ writer.add(ie.getKey(), ie.getNumEntries(), ie.getOffset(),
ie.getCompressedSize(), ie.getRawSize());
+ }
+
+ buffered = 0;
+ baos = new ByteArrayOutputStream(1 << 20);
+ buffer = new DataOutputStream(baos);
+
+ }
+
+ public void add(Key key, int data, long offset, long compressedSize, long
rawSize) throws IOException {
+ if (buffer.size() > (10 * 1 << 20)) {
+ flush();
+ }
+
+ new IndexEntry(key, data, offset, compressedSize, rawSize).write(buffer);
+ buffered++;
+ }
+
+ public void addLast(Key key, int data, long offset, long compressedSize,
long rawSize) throws IOException {
+ flush();
+ writer.addLast(key, data, offset, compressedSize, rawSize);
+ }
+
+ public void close(DataOutput out) throws IOException {
+ writer.close(out);
+ }
+ }
+
public static class Writer {
private int threshold;
Modified:
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java
URL:
http://svn.apache.org/viewvc/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java?rev=1329792&r1=1329791&r2=1329792&view=diff
==============================================================================
---
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java
(original)
+++
accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java
Tue Apr 24 15:33:06 2012
@@ -103,7 +103,7 @@ public class RFile {
private String name;
private Set<ByteSequence> previousColumnFamilies;
- private MultiLevelIndex.Writer indexWriter;
+ private MultiLevelIndex.BufferedWriter indexWriter;
private MultiLevelIndex.Reader indexReader;
public LocalityGroupMetadata(int version, BlockFileReader br) {
@@ -117,7 +117,7 @@ public class RFile {
columnFamilies = new HashMap<ByteSequence,Count>();
previousColumnFamilies = pcf;
- indexWriter = new MultiLevelIndex.Writer(bfw, indexBlockSize);
+ indexWriter = new MultiLevelIndex.BufferedWriter(new
MultiLevelIndex.Writer(bfw, indexBlockSize));
}
public LocalityGroupMetadata(String name, Set<ByteSequence> cfset, int
nextBlock, int indexBlockSize, BlockFileWriter bfw) {
@@ -129,7 +129,7 @@ public class RFile {
columnFamilies.put(cf, new Count(0));
}
- indexWriter = new MultiLevelIndex.Writer(bfw, indexBlockSize);
+ indexWriter = new MultiLevelIndex.BufferedWriter(new
MultiLevelIndex.Writer(bfw, indexBlockSize));
}
private Key getFirstKey() {
Modified:
accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/file/rfile/MultiLevelIndexTest.java
URL:
http://svn.apache.org/viewvc/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/file/rfile/MultiLevelIndexTest.java?rev=1329792&r1=1329791&r2=1329792&view=diff
==============================================================================
---
accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/file/rfile/MultiLevelIndexTest.java
(original)
+++
accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/file/rfile/MultiLevelIndexTest.java
Tue Apr 24 15:33:06 2012
@@ -26,11 +26,11 @@ import org.apache.accumulo.core.data.Key
import org.apache.accumulo.core.file.blockfile.ABlockWriter;
import org.apache.accumulo.core.file.blockfile.impl.CachableBlockFile;
import
org.apache.accumulo.core.file.blockfile.impl.CachableBlockFile.BlockRead;
-import org.apache.accumulo.core.file.rfile.RFile;
+import org.apache.accumulo.core.file.rfile.MultiLevelIndex.BufferedWriter;
import org.apache.accumulo.core.file.rfile.MultiLevelIndex.IndexEntry;
import org.apache.accumulo.core.file.rfile.MultiLevelIndex.Reader;
-import org.apache.accumulo.core.file.rfile.MultiLevelIndex.Writer;
import
org.apache.accumulo.core.file.rfile.MultiLevelIndex.Reader.IndexIterator;
+import org.apache.accumulo.core.file.rfile.MultiLevelIndex.Writer;
import
org.apache.accumulo.core.file.rfile.RFileTest.SeekableByteArrayInputStream;
import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.hadoop.fs.FSDataInputStream;
@@ -55,7 +55,7 @@ public class MultiLevelIndexTest extends
FSDataOutputStream dos = new FSDataOutputStream(baos, new
FileSystem.Statistics("a"));
CachableBlockFile.Writer _cbw = new CachableBlockFile.Writer(dos, "gz",
CachedConfiguration.getInstance());
- Writer mliw = new Writer(_cbw, maxBlockSize);
+ BufferedWriter mliw = new BufferedWriter(new Writer(_cbw, maxBlockSize));
for (int i = 0; i < num; i++)
mliw.add(new Key(String.format("%05d000", i)), i, 0, 0, 0);
Propchange: accumulo/branches/1.4/src/server/
------------------------------------------------------------------------------
Merged /accumulo/trunk/src/server:r1329425
Merged /accumulo/trunk/server:r1329425
