提交 5df4131c authored 作者: Thomas Mueller's avatar Thomas Mueller

New experimental page store.

上级 5785fdf3
...@@ -57,9 +57,14 @@ public class Page { ...@@ -57,9 +57,14 @@ public class Page {
public static final int TYPE_FREE_LIST = 7; public static final int TYPE_FREE_LIST = 7;
/** /**
* A log page. * A stream trunk page.
*/ */
public static final int TYPE_LOG = 8; public static final int TYPE_STREAM_TRUNK = 8;
/**
* A stream data page.
*/
public static final int TYPE_STREAM_DATA = 9;
/** /**
* This is a root page. * This is a root page.
......
...@@ -26,25 +26,15 @@ public class PageFreeList extends Record { ...@@ -26,25 +26,15 @@ public class PageFreeList extends Record {
private final PageStore store; private final PageStore store;
private final BitField used = new BitField(); private final BitField used = new BitField();
private final int firstAddressed;
private final int pageCount; private final int pageCount;
private final int nextPage;
private boolean full; private boolean full;
private DataPage data; private DataPage data;
PageFreeList(PageStore store, int pageId, int firstAddressed) { PageFreeList(PageStore store, int pageId) {
setPos(pageId); setPos(pageId);
this.store = store; this.store = store;
this.firstAddressed = firstAddressed;
pageCount = (store.getPageSize() - DATA_START) * 8; pageCount = (store.getPageSize() - DATA_START) * 8;
for (int i = firstAddressed; i <= pageId; i++) { used.set(pageId);
used.set(getAddress(i));
}
nextPage = firstAddressed + pageCount;
}
private int getAddress(int pageId) {
return pageId - firstAddressed;
} }
/** /**
...@@ -54,20 +44,16 @@ public class PageFreeList extends Record { ...@@ -54,20 +44,16 @@ public class PageFreeList extends Record {
*/ */
int allocate() throws SQLException { int allocate() throws SQLException {
if (full) { if (full) {
PageFreeList next = getNext();
if (next == null) {
return -1; return -1;
} }
return next.allocate();
}
int free = used.nextClearBit(0); int free = used.nextClearBit(0);
if (free > pageCount) { if (free > pageCount) {
full = true; full = true;
return allocate(); return -1;
} }
used.set(free); used.set(free);
store.updateRecord(this, true, data); store.updateRecord(this, true, data);
return free + firstAddressed; return free + getPos();
} }
/** /**
...@@ -81,25 +67,8 @@ public class PageFreeList extends Record { ...@@ -81,25 +67,8 @@ public class PageFreeList extends Record {
return allocate(pos); return allocate(pos);
} }
public int getLastUsed() throws SQLException { int getLastUsed() {
if (nextPage < store.getPageCount()) { return used.getLastSetBit() + getPos();
PageFreeList next = getNext();
// TODO avoid recursion
return next.getLastUsed();
}
return used.getLastSetBit() + firstAddressed;
}
private PageFreeList getNext() throws SQLException {
PageFreeList next = (PageFreeList) store.getRecord(nextPage);
if (next == null) {
if (nextPage < store.getPageCount()) {
next = new PageFreeList(store, nextPage, nextPage);
next.read();
store.updateRecord(next, false, null);
}
}
return next;
} }
/** /**
...@@ -109,16 +78,9 @@ public class PageFreeList extends Record { ...@@ -109,16 +78,9 @@ public class PageFreeList extends Record {
* @return the page id, or -1 * @return the page id, or -1
*/ */
int allocate(int pos) throws SQLException { int allocate(int pos) throws SQLException {
if (pos - firstAddressed > pageCount) { int idx = pos - getPos();
PageFreeList next = getNext();
if (next == null) {
return -1;
}
return next.allocate(pos);
}
int idx = pos - firstAddressed;
if (idx >= 0 && !used.get(idx)) { if (idx >= 0 && !used.get(idx)) {
used.set(pos - firstAddressed); used.set(idx);
store.updateRecord(this, true, data); store.updateRecord(this, true, data);
} }
return pos; return pos;
...@@ -131,7 +93,7 @@ public class PageFreeList extends Record { ...@@ -131,7 +93,7 @@ public class PageFreeList extends Record {
*/ */
void free(int pageId) throws SQLException { void free(int pageId) throws SQLException {
full = false; full = false;
used.clear(pageId - firstAddressed); used.clear(pageId - getPos());
store.updateRecord(this, true, data); store.updateRecord(this, true, data);
} }
...@@ -153,6 +115,7 @@ public class PageFreeList extends Record { ...@@ -153,6 +115,7 @@ public class PageFreeList extends Record {
for (int i = 0; i < pageCount; i += 8) { for (int i = 0; i < pageCount; i += 8) {
used.setByte(i, data.readByte()); used.setByte(i, data.readByte());
} }
full = used.nextClearBit(0) >= pageCount * 8;
} }
public int getByteCount(DataPage dummy) { public int getByteCount(DataPage dummy) {
...@@ -170,4 +133,18 @@ public class PageFreeList extends Record { ...@@ -170,4 +133,18 @@ public class PageFreeList extends Record {
store.writePage(getPos(), data); store.writePage(getPos(), data);
} }
boolean isFull() {
return full;
}
/**
* Get the number of pages that can fit in a free list.
*
* @param pageSize the page size
* @return the number of pages
*/
static int getPagesAddressed(int pageSize) {
return (pageSize - DATA_START) * 8;
}
} }
...@@ -6,49 +6,30 @@ ...@@ -6,49 +6,30 @@
*/ */
package org.h2.store; package org.h2.store;
import java.io.EOFException;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.sql.SQLException; import java.sql.SQLException;
import org.h2.index.Page;
import org.h2.message.Message; import org.h2.message.Message;
import org.h2.message.Trace; import org.h2.message.Trace;
/** /**
* An output stream that writes into a page store. * An input stream that reads from a page store.
* The format is:
* <ul><li>0-3: parent page id
* </li><li>4-4: page type
* </li><li>5-5: stream id
* </li><li>6-9: the next page (if there is one) or length
* </li><li>10-remainder: data
* </li></ul>
*/ */
public class PageInputStream extends InputStream { public class PageInputStream extends InputStream {
/**
* The number of header bytes per stream page.
*/
public static final int OVERHEAD = 10;
private PageStore store; private PageStore store;
private final Trace trace; private final Trace trace;
private int parentPage; private int trunkNext;
private int type; private PageStreamTrunk trunk;
private int streamId = -1; private PageStreamData data;
private int nextPage;
private DataPage page;
private boolean endOfFile; private boolean endOfFile;
private int remaining; private int remaining;
private byte[] buffer = new byte[1]; private byte[] buffer = new byte[1];
public PageInputStream(PageStore store, int parentPage, int headPage, int type) { public PageInputStream(PageStore store, int trunkPage) {
this.store = store; this.store = store;
this.trace = store.getTrace(); this.trace = store.getTrace();
this.parentPage = parentPage; this.trunkNext = trunkPage;
this.type = type;
nextPage = headPage;
page = store.createDataPage();
} }
public int read() throws IOException { public int read() throws IOException {
...@@ -78,53 +59,47 @@ public class PageInputStream extends InputStream { ...@@ -78,53 +59,47 @@ public class PageInputStream extends InputStream {
} }
private int readBlock(byte[] buff, int off, int len) throws IOException { private int readBlock(byte[] buff, int off, int len) throws IOException {
try {
fillBuffer(); fillBuffer();
if (endOfFile) { if (endOfFile) {
return -1; return -1;
} }
int l = Math.min(remaining, len); int l = Math.min(remaining, len);
page.read(buff, off, l); data.read(buff, off, l);
remaining -= l; remaining -= l;
return l; return l;
} catch (SQLException e) {
throw Message.convertToIOException(e);
}
} }
private void fillBuffer() throws IOException { private void fillBuffer() throws SQLException {
if (remaining > 0 || endOfFile) { if (remaining > 0 || endOfFile) {
return; return;
} }
if (nextPage == 0) { if (trunkNext == 0) {
endOfFile = true; endOfFile = true;
return; return;
} }
page.reset(); if (trunk == null) {
try { trunk = new PageStreamTrunk(store, trunkNext);
store.readPage(nextPage, page); trunk.read();
} catch (SQLException e) { }
throw Message.convertToIOException(e); int next;
while (true) {
next = trunk.getNextPage();
if (next >= 0) {
break;
} }
int p = page.readInt(); trunk = new PageStreamTrunk(store, trunkNext);
int t = page.readByte(); trunk.read();
int id = page.readByte();
if (streamId == -1) {
// set the stream id on the first page
streamId = id;
}
boolean last = (t & Page.FLAG_LAST) != 0;
t &= ~Page.FLAG_LAST;
if (type != t || p != parentPage || id != streamId) {
throw new EOFException();
}
parentPage = nextPage;
if (last) {
nextPage = 0;
remaining = page.readInt();
} else {
nextPage = page.readInt();
remaining = store.getPageSize() - page.length();
} }
if (trace.isDebugEnabled()) { if (trace.isDebugEnabled()) {
trace.debug("pageIn.readPage " + parentPage + " next:" + nextPage); trace.debug("pageIn.readPage " + next);
} }
data = new PageStreamData(store, next, 0);
data.read();
remaining = data.getLength();
} }
} }
...@@ -59,7 +59,6 @@ public class PageLog { ...@@ -59,7 +59,6 @@ public class PageLog {
public static final int REMOVE = 4; public static final int REMOVE = 4;
private final PageStore store; private final PageStore store;
private int id;
private int pos; private int pos;
private Trace trace; private Trace trace;
...@@ -70,7 +69,6 @@ public class PageLog { ...@@ -70,7 +69,6 @@ public class PageLog {
private DataPage data; private DataPage data;
private long operation; private long operation;
private BitField undo = new BitField(); private BitField undo = new BitField();
private int[] reservedPages = new int[3];
PageLog(PageStore store, int firstPage) { PageLog(PageStore store, int firstPage) {
this.store = store; this.store = store;
...@@ -82,35 +80,20 @@ public class PageLog { ...@@ -82,35 +80,20 @@ public class PageLog {
/** /**
* Open the log for writing. For an existing database, the recovery * Open the log for writing. For an existing database, the recovery
* must be run first. * must be run first.
*
* @param id the log id
*/ */
void openForWriting(int id) throws SQLException { void openForWriting() {
this.id = id; trace.debug("log openForWriting firstPage:" + firstPage);
trace.debug("log openForWriting " + id + " firstPage:" + firstPage); pageOut = new PageOutputStream(store, firstPage);
pageOut = new PageOutputStream(store, 0, firstPage, Page.TYPE_LOG, id, true);
out = new DataOutputStream(pageOut); out = new DataOutputStream(pageOut);
try {
out.writeInt(id);
out.flush();
} catch (IOException e) {
throw Message.convertIOException(e, null);
}
} }
/** /**
* Open the log for reading. This will also read the log id. * Open the log for reading.
*
* @return the log id
*/ */
int openForReading() { void openForReading() {
in = new DataInputStream(new PageInputStream(store, 0, firstPage, Page.TYPE_LOG)); in = new DataInputStream(new PageInputStream(store, firstPage));
try { if (trace.isDebugEnabled()) {
id = in.readInt(); trace.debug("log openForReading firstPage:" + firstPage);
trace.debug("log openForReading " + id + " firstPage:" + firstPage + " id:" + id);
return id;
} catch (IOException e) {
return 0;
} }
} }
...@@ -123,8 +106,9 @@ public class PageLog { ...@@ -123,8 +106,9 @@ public class PageLog {
*/ */
void recover(boolean undo) throws SQLException { void recover(boolean undo) throws SQLException {
if (trace.isDebugEnabled()) { if (trace.isDebugEnabled()) {
trace.debug("log recover " + id + " undo:" + undo); trace.debug("log recover undo:" + undo);
} }
int logId = 0;
DataPage data = store.createDataPage(); DataPage data = store.createDataPage();
try { try {
pos = 0; pos = 0;
...@@ -148,7 +132,7 @@ public class PageLog { ...@@ -148,7 +132,7 @@ public class PageLog {
int tableId = in.readInt(); int tableId = in.readInt();
Row row = readRow(in, data); Row row = readRow(in, data);
if (!undo) { if (!undo) {
if (store.isSessionCommitted(sessionId, id, pos)) { if (store.isSessionCommitted(sessionId, logId, pos)) {
if (trace.isDebugEnabled()) { if (trace.isDebugEnabled()) {
trace.debug("log redo " + (x == ADD ? "+" : "-") + " table:" + tableId + " " + row); trace.debug("log redo " + (x == ADD ? "+" : "-") + " table:" + tableId + " " + row);
} }
...@@ -162,10 +146,10 @@ public class PageLog { ...@@ -162,10 +146,10 @@ public class PageLog {
} else if (x == COMMIT) { } else if (x == COMMIT) {
int sessionId = in.readInt(); int sessionId = in.readInt();
if (trace.isDebugEnabled()) { if (trace.isDebugEnabled()) {
trace.debug("log commit " + sessionId + " id:" + id + " pos:" + pos); trace.debug("log commit " + sessionId + " pos:" + pos);
} }
if (undo) { if (undo) {
store.setLastCommitForSession(sessionId, id, pos); store.setLastCommitForSession(sessionId, logId, pos);
} }
} else { } else {
if (trace.isDebugEnabled()) { if (trace.isDebugEnabled()) {
...@@ -221,7 +205,7 @@ public class PageLog { ...@@ -221,7 +205,7 @@ public class PageLog {
trace.debug("log undo " + pageId); trace.debug("log undo " + pageId);
} }
undo.set(pageId); undo.set(pageId);
reservePages(3); pageOut.prepareWriting(store.getPageSize() * 3);
out.write(UNDO); out.write(UNDO);
out.writeInt(pageId); out.writeInt(pageId);
out.write(page.getBytes(), 0, store.getPageSize()); out.write(page.getBytes(), 0, store.getPageSize());
...@@ -230,19 +214,6 @@ public class PageLog { ...@@ -230,19 +214,6 @@ public class PageLog {
} }
} }
private void reservePages(int pageCount) throws SQLException {
int todoThisIsSlow;
if (pageCount > reservedPages.length) {
reservedPages = new int[pageCount];
}
for (int i = 0; i < pageCount; i++) {
reservedPages[i] = store.allocatePage(true);
}
for (int i = pageCount - 1; i >= 0; i--) {
store.freePage(reservedPages[i], false, null);
}
}
/** /**
* Mark a committed transaction. * Mark a committed transaction.
* *
...@@ -258,7 +229,7 @@ public class PageLog { ...@@ -258,7 +229,7 @@ public class PageLog {
// database already closed // database already closed
return; return;
} }
reservePages(1); pageOut.prepareWriting(store.getPageSize());
out.write(COMMIT); out.write(COMMIT);
out.writeInt(session.getId()); out.writeInt(session.getId());
if (log.getFlushOnEachCommit()) { if (log.getFlushOnEachCommit()) {
...@@ -291,8 +262,7 @@ public class PageLog { ...@@ -291,8 +262,7 @@ public class PageLog {
int todoWriteIntoOutputDirectly; int todoWriteIntoOutputDirectly;
row.write(data); row.write(data);
reservePages(3 + data.length() / (store.getPageSize() - PageInputStream.OVERHEAD)); pageOut.prepareWriting(data.length() + store.getPageSize());
out.write(add ? ADD : REMOVE); out.write(add ? ADD : REMOVE);
out.writeInt(session.getId()); out.writeInt(session.getId());
out.writeInt(tableId); out.writeInt(tableId);
...@@ -309,7 +279,7 @@ public class PageLog { ...@@ -309,7 +279,7 @@ public class PageLog {
*/ */
void close() throws SQLException { void close() throws SQLException {
try { try {
trace.debug("log close " + id); trace.debug("log close");
if (out != null) { if (out != null) {
out.close(); out.close();
} }
...@@ -324,11 +294,11 @@ public class PageLog { ...@@ -324,11 +294,11 @@ public class PageLog {
* *
* @param id the new log id * @param id the new log id
*/ */
private void reopen(int id) throws SQLException { private void reopen() throws SQLException {
try { try {
trace.debug("log reopen"); trace.debug("log reopen");
out.close(); out.close();
openForWriting(id); openForWriting();
flush(); flush();
int todoDeleteOrReUsePages; int todoDeleteOrReUsePages;
} catch (IOException e) { } catch (IOException e) {
...@@ -347,15 +317,6 @@ public class PageLog { ...@@ -347,15 +317,6 @@ public class PageLog {
} }
} }
/**
* Get the log id.
*
* @return the log id
*/
int getId() {
return id;
}
/** /**
* Flush and close the log. * Flush and close the log.
*/ */
......
...@@ -9,50 +9,69 @@ package org.h2.store; ...@@ -9,50 +9,69 @@ package org.h2.store;
import java.io.IOException; import java.io.IOException;
import java.io.OutputStream; import java.io.OutputStream;
import java.sql.SQLException; import java.sql.SQLException;
import org.h2.index.Page;
import org.h2.message.Message; import org.h2.message.Message;
import org.h2.message.Trace; import org.h2.message.Trace;
import org.h2.util.IntArray;
/** /**
* An output stream that writes into a page store. * An output stream that writes into a page store.
*/ */
public class PageOutputStream extends OutputStream { public class PageOutputStream extends OutputStream {
private final Trace trace;
private PageStore store; private PageStore store;
private int type; private final Trace trace;
private int parentPage; private int trunkPageId;
private int pageId; private int trunkNext;
private int nextPage; private IntArray reservedPages = new IntArray();
private DataPage page; private PageStreamTrunk trunk;
private PageStreamData data;
private int reserved;
private int remaining; private int remaining;
private final boolean allocateAtEnd;
private byte[] buffer = new byte[1]; private byte[] buffer = new byte[1];
private boolean needFlush; private boolean needFlush;
private final int streamId;
private boolean writing; private boolean writing;
/** /**
* Create a new page output stream. * Create a new page output stream.
* *
* @param store the page store * @param store the page store
* @param parentPage the parent page id * @param trunkPage the first trunk page (already allocated)
* @param headPage the first page
* @param type the page type
* @param streamId the stream identifier
* @param allocateAtEnd whether new pages should be allocated at the end of
* the file
*/ */
public PageOutputStream(PageStore store, int parentPage, int headPage, int type, int streamId, boolean allocateAtEnd) { public PageOutputStream(PageStore store, int trunkPage) {
this.trace = store.getTrace(); this.trace = store.getTrace();
this.store = store; this.store = store;
this.parentPage = parentPage; this.trunkPageId = trunkPage;
this.pageId = headPage; }
this.type = type;
this.allocateAtEnd = allocateAtEnd; /**
this.streamId = streamId; * Allocate the required pages so that no pages need to be allocated while
page = store.createDataPage(); * writing.
initPage(); *
* @param minBuffer the number of bytes to allocate
*/
void prepareWriting(int minBuffer) throws SQLException {
if (reserved < minBuffer) {
int pageSize = store.getPageSize();
int capacityPerPage = PageStreamData.getCapacity(pageSize);
int pages = PageStreamTrunk.getPagesAddressed(pageSize);
// allocate x data pages
int pagesToAllocate = pages;
// the first trunk page is already allocated
if (reservedPages.size() == 0) {
reservedPages.add(trunkPageId);
}
int totalCapacity = pages * capacityPerPage;
while (totalCapacity < minBuffer) {
pagesToAllocate += pagesToAllocate;
totalCapacity += totalCapacity;
}
// allocate the next trunk page as well
pagesToAllocate++;
for (int i = 0; i < pagesToAllocate; i++) {
int page = store.allocatePage();
reservedPages.add(page);
}
}
} }
public void write(int b) throws IOException { public void write(int b) throws IOException {
...@@ -64,13 +83,27 @@ public class PageOutputStream extends OutputStream { ...@@ -64,13 +83,27 @@ public class PageOutputStream extends OutputStream {
write(b, 0, b.length); write(b, 0, b.length);
} }
private void initPage() { private void initNextData() {
page.reset(); int nextData = trunk == null ? -1 : trunk.getNextPage();
page.writeInt(parentPage); if (nextData < 0) {
page.writeByte((byte) type); int parent = trunkPageId;
page.writeByte((byte) streamId); if (trunkNext == 0) {
page.writeInt(0); trunkPageId = reservedPages.get(0);
remaining = store.getPageSize() - page.length(); reservedPages.remove(0);
} else {
trunkPageId = trunkNext;
}
int len = PageStreamTrunk.getPagesAddressed(store.getPageSize());
int[] pageIds = new int[len];
for (int i = 0; i < len; i++) {
pageIds[i] = reservedPages.get(i);
}
trunkNext = reservedPages.get(len);
trunk = new PageStreamTrunk(store, parent, trunkPageId, trunkNext, pageIds);
reservedPages.removeRange(0, len + 1);
}
data = new PageStreamData(store, trunk.getNextPage(), trunk.getPos());
data.initWrite();
} }
public void write(byte[] b, int off, int len) throws IOException { public void write(byte[] b, int off, int len) throws IOException {
...@@ -78,31 +111,24 @@ public class PageOutputStream extends OutputStream { ...@@ -78,31 +111,24 @@ public class PageOutputStream extends OutputStream {
return; return;
} }
if (writing) { if (writing) {
throw Message.throwInternalError("writing while still writing"); Message.throwInternalError("writing while still writing");
} }
writing = true; writing = true;
try { try {
while (len >= remaining) { while (len >= 0) {
page.write(b, off, remaining); int l = data.write(b, off, len);
off += remaining; if (l <= len) {
len -= remaining; data.write(null);
try { initNextData();
nextPage = store.allocatePage(allocateAtEnd);
} catch (SQLException e) {
throw Message.convertToIOException(e);
} }
page.setPos(4); reserved -= l;
page.writeByte((byte) type); off += l;
page.writeByte((byte) streamId); len -= l;
page.writeInt(nextPage);
storePage();
parentPage = pageId;
pageId = nextPage;
initPage();
} }
page.write(b, off, len);
needFlush = true; needFlush = true;
remaining -= len; remaining -= len;
} catch (SQLException e) {
throw Message.convertToIOException(e);
} finally { } finally {
writing = false; writing = false;
} }
...@@ -111,9 +137,9 @@ public class PageOutputStream extends OutputStream { ...@@ -111,9 +137,9 @@ public class PageOutputStream extends OutputStream {
private void storePage() throws IOException { private void storePage() throws IOException {
try { try {
if (trace.isDebugEnabled()) { if (trace.isDebugEnabled()) {
trace.debug("pageOut.storePage " + pageId + " next:" + nextPage); trace.debug("pageOut.storePage " + data.getPos());
} }
store.writePage(pageId, page); data.write(null);
} catch (SQLException e) { } catch (SQLException e) {
throw Message.convertToIOException(e); throw Message.convertToIOException(e);
} }
...@@ -121,12 +147,6 @@ public class PageOutputStream extends OutputStream { ...@@ -121,12 +147,6 @@ public class PageOutputStream extends OutputStream {
public void flush() throws IOException { public void flush() throws IOException {
if (needFlush) { if (needFlush) {
int len = page.length();
page.setPos(4);
page.writeByte((byte) (type | Page.FLAG_LAST));
page.writeByte((byte) streamId);
page.writeInt(store.getPageSize() - remaining - 9);
page.setPos(len);
storePage(); storePage();
needFlush = false; needFlush = false;
} }
......
...@@ -10,6 +10,7 @@ import java.io.IOException; ...@@ -10,6 +10,7 @@ import java.io.IOException;
import java.io.OutputStream; import java.io.OutputStream;
import java.sql.SQLException; import java.sql.SQLException;
import java.util.HashMap; import java.util.HashMap;
import java.util.zip.CRC32;
import org.h2.constant.ErrorCode; import org.h2.constant.ErrorCode;
import org.h2.engine.Database; import org.h2.engine.Database;
import org.h2.engine.Session; import org.h2.engine.Session;
...@@ -42,20 +43,25 @@ import org.h2.value.ValueInt; ...@@ -42,20 +43,25 @@ import org.h2.value.ValueInt;
import org.h2.value.ValueString; import org.h2.value.ValueString;
/** /**
* This class represents a file that is organized as a number of pages. The * This class represents a file that is organized as a number of pages. Page 0
* first page (page 0) contains the file header, which is never modified once * contains a static file header, and pages 1 and 2 both contain the variable
* the database is created. The format is: * file header (page 2 is a copy of page 1 and is only read if the checksum of
* page 1 is invalid). The format of page 0 is:
* <ul> * <ul>
* <li>0-47: file header (3 time "-- H2 0.5/B -- \n")</li> * <li>0-47: file header (3 time "-- H2 0.5/B -- \n")</li>
* <li>48-51: database page size in bytes * <li>48-51: page size in bytes (512 - 32768, must be a power of 2)</li>
* (512 - 32768, must be a power of 2)</li> * <li>52: write version (if not 0 the file is opened in read-only mode)</li>
* <li>52: write version (0, otherwise the file is opened in read-only mode)</li> * <li>53: read version (if not 0 opening the file fails)</li>
* <li>53: read version (0, otherwise opening the file fails)</li>
* <li>54-57: meta table root page number (usually 1)</li>
* <li>58-61: free list head page number (usually 2)</li>
* <li>62-65: log[0] head page number (usually 3)</li>
* <li>66-69: log[1] head page number (usually 4)</li>
* </ul> * </ul>
* The format of page 1 and 2 is:
* <ul>
* <li>0-7: write counter (incremented each time the header changes)</li>
* <li>8-11: log head page (initially 4)</li>
* <li>12-19: checksum of bytes 0-16 (CRC32)</li>
* </ul>
* Page 2 contains the first free list page.
* Page 3 contains the meta table root page.
* For a new database, page 4 contains the first log trunk page.
*/ */
public class PageStore implements CacheWriter { public class PageStore implements CacheWriter {
...@@ -64,6 +70,7 @@ public class PageStore implements CacheWriter { ...@@ -64,6 +70,7 @@ public class PageStore implements CacheWriter {
// TODO PageStore.openMetaIndex (desc and nulls first / last) // TODO PageStore.openMetaIndex (desc and nulls first / last)
// TODO btree index with fixed size values doesn't need offset and so on // TODO btree index with fixed size values doesn't need offset and so on
// TODO better checksums (for example, multiple fletcher)
// TODO log block allocation // TODO log block allocation
// TODO block compression: maybe http://en.wikipedia.org/wiki/LZJB // TODO block compression: maybe http://en.wikipedia.org/wiki/LZJB
// with RLE, specially for 0s. // with RLE, specially for 0s.
...@@ -93,6 +100,7 @@ public class PageStore implements CacheWriter { ...@@ -93,6 +100,7 @@ public class PageStore implements CacheWriter {
// TODO add a setting (that can be changed at runtime) to call fsync // TODO add a setting (that can be changed at runtime) to call fsync
// and delay on each commit // and delay on each commit
// TODO var int: see google protocol buffers // TODO var int: see google protocol buffers
// TODO SessionState.logId is no longer needed
/** /**
* The smallest possible page size. * The smallest possible page size.
...@@ -109,12 +117,11 @@ public class PageStore implements CacheWriter { ...@@ -109,12 +117,11 @@ public class PageStore implements CacheWriter {
*/ */
public static final int PAGE_SIZE_DEFAULT = 1024; public static final int PAGE_SIZE_DEFAULT = 1024;
/** private static final int PAGE_ID_FREE_LIST_ROOT = 2;
* The number of log streams. private static final int PAGE_ID_META_ROOT = 3;
*/
public static final int LOG_COUNT = 2; private static final int INCREMENT_PAGES = 128;
static final int INCREMENT_PAGES = 128;
private static final int READ_VERSION = 0; private static final int READ_VERSION = 0;
private static final int WRITE_VERSION = 0; private static final int WRITE_VERSION = 0;
...@@ -127,17 +134,16 @@ public class PageStore implements CacheWriter { ...@@ -127,17 +134,16 @@ public class PageStore implements CacheWriter {
private String fileName; private String fileName;
private FileStore file; private FileStore file;
private String accessMode; private String accessMode;
private int pageSize;
private int pageSizeShift;
private long writeCounter;
private int logFirstTrunkPage;
private int cacheSize; private int cacheSize;
private Cache cache; private Cache cache;
private int pageSize; private int freeListPagesPerList;
private int pageSizeShift;
private int metaTableRootPageId;
private int freeListRootPageId;
private int lastUsedPage;
private int activeLog;
private int[] logRootPageIds = new int[LOG_COUNT];
private boolean recoveryRunning; private boolean recoveryRunning;
private HashMap<Integer, SessionState> sessionStates = New.hashMap(); private HashMap<Integer, SessionState> sessionStates = New.hashMap();
...@@ -151,10 +157,7 @@ public class PageStore implements CacheWriter { ...@@ -151,10 +157,7 @@ public class PageStore implements CacheWriter {
*/ */
private int pageCount; private int pageCount;
/** private PageLog log;
* The transaction logs.
*/
private PageLog[] logs = new PageLog[LOG_COUNT];
private Schema metaSchema; private Schema metaSchema;
private TableData metaTable; private TableData metaTable;
...@@ -176,7 +179,7 @@ public class PageStore implements CacheWriter { ...@@ -176,7 +179,7 @@ public class PageStore implements CacheWriter {
this.database = database; this.database = database;
trace = database.getTrace(Trace.PAGE_STORE); trace = database.getTrace(Trace.PAGE_STORE);
int test; int test;
trace.setLevel(TraceSystem.DEBUG); // trace.setLevel(TraceSystem.DEBUG);
this.cacheSize = cacheSizeDefault; this.cacheSize = cacheSizeDefault;
String cacheType = database.getCacheType(); String cacheType = database.getCacheType();
this.cache = CacheLRU.getCache(this, cacheType, cacheSize); this.cache = CacheLRU.getCache(this, cacheType, cacheSize);
...@@ -215,51 +218,43 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -215,51 +218,43 @@ trace.setLevel(TraceSystem.DEBUG);
// existing // existing
file = database.openFile(fileName, accessMode, true); file = database.openFile(fileName, accessMode, true);
readHeader(); readHeader();
freeListPagesPerList = PageFreeList.getPagesAddressed(pageSize);
fileLength = file.length(); fileLength = file.length();
pageCount = (int) (fileLength / pageSize); pageCount = (int) (fileLength / pageSize);
initLogs(); initLog();
recover(true); recover(true);
recover(false); recover(false);
checkpoint(); checkpoint();
} else { } else {
// new // new
setPageSize(PAGE_SIZE_DEFAULT); setPageSize(PAGE_SIZE_DEFAULT);
freeListPagesPerList = PageFreeList.getPagesAddressed(pageSize);
file = database.openFile(fileName, accessMode, false); file = database.openFile(fileName, accessMode, false);
metaTableRootPageId = 1; logFirstTrunkPage = 4;
freeListRootPageId = 2; pageCount = 5;
pageCount = 3 + LOG_COUNT;
increaseFileSize(INCREMENT_PAGES - pageCount); increaseFileSize(INCREMENT_PAGES - pageCount);
getFreeList(); writeStaticHeader();
for (int i = 0; i < LOG_COUNT; i++) { writeVariableHeader();
logRootPageIds[i] = 3 + i; initLog();
}
writeHeader();
initLogs();
openMetaIndex(); openMetaIndex();
getLog().openForWriting(0); log.openForWriting();
switchLogIfPossible(); switchLog();
getLog().flush(); log.flush();
systemTableHeadPos = Index.EMPTY_HEAD; systemTableHeadPos = Index.EMPTY_HEAD;
} }
lastUsedPage = getFreeList().getLastUsed() + 1; // lastUsedPage = getFreeList().getLastUsed() + 1;
} catch (SQLException e) { } catch (SQLException e) {
close(); close();
throw e; throw e;
} }
} }
private void initLogs() {
for (int i = 0; i < LOG_COUNT; i++) {
logs[i] = new PageLog(this, logRootPageIds[i]);
}
}
/** /**
* Flush all pending changes to disk, and re-open the log file. * Flush all pending changes to disk, and re-open the log file.
*/ */
public void checkpoint() throws SQLException { public void checkpoint() throws SQLException {
trace.debug("checkpoint"); trace.debug("checkpoint");
if (getLog() == null || database.isReadOnly()) { if (log == null || database.isReadOnly()) {
// the file was never fully opened // the file was never fully opened
return; return;
} }
...@@ -271,26 +266,27 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -271,26 +266,27 @@ trace.setLevel(TraceSystem.DEBUG);
writeBack(rec); writeBack(rec);
} }
int todoFlushBeforeReopen; int todoFlushBeforeReopen;
// switch twice so there are no redo entries switchLog();
switchLogIfPossible();
switchLogIfPossible();
int todoWriteDeletedPages; int todoWriteDeletedPages;
} }
int pageCount = getFreeList().getLastUsed() + 1; // TODO shrink file if required here
trace.debug("pageCount:" + pageCount); // int pageCount = getFreeList().getLastUsed() + 1;
file.setLength(pageSize * pageCount); // trace.debug("pageCount:" + pageCount);
// file.setLength(pageSize * pageCount);
}
private void initLog() {
log = new PageLog(this, logFirstTrunkPage);
} }
private void switchLogIfPossible() throws SQLException { private void switchLog() throws SQLException {
trace.debug("switchLogIfPossible"); trace.debug("switchLogIfPossible");
if (database.isReadOnly()) { if (database.isReadOnly()) {
return; return;
} }
int id = getLog().getId(); log.close();
getLog().close();
activeLog = (activeLog + 1) % LOG_COUNT;
int todoCanOnlyReuseAfterLoggedChangesAreWritten; int todoCanOnlyReuseAfterLoggedChangesAreWritten;
getLog().openForWriting(id + 1); log.openForWriting();
// Session[] sessions = database.getSessions(true); // Session[] sessions = database.getSessions(true);
// int firstUncommittedLog = getLog().getId(); // int firstUncommittedLog = getLog().getId();
...@@ -316,7 +312,7 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -316,7 +312,7 @@ trace.setLevel(TraceSystem.DEBUG);
private void readHeader() throws SQLException { private void readHeader() throws SQLException {
long length = file.length(); long length = file.length();
if (length < PAGE_SIZE_MIN) { if (length < PAGE_SIZE_MIN * 2) {
throw Message.getSQLException(ErrorCode.FILE_CORRUPTED_1, fileName); throw Message.getSQLException(ErrorCode.FILE_CORRUPTED_1, fileName);
} }
database.notifyFileSize(length); database.notifyFileSize(length);
...@@ -338,10 +334,22 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -338,10 +334,22 @@ trace.setLevel(TraceSystem.DEBUG);
accessMode = "r"; accessMode = "r";
file = database.openFile(fileName, accessMode, true); file = database.openFile(fileName, accessMode, true);
} }
metaTableRootPageId = page.readInt(); CRC32 crc = new CRC32();
freeListRootPageId = page.readInt(); for (int i = 1;; i++) {
for (int i = 0; i < LOG_COUNT; i++) { if (i == 3) {
logRootPageIds[i] = page.readInt(); throw Message.getSQLException(ErrorCode.FILE_CORRUPTED_1, fileName);
}
page.reset();
readPage(i, page);
writeCounter = page.readLong();
logFirstTrunkPage = page.readInt();
crc.update(page.getBytes(), 0, 12);
long expected = crc.getValue();
long got = page.readLong();
if (expected == got) {
break;
}
crc.reset();
} }
} }
...@@ -372,21 +380,28 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -372,21 +380,28 @@ trace.setLevel(TraceSystem.DEBUG);
pageSizeShift = shift; pageSizeShift = shift;
} }
private void writeHeader() throws SQLException { private void writeStaticHeader() throws SQLException {
int todoChecksumForHeader;
DataPage page = DataPage.create(database, new byte[pageSize - FileStore.HEADER_LENGTH]); DataPage page = DataPage.create(database, new byte[pageSize - FileStore.HEADER_LENGTH]);
page.writeInt(pageSize); page.writeInt(pageSize);
page.writeByte((byte) WRITE_VERSION); page.writeByte((byte) WRITE_VERSION);
page.writeByte((byte) READ_VERSION); page.writeByte((byte) READ_VERSION);
page.writeInt(metaTableRootPageId);
page.writeInt(freeListRootPageId);
for (int i = 0; i < LOG_COUNT; i++) {
page.writeInt(logRootPageIds[i]);
}
file.seek(FileStore.HEADER_LENGTH); file.seek(FileStore.HEADER_LENGTH);
file.write(page.getBytes(), 0, pageSize - FileStore.HEADER_LENGTH); file.write(page.getBytes(), 0, pageSize - FileStore.HEADER_LENGTH);
} }
private void writeVariableHeader() throws SQLException {
DataPage page = DataPage.create(database, new byte[pageSize]);
page.writeLong(writeCounter);
page.writeInt(logFirstTrunkPage);
CRC32 crc = new CRC32();
crc.update(page.getBytes(), 0, 12);
page.writeLong(crc.getValue());
file.seek(pageSize);
file.write(page.getBytes(), 0, page.length());
file.seek(pageSize + pageSize);
file.write(page.getBytes(), 0, page.length());
}
/** /**
* Close the file without writing anything. * Close the file without writing anything.
*/ */
...@@ -403,7 +418,7 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -403,7 +418,7 @@ trace.setLevel(TraceSystem.DEBUG);
} }
public void flushLog() throws SQLException { public void flushLog() throws SQLException {
getLog().flush(); log.flush();
} }
public Trace getTrace() { public Trace getTrace() {
...@@ -436,15 +451,17 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -436,15 +451,17 @@ trace.setLevel(TraceSystem.DEBUG);
trace.debug("updateRecord " + record.toString()); trace.debug("updateRecord " + record.toString());
} }
} }
database.checkWritingAllowed();
record.setChanged(true); record.setChanged(true);
int pos = record.getPos(); int pos = record.getPos();
getFreeList().allocate(pos); allocatePage(pos);
// getFreeList().allocate(pos);
cache.update(pos, record); cache.update(pos, record);
if (logUndo && !recoveryRunning) { if (logUndo && !recoveryRunning) {
if (old == null) { if (old == null) {
old = readPage(pos); old = readPage(pos);
} }
getLog().addUndo(record.getPos(), old); log.addUndo(record.getPos(), old);
} }
} }
} }
...@@ -458,40 +475,65 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -458,40 +475,65 @@ trace.setLevel(TraceSystem.DEBUG);
return allocatePage(false); return allocatePage(false);
} }
private PageFreeList getFreeList() throws SQLException { private PageFreeList getFreeList(int i) throws SQLException {
PageFreeList free = (PageFreeList) cache.find(freeListRootPageId); int p;
if (free == null) { if (i == 0) {
free = new PageFreeList(this, freeListRootPageId, 3 + LOG_COUNT); // TODO simplify
free.read(); p = PAGE_ID_FREE_LIST_ROOT;
cache.put(free); } else {
p = i * freeListPagesPerList;
}
PageFreeList list = (PageFreeList) getRecord(p);
if (list == null) {
list = new PageFreeList(this, p);
list.read();
cache.put(list);
}
return list;
}
private void freePage(int pageId) throws SQLException {
if (pageId < 0) {
System.out.println("stop");
} }
return free; PageFreeList list = getFreeList(pageId / freeListPagesPerList);
list.free(pageId);
}
private void allocatePage(int pageId) throws SQLException {
PageFreeList list = getFreeList(pageId / freeListPagesPerList);
list.allocate(pageId % freeListPagesPerList);
} }
/** /**
* Allocate a page. * Allocate a new page.
* *
* @param atEnd if the allocated page must be at the end of the file * @param atEnd allocate at the end of the file
* @return the page id * @return the page id
*/ */
public int allocatePage(boolean atEnd) throws SQLException { int allocatePage(boolean atEnd) throws SQLException {
PageFreeList list = getFreeList(); int pos;
int id; if (atEnd) {
while (true) { PageFreeList list = getFreeList(pageCount / freeListPagesPerList);
id = atEnd ? list.allocateAtEnd(++lastUsedPage) : list.allocate(); pos = list.getLastUsed() + 1;
if (id < 0) { list.allocate(pos);
increaseFileSize(INCREMENT_PAGES);
} else { } else {
// TODO could remember the first possible free list page
for (int i = 0;; i++) {
if (i * freeListPagesPerList > pageCount) {
increaseFileSize(INCREMENT_PAGES);
}
PageFreeList list = getFreeList(i);
if (!list.isFull()) {
pos = list.allocate();
break; break;
} }
} }
if (trace.isDebugEnabled()) {
trace.debug("allocated " + id + " atEnd:" + atEnd);
} }
if (id >= pageCount) { if (pos > pageCount) {
increaseFileSize(INCREMENT_PAGES); increaseFileSize(INCREMENT_PAGES);
} }
return id; return pos;
} }
private void increaseFileSize(int increment) throws SQLException { private void increaseFileSize(int increment) throws SQLException {
...@@ -512,18 +554,15 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -512,18 +554,15 @@ trace.setLevel(TraceSystem.DEBUG);
if (trace.isDebugEnabled()) { if (trace.isDebugEnabled()) {
trace.debug("freePage " + pageId); trace.debug("freePage " + pageId);
} }
if (lastUsedPage == pageId) {
lastUsedPage--;
}
cache.remove(pageId); cache.remove(pageId);
getFreeList().free(pageId); freePage(pageId);
if (recoveryRunning) { if (recoveryRunning) {
writePage(pageId, createDataPage()); writePage(pageId, createDataPage());
} else if (logUndo) { } else if (logUndo) {
if (old == null) { if (old == null) {
old = readPage(pageId); old = readPage(pageId);
} }
getLog().addUndo(pageId, old); log.addUndo(pageId, old);
} }
} }
...@@ -612,25 +651,6 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -612,25 +651,6 @@ trace.setLevel(TraceSystem.DEBUG);
cache.remove(pageId); cache.remove(pageId);
} }
/**
* Set the root page of the free list.
*
* @param pageId the first free list page
* @param existing if the page already exists
* @param next the next page
*/
void setFreeListRootPage(int pageId, boolean existing, int next) throws SQLException {
this.freeListRootPageId = pageId;
if (!existing) {
PageFreeList free = new PageFreeList(this, pageId, next);
updateRecord(free, false, null);
}
}
PageLog getLog() {
return logs[activeLog];
}
Database getDatabase() { Database getDatabase() {
return database; return database;
} }
...@@ -652,26 +672,26 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -652,26 +672,26 @@ trace.setLevel(TraceSystem.DEBUG);
try { try {
recoveryRunning = true; recoveryRunning = true;
int maxId = 0; int maxId = 0;
for (int i = 0; i < LOG_COUNT; i++) { // for (int i = 0; i < LOG_COUNT; i++) {
int id = logs[i].openForReading(); // int id = logs[i].openForReading();
if (id > maxId) { // if (id > maxId) {
maxId = id; // maxId = id;
activeLog = i; // activeLog = i;
} // }
} // }
for (int i = 0; i < LOG_COUNT; i++) { // for (int i = 0; i < LOG_COUNT; i++) {
int j; // int j;
if (undo) { // if (undo) {
// undo: start with the newest file and go backward // // undo: start with the newest file and go backward
j = Math.abs(activeLog - i) % LOG_COUNT; // j = Math.abs(activeLog - i) % LOG_COUNT;
} else { // } else {
// redo: start with the oldest log file // // redo: start with the oldest log file
j = (activeLog + 1 + i) % LOG_COUNT; // j = (activeLog + 1 + i) % LOG_COUNT;
} // }
logs[j].recover(undo); // logs[j].recover(undo);
} // }
if (!undo) { if (!undo) {
switchLogIfPossible(); switchLog();
int todoProbablyStillRequiredForTwoPhaseCommit; int todoProbablyStillRequiredForTwoPhaseCommit;
sessionStates = New.hashMap(); sessionStates = New.hashMap();
} }
...@@ -711,7 +731,7 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -711,7 +731,7 @@ trace.setLevel(TraceSystem.DEBUG);
*/ */
public void logAddOrRemoveRow(Session session, int tableId, Row row, boolean add) throws SQLException { public void logAddOrRemoveRow(Session session, int tableId, Row row, boolean add) throws SQLException {
if (!recoveryRunning) { if (!recoveryRunning) {
getLog().logAddOrRemoveRow(session, tableId, row, add); log.logAddOrRemoveRow(session, tableId, row, add);
} }
} }
...@@ -721,7 +741,7 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -721,7 +741,7 @@ trace.setLevel(TraceSystem.DEBUG);
* @param session the session * @param session the session
*/ */
public void commit(Session session) throws SQLException { public void commit(Session session) throws SQLException {
getLog().commit(session); log.commit(session);
} }
/** /**
...@@ -817,7 +837,7 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -817,7 +837,7 @@ trace.setLevel(TraceSystem.DEBUG);
cols.add(new Column("OPTIONS", Value.STRING)); cols.add(new Column("OPTIONS", Value.STRING));
cols.add(new Column("COLUMNS", Value.STRING)); cols.add(new Column("COLUMNS", Value.STRING));
metaSchema = new Schema(database, 0, "", null, true); metaSchema = new Schema(database, 0, "", null, true);
int headPos = metaTableRootPageId; int headPos = PAGE_ID_META_ROOT;
metaTable = new TableData(metaSchema, "PAGE_INDEX", metaTable = new TableData(metaSchema, "PAGE_INDEX",
META_TABLE_ID, cols, true, true, false, headPos, database.getSystemSession()); META_TABLE_ID, cols, true, true, false, headPos, database.getSystemSession());
metaIndex = (PageScanIndex) metaTable.getScanIndex( metaIndex = (PageScanIndex) metaTable.getScanIndex(
...@@ -927,4 +947,30 @@ trace.setLevel(TraceSystem.DEBUG); ...@@ -927,4 +947,30 @@ trace.setLevel(TraceSystem.DEBUG);
metaIndex.remove(session, row); metaIndex.remove(session, row);
} }
private void updateChecksum(byte[] d, int pos) {
int ps = pageSize;
int s1 = 255 + (d[0] & 255), s2 = 255 + s1;
s2 += s1 += d[1] & 255;
s2 += s1 += d[(ps >> 1) - 1] & 255;
s2 += s1 += d[ps >> 1] & 255;
s2 += s1 += d[ps - 2] & 255;
s2 += s1 += d[ps - 1] & 255;
d[5] = (byte) (((s1 & 255) + (s1 >> 8)) ^ pos);
d[6] = (byte) (((s2 & 255) + (s2 >> 8)) ^ (pos >> 8));
}
private void verifyChecksum(byte[] d, int pos) throws SQLException {
int ps = pageSize;
int s1 = 255 + (d[0] & 255), s2 = 255 + s1;
s2 += s1 += d[1] & 255;
s2 += s1 += d[(ps >> 1) - 1] & 255;
s2 += s1 += d[ps >> 1] & 255;
s2 += s1 += d[ps - 2] & 255;
s2 += s1 += d[ps - 1] & 255;
if (d[5] != (byte) (((s1 & 255) + (s1 >> 8)) ^ pos)
|| d[6] != (byte) (((s2 & 255) + (s2 >> 8)) ^ (pos >> 8))) {
throw Message.getSQLException(ErrorCode.FILE_CORRUPTED_1, "wrong checksum");
}
}
} }
/*
* Copyright 2004-2009 H2 Group. Multiple-Licensed under the H2 License,
* Version 1.0, and under the Eclipse Public License, Version 1.0
* (http://h2database.com/html/license.html).
* Initial Developer: H2 Group
*/
package org.h2.store;
import java.sql.SQLException;
import org.h2.constant.ErrorCode;
import org.h2.index.Page;
import org.h2.message.Message;
/**
* A data page of a stream. The format is:
* <ul>
* <li>0-3: the trunk page id</li>
* <li>4-4: page type</li>
* <li>5-8: the number of bytes used</li>
* <li>9-remainder: data</li>
* </ul>
*/
public class PageStreamData extends Record {
private static final int LENGTH_START = 5;
private static final int DATA_START = 9;
private final PageStore store;
private final int trunk;
private DataPage data;
private int remaining;
private int length;
PageStreamData(PageStore store, int pageId, int trunk) {
setPos(pageId);
this.store = store;
this.trunk = trunk;
}
/**
* Read the page from the disk.
*/
void read() throws SQLException {
data = store.createDataPage();
store.readPage(getPos(), data);
int t = data.readByte();
if (t != Page.TYPE_STREAM_DATA) {
throw Message.getSQLException(ErrorCode.FILE_CORRUPTED_1, "pos:" + getPos() + " type:" + t +
" expected type:" + Page.TYPE_STREAM_DATA);
}
length = data.readInt();
}
public int getByteCount(DataPage dummy) {
return store.getPageSize();
}
/**
* Write the header data.
*/
void initWrite() {
data = store.createDataPage();
data.writeInt(trunk);
data.writeByte((byte) Page.TYPE_STREAM_DATA);
data.writeInt(0);
remaining = store.getPageSize() - data.length();
}
/**
* Write the data to the buffer.
*
* @param buff the source data
* @param off the offset in the source buffer
* @param len the number of bytes to write
* @return the number of bytes written
*/
int write(byte[] buff, int offset, int len) {
int max = Math.min(remaining, len);
data.write(buff, offset, max);
length += max;
remaining -= max;
return max;
}
public void write(DataPage buff) throws SQLException {
data.setInt(LENGTH_START, length);
store.writePage(getPos(), data);
}
/**
* Get the number of bytes that fit in a page.
*
* @param pageSize the page size
* @return the number of bytes
*/
static int getCapacity(int pageSize) {
return pageSize - DATA_START;
}
int getLength() {
return length;
}
/**
* Read the next bytes from the buffer.
*
* @param buff the target buffer
* @param off the offset in the target buffer
* @param len the number of bytes to read
*/
void read(byte[] buff, int off, int len) {
data.read(buff, off, len);
}
}
\ No newline at end of file
/*
* Copyright 2004-2009 H2 Group. Multiple-Licensed under the H2 License,
* Version 1.0, and under the Eclipse Public License, Version 1.0
* (http://h2database.com/html/license.html).
* Initial Developer: H2 Group
*/
package org.h2.store;
import java.sql.SQLException;
import org.h2.constant.ErrorCode;
import org.h2.index.Page;
import org.h2.message.Message;
/**
* A trunk page of a stream. It contains the page numbers of the stream, and
* the page number of the next trunk. The format is:
* <ul>
* <li>0-3: the last trunk page, or 0 if none</li>
* <li>4-4: page type</li>
* <li>5-8: the next trunk page</li>
* <li>9-12: the number of pages</li>
* <li>13-remainder: page ids</li>
* </ul>
*/
public class PageStreamTrunk extends Record {
private static final int DATA_START = 13;
private final PageStore store;
private int parent;
private int nextTrunk;
private int[] pageIds;
private int pageCount;
private DataPage data;
private int index;
PageStreamTrunk(PageStore store, int parent, int pageId, int next, int[] pageIds) {
setPos(pageId);
this.parent = parent;
this.store = store;
this.nextTrunk = next;
this.pageCount = pageIds.length;
this.pageIds = pageIds;
}
public PageStreamTrunk(PageStore store, int pageId) {
setPos(pageId);
this.store = store;
}
/**
* Read the page from the disk.
*/
void read() throws SQLException {
data = store.createDataPage();
store.readPage(getPos(), data);
parent = data.readInt();
int t = data.readByte();
if (t != Page.TYPE_STREAM_TRUNK) {
throw Message.getSQLException(ErrorCode.FILE_CORRUPTED_1, "pos:" + getPos() + " type:" + t + " parent:" + parent
+ " expected type:" + Page.TYPE_STREAM_TRUNK);
}
nextTrunk = data.readInt();
pageCount = data.readInt();
for (int i = 0; i < pageCount; i++) {
pageIds[i] = data.readInt();
}
}
void setNextPage(int page) {
pageIds[index++] = page;
}
int getNextPage() {
if (index >= pageIds.length) {
return -1;
}
return pageIds[index++];
}
int getNextTrunk() {
return nextTrunk;
}
public int getByteCount(DataPage dummy) {
return store.getPageSize();
}
public void write(DataPage buff) throws SQLException {
data = store.createDataPage();
data.writeInt(parent);
data.writeByte((byte) Page.TYPE_STREAM_TRUNK);
data.writeInt(nextTrunk);
data.writeInt(pageCount);
for (int i = 0; i < pageCount; i++) {
data.writeInt(pageIds[i]);
}
store.writePage(getPos(), data);
}
/**
* Get the number of pages that can be addressed in a stream trunk page.
*
* @param pageSize the page size
* @return the number of pages
*/
static int getPagesAddressed(int pageSize) {
return (pageSize - DATA_START) / 4;
}
}
...@@ -790,9 +790,11 @@ public class Recover extends Tool implements DataHandler { ...@@ -790,9 +790,11 @@ public class Recover extends Tool implements DataHandler {
case Page.TYPE_FREE_LIST: case Page.TYPE_FREE_LIST:
writer.println("-- page " + page + ": free list " + (last ? "(last)" : "")); writer.println("-- page " + page + ": free list " + (last ? "(last)" : ""));
break; break;
case Page.TYPE_LOG: case Page.TYPE_STREAM_TRUNK:
writer.println("-- page " + page + ": log " + (last ? "(last)" : "")); writer.println("-- page " + page + ": log trunk");
dumpPageLog(writer, s, last); break;
case Page.TYPE_STREAM_DATA:
writer.println("-- page " + page + ": log data");
break; break;
default: default:
writer.println("-- page " + page + ": ERROR unknown type " + type); writer.println("-- page " + page + ": ERROR unknown type " + type);
...@@ -800,9 +802,9 @@ public class Recover extends Tool implements DataHandler { ...@@ -800,9 +802,9 @@ public class Recover extends Tool implements DataHandler {
} }
} }
writeSchema(writer); writeSchema(writer);
for (int i = 0; i < PageStore.LOG_COUNT; i++) { // for (int i = 0; i < PageStore.LOG_COUNT; i++) {
dumpPageLogStream(writer, store, logHead + i, pageSize); // dumpPageLogStream(writer, store, logHead + i, pageSize);
} // }
writer.close(); writer.close();
} catch (Throwable e) { } catch (Throwable e) {
writeError(writer, e); writeError(writer, e);
...@@ -815,7 +817,8 @@ public class Recover extends Tool implements DataHandler { ...@@ -815,7 +817,8 @@ public class Recover extends Tool implements DataHandler {
private void dumpPageLogStream(PrintWriter writer, FileStore store, int logHead, int pageSize) throws IOException, SQLException { private void dumpPageLogStream(PrintWriter writer, FileStore store, int logHead, int pageSize) throws IOException, SQLException {
DataPage s = DataPage.create(this, pageSize); DataPage s = DataPage.create(this, pageSize);
DataInputStream in = new DataInputStream( DataInputStream in = new DataInputStream(
new PageInputStream(writer, this, store, logHead, pageSize, 0, Page.TYPE_LOG) new PageInputStream(writer, this, store, logHead, pageSize, 0,
Page.TYPE_STREAM_TRUNK)
); );
int logId = in.readInt(); int logId = in.readInt();
writer.println("-- log " + logId); writer.println("-- log " + logId);
...@@ -843,7 +846,6 @@ public class Recover extends Tool implements DataHandler { ...@@ -843,7 +846,6 @@ public class Recover extends Tool implements DataHandler {
break; break;
} }
} }
} }
private void setStorage(int storageId) { private void setStorage(int storageId) {
......
...@@ -310,4 +310,20 @@ public class IntArray { ...@@ -310,4 +310,20 @@ public class IntArray {
return buff.append('}').toString(); return buff.append('}').toString();
} }
/**
* Remove a number of elements.
*
* @param fromIndex the index of the first item to remove
* @param toIndex upper bound (exclusive)
*/
public void removeRange(int fromIndex, int toIndex) {
if (SysProperties.CHECK) {
if (fromIndex > toIndex || toIndex >= size) {
throw new ArrayIndexOutOfBoundsException("from=" + fromIndex + " to=" + toIndex + " size=" + size);
}
}
System.arraycopy(data, toIndex, data, fromIndex, size - toIndex);
size -= toIndex - fromIndex;
}
} }
...@@ -28,6 +28,16 @@ public class TestIntArray extends TestBase { ...@@ -28,6 +28,16 @@ public class TestIntArray extends TestBase {
public void test() { public void test() {
testInit(); testInit();
testRandom(); testRandom();
testRemoveRange();
}
private void testRemoveRange() {
IntArray array = new IntArray(new int[] {1, 2, 3, 4, 5});
array.removeRange(1, 3);
assertEquals(3, array.size());
assertEquals(1, array.get(0));
assertEquals(4, array.get(1));
assertEquals(5, array.get(2));
} }
private void testInit() { private void testInit() {
......
...@@ -313,7 +313,7 @@ public class TestPageStore extends TestBase { ...@@ -313,7 +313,7 @@ public class TestPageStore extends TestBase {
if (file) { if (file) {
out = new BufferedOutputStream(new FileOutputStream(f), 4 * 1024); out = new BufferedOutputStream(new FileOutputStream(f), 4 * 1024);
} else { } else {
out = new PageOutputStream(store, 0, head, Page.TYPE_LOG, 0, false); out = new PageOutputStream(store, 0);
} }
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
out.write(buff); out.write(buff);
...@@ -322,7 +322,7 @@ public class TestPageStore extends TestBase { ...@@ -322,7 +322,7 @@ public class TestPageStore extends TestBase {
if (file) { if (file) {
in = new BufferedInputStream(new FileInputStream(f), 4 * 1024); in = new BufferedInputStream(new FileInputStream(f), 4 * 1024);
} else { } else {
in = new PageInputStream(store, 0, head, Page.TYPE_LOG); in = new PageInputStream(store, 0);
} }
while (true) { while (true) {
int len = in.read(buff); int len = in.read(buff);
...@@ -353,14 +353,14 @@ public class TestPageStore extends TestBase { ...@@ -353,14 +353,14 @@ public class TestPageStore extends TestBase {
byte[] data = new byte[len]; byte[] data = new byte[len];
random.nextBytes(data); random.nextBytes(data);
int head = store.allocatePage(); int head = store.allocatePage();
PageOutputStream out = new PageOutputStream(store, 0, head, Page.TYPE_LOG, 0, false); PageOutputStream out = new PageOutputStream(store, 0);
for (int p = 0; p < len;) { for (int p = 0; p < len;) {
int l = len == 0 ? 0 : Math.min(len - p, random.nextInt(len / 10)); int l = len == 0 ? 0 : Math.min(len - p, random.nextInt(len / 10));
out.write(data, p, l); out.write(data, p, l);
p += l; p += l;
} }
out.close(); out.close();
PageInputStream in = new PageInputStream(store, 0, head, Page.TYPE_LOG); PageInputStream in = new PageInputStream(store, 0);
byte[] data2 = new byte[len]; byte[] data2 = new byte[len];
for (int off = 0;;) { for (int off = 0;;) {
int l = random.nextInt(1 + len / 10) + 1; int l = random.nextInt(1 + len / 10) + 1;
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论