File D:\maven-qcr-plugin-testsunitaires\qcr4java\h2-1.3.170-sources\target\site\qc\lecode\store\PageLog.java.html

1 /*
2 * Copyright 2004-2011 H2 Group. Multiple-Licensed under the H2 License,
3 * Version 1.0, and under the Eclipse Public License, Version 1.0
4 * (http://h2database.com/html/license.html).
5 * Initial Developer: H2 Group
6 */
7 package org.h2.store;
8
9 import java.io.IOException;
10 import java.util.ArrayList;
11 import java.util.Arrays;
12 import java.util.HashMap;
13 import org.h2.compress.CompressLZF;
14 import org.h2.constant.ErrorCode;
15 import org.h2.constant.SysProperties;
16 import org.h2.engine.Session;
17 import org.h2.message.DbException;
18 import org.h2.message.Trace;
19 import org.h2.result.Row;
20 import org.h2.util.BitField;
21 import org.h2.util.IntArray;
22 import org.h2.util.IntIntHashMap;
23 import org.h2.util.New;
24 import org.h2.value.Value;
25 import org.h2.value.ValueNull;
26
27 /**
28 * Transaction log mechanism. The stream contains a list of records. The data
29 * format for a record is:
30 * <ul>
31 * <li>type (0: no-op, 1: undo, 2: commit, ...)</li>
32 * <li>data</li>
33 * </ul>
34 * The transaction log is split into sections.
35 * A checkpoint starts a new section.
36 */
37 public class PageLog {
38
39 /**
40 * No operation.
41 */
42 public static final int NOOP = 0;
43
44 /**
45 * An undo log entry. Format: page id: varInt, size, page. Size 0 means
46 * uncompressed, size 1 means empty page, otherwise the size is the number
47 * of compressed bytes.
48 */
49 public static final int UNDO = 1;
50
51 /**
52 * A commit entry of a session.
53 * Format: session id: varInt.
54 */
55 public static final int COMMIT = 2;
56
57 /**
58 * A prepare commit entry for a session.
59 * Format: session id: varInt, transaction name: string.
60 */
61 public static final int PREPARE_COMMIT = 3;
62
63 /**
64 * Roll back a prepared transaction.
65 * Format: session id: varInt.
66 */
67 public static final int ROLLBACK = 4;
68
69 /**
70 * Add a record to a table.
71 * Format: session id: varInt, table id: varInt, row.
72 */
73 public static final int ADD = 5;
74
75 /**
76 * Remove a record from a table.
77 * Format: session id: varInt, table id: varInt, row.
78 */
79 public static final int REMOVE = 6;
80
81 /**
82 * Truncate a table.
83 * Format: session id: varInt, table id: varInt.
84 */
85 public static final int TRUNCATE = 7;
86
87 /**
88 * Perform a checkpoint. The log section id is incremented.
89 * Format: -
90 */
91 public static final int CHECKPOINT = 8;
92
93 /**
94 * Free a log page.
95 * Format: count: varInt, page ids: varInt
96 */
97 public static final int FREE_LOG = 9;
98
99 /**
100 * The recovery stage to undo changes (re-apply the backup).
101 */
102 static final int RECOVERY_STAGE_UNDO = 0;
103
104 /**
105 * The recovery stage to allocate pages used by the transaction log.
106 */
107 static final int RECOVERY_STAGE_ALLOCATE = 1;
108
109 /**
110 * The recovery stage to redo operations.
111 */
112 static final int RECOVERY_STAGE_REDO = 2;
113
114 private static final boolean COMPRESS_UNDO = true;
115
116 private final PageStore store;
117 private final Trace trace;
118
119 private Data writeBuffer;
120 private PageOutputStream pageOut;
121 private int firstTrunkPage;
122 private int firstDataPage;
123 private final Data dataBuffer;
124 private int logKey;
125 private int logSectionId, logPos;
126 private int firstSectionId;
127
128 private final CompressLZF compress;
129 private final byte[] compressBuffer;
130
131 /**
132 * If the bit is set, the given page was written to the current log section.
133 * The undo entry of these pages doesn't need to be written again.
134 */
135 private BitField undo = new BitField();
136
137 /**
138 * The undo entry of those pages was written in any log section.
139 * These pages may not be used in the transaction log.
140 */
141 private final BitField undoAll = new BitField();
142
143 /**
144 * The map of section ids (key) and data page where the section starts (value).
145 */
146 private final IntIntHashMap logSectionPageMap = new IntIntHashMap();
147
148 /**
149 * The session state map.
150 * Only used during recovery.
151 */
152 private HashMap<Integer, SessionState> sessionStates = New.hashMap();
153
154 /**
155 * The map of pages used by the transaction log.
156 * Only used during recovery.
157 */
158 private BitField usedLogPages;
159
160 /**
161 * This flag is set while freeing up pages.
162 */
163 private boolean freeing;
164
165 PageLog(PageStore store) {
166 this.store = store;
167 dataBuffer = store.createData();
168 trace = store.getTrace();
169 compress = new CompressLZF();
170 compressBuffer = new byte[store.getPageSize() * 2];
171 }
172
173 /**
174 * Open the log for writing. For an existing database, the recovery
175 * must be run first.
176 *
177 * @param newFirstTrunkPage the first trunk page
178 * @param atEnd whether only pages at the end of the file should be used
179 */
180 void openForWriting(int newFirstTrunkPage, boolean atEnd) {
181 trace.debug("log openForWriting firstPage: " + newFirstTrunkPage);
182 this.firstTrunkPage = newFirstTrunkPage;
183 logKey++;
184 pageOut = new PageOutputStream(store, newFirstTrunkPage, undoAll, logKey, atEnd);
185 pageOut.reserve(1);
186 // pageBuffer = new BufferedOutputStream(pageOut, 8 * 1024);
187 store.setLogFirstPage(logKey, newFirstTrunkPage, pageOut.getCurrentDataPageId());
188 writeBuffer = store.createData();
189 }
190
191 /**
192 * Free up all pages allocated by the log.
193 */
194 void free() {
195 if (trace.isDebugEnabled()) {
196 trace.debug("log free");
197 }
198 int currentDataPage = 0;
199 if (pageOut != null) {
200 currentDataPage = pageOut.getCurrentDataPageId();
201 pageOut.freeReserved();
202 }
203 try {
204 freeing = true;
205 int first = 0;
206 int loopDetect = 1024, loopCount = 0;
207 PageStreamTrunk.Iterator it = new PageStreamTrunk.Iterator(store, firstTrunkPage);
208 while (firstTrunkPage != 0 && firstTrunkPage < store.getPageCount()) {
209 PageStreamTrunk t = it.next();
210 if (t == null) {
211 if (it.canDelete()) {
212 store.free(firstTrunkPage, false);
213 }
214 break;
215 }
216 if (loopCount++ >= loopDetect) {
217 first = t.getPos();
218 loopCount = 0;
219 loopDetect *= 2;
220 } else if (first != 0 && first == t.getPos()) {
221 throw DbException.throwInternalError("endless loop at " + t);
222 }
223 t.free(currentDataPage);
224 firstTrunkPage = t.getNextTrunk();
225 }
226 } finally {
227 freeing = false;
228 }
229 }
230
231 /**
232 * Open the log for reading.
233 *
234 * @param newLogKey the first expected log key
235 * @param newFirstTrunkPage the first trunk page
236 * @param newFirstDataPage the index of the first data page
237 */
238 void openForReading(int newLogKey, int newFirstTrunkPage, int newFirstDataPage) {
239 this.logKey = newLogKey;
240 this.firstTrunkPage = newFirstTrunkPage;
241 this.firstDataPage = newFirstDataPage;
242 }
243
244 /**
245 * Run one recovery stage. There are three recovery stages: 0: only the undo
246 * steps are run (restoring the state before the last checkpoint). 1: the
247 * pages that are used by the transaction log are allocated. 2: the
248 * committed operations are re-applied.
249 *
250 * @param stage the recovery stage
251 * @return whether the transaction log was empty
252 */
253 boolean recover(int stage) {
254 if (trace.isDebugEnabled()) {
255 trace.debug("log recover stage: " + stage);
256 }
257 if (stage == RECOVERY_STAGE_ALLOCATE) {
258 PageInputStream in = new PageInputStream(store, logKey, firstTrunkPage, firstDataPage);
259 usedLogPages = in.allocateAllPages();
260 in.close();
261 return true;
262 }
263 PageInputStream pageIn = new PageInputStream(store, logKey, firstTrunkPage, firstDataPage);
264 DataReader in = new DataReader(pageIn);
265 int logId = 0;
266 Data data = store.createData();
267 boolean isEmpty = true;
268 try {
269 int pos = 0;
270 while (true) {
271 int x = in.readByte();
272 if (x < 0) {
273 break;
274 }
275 pos++;
276 isEmpty = false;
277 if (x == UNDO) {
278 int pageId = in.readVarInt();
279 int size = in.readVarInt();
280 if (size == 0) {
281 in.readFully(data.getBytes(), 0, store.getPageSize());
282 } else if (size == 1) {
283 // empty
284 Arrays.fill(data.getBytes(), 0, store.getPageSize(), (byte) 0);
285 } else {
286 in.readFully(compressBuffer, 0, size);
287 try {
288 compress.expand(compressBuffer, 0, size, data.getBytes(), 0, store.getPageSize());
289 } catch (ArrayIndexOutOfBoundsException e) {
290 DbException.convertToIOException(e);
291 }
292 }
293 if (stage == RECOVERY_STAGE_UNDO) {
294 if (!undo.get(pageId)) {
295 if (trace.isDebugEnabled()) {
296 trace.debug("log undo {0}", pageId);
297 }
298 store.writePage(pageId, data);
299 undo.set(pageId);
300 undoAll.set(pageId);
301 } else {
302 if (trace.isDebugEnabled()) {
303 trace.debug("log undo skip {0}", pageId);
304 }
305 }
306 }
307 } else if (x == ADD) {
308 int sessionId = in.readVarInt();
309 int tableId = in.readVarInt();
310 Row row = readRow(in, data);
311 if (stage == RECOVERY_STAGE_UNDO) {
312 store.allocateIfIndexRoot(pos, tableId, row);
313 } else if (stage == RECOVERY_STAGE_REDO) {
314 if (isSessionCommitted(sessionId, logId, pos)) {
315 if (trace.isDebugEnabled()) {
316 trace.debug("log redo + table: " + tableId + " s: " + sessionId + " " + row);
317 }
318 store.redo(tableId, row, true);
319 } else {
320 if (trace.isDebugEnabled()) {
321 trace.debug("log ignore s: " + sessionId + " + table: " + tableId + " " + row);
322 }
323 }
324 }
325 } else if (x == REMOVE) {
326 int sessionId = in.readVarInt();
327 int tableId = in.readVarInt();
328 long key = in.readVarLong();
329 if (stage == RECOVERY_STAGE_REDO) {
330 if (isSessionCommitted(sessionId, logId, pos)) {
331 if (trace.isDebugEnabled()) {
332 trace.debug("log redo - table: " + tableId + " s:" + sessionId + " key: " + key);
333 }
334 store.redoDelete(tableId, key);
335 } else {
336 if (trace.isDebugEnabled()) {
337 trace.debug("log ignore s: " + sessionId + " - table: " + tableId + " " + key);
338 }
339 }
340 }
341 } else if (x == TRUNCATE) {
342 int sessionId = in.readVarInt();
343 int tableId = in.readVarInt();
344 if (stage == RECOVERY_STAGE_REDO) {
345 if (isSessionCommitted(sessionId, logId, pos)) {
346 if (trace.isDebugEnabled()) {
347 trace.debug("log redo truncate table: " + tableId);
348 }
349 store.redoTruncate(tableId);
350 } else {
351 if (trace.isDebugEnabled()) {
352 trace.debug("log ignore s: "+ sessionId + " truncate table: " + tableId);
353 }
354 }
355 }
356 } else if (x == PREPARE_COMMIT) {
357 int sessionId = in.readVarInt();
358 String transaction = in.readString();
359 if (trace.isDebugEnabled()) {
360 trace.debug("log prepare commit " + sessionId + " " + transaction + " pos: " + pos);
361 }
362 if (stage == RECOVERY_STAGE_UNDO) {
363 int page = pageIn.getDataPage();
364 setPrepareCommit(sessionId, page, transaction);
365 }
366 } else if (x == ROLLBACK) {
367 int sessionId = in.readVarInt();
368 if (trace.isDebugEnabled()) {
369 trace.debug("log rollback " + sessionId + " pos: " + pos);
370 }
371 // ignore - this entry is just informational
372 } else if (x == COMMIT) {
373 int sessionId = in.readVarInt();
374 if (trace.isDebugEnabled()) {
375 trace.debug("log commit " + sessionId + " pos: " + pos);
376 }
377 if (stage == RECOVERY_STAGE_UNDO) {
378 setLastCommitForSession(sessionId, logId, pos);
379 }
380 } else if (x == NOOP) {
381 // nothing to do
382 } else if (x == CHECKPOINT) {
383 logId++;
384 } else if (x == FREE_LOG) {
385 int count = in.readVarInt();
386 for (int i = 0; i < count; i++) {
387 int pageId = in.readVarInt();
388 if (stage == RECOVERY_STAGE_REDO) {
389 if (!usedLogPages.get(pageId)) {
390 store.free(pageId, false);
391 }
392 }
393 }
394 } else {
395 if (trace.isDebugEnabled()) {
396 trace.debug("log end");
397 break;
398 }
399 }
400 }
401 } catch (DbException e) {
402 if (e.getErrorCode() == ErrorCode.FILE_CORRUPTED_1) {
403 trace.debug("log recovery stopped");
404 } else {
405 throw e;
406 }
407 } catch (IOException e) {
408 trace.debug("log recovery completed");
409 }
410 undo = new BitField();
411 if (stage == RECOVERY_STAGE_REDO) {
412 usedLogPages = null;
413 }
414 return isEmpty;
415 }
416
417 /**
418 * This method is called when a 'prepare commit' log entry is read when
419 * opening the database.
420 *
421 * @param sessionId the session id
422 * @param pageId the data page with the prepare entry
423 * @param transaction the transaction name, or null to rollback
424 */
425 private void setPrepareCommit(int sessionId, int pageId, String transaction) {
426 SessionState state = getOrAddSessionState(sessionId);
427 InDoubtTransaction doubt;
428 if (transaction == null) {
429 doubt = null;
430 } else {
431 doubt = new InDoubtTransaction(store, sessionId, pageId, transaction);
432 }
433 state.inDoubtTransaction = doubt;
434 }
435
436 /**
437 * Read a row from an input stream.
438 *
439 * @param in the input stream
440 * @param data a temporary buffer
441 * @return the row
442 */
443 public static Row readRow(DataReader in, Data data) throws IOException {
444 long key = in.readVarLong();
445 int len = in.readVarInt();
446 data.reset();
447 data.checkCapacity(len);
448 in.readFully(data.getBytes(), 0, len);
449 int columnCount = data.readVarInt();
450 Value[] values = new Value[columnCount];
451 for (int i = 0; i < columnCount; i++) {
452 values[i] = data.readValue();
453 }
454 Row row = new Row(values, Row.MEMORY_CALCULATE);
455 row.setKey(key);
456 return row;
457 }
458
459 /**
460 * Check if the undo entry was already written for the given page.
461 *
462 * @param pageId the page
463 * @return true if it was written
464 */
465 boolean getUndo(int pageId) {
466 return undo.get(pageId);
467 }
468
469 /**
470 * Add an undo entry to the log. The page data is only written once until
471 * the next checkpoint.
472 *
473 * @param pageId the page id
474 * @param page the old page data
475 */
476 void addUndo(int pageId, Data page) {
477 if (undo.get(pageId) || freeing) {
478 return;
479 }
480 if (trace.isDebugEnabled()) {
481 trace.debug("log undo " + pageId);
482 }
483 if (SysProperties.CHECK) {
484 if (page == null) {
485 DbException.throwInternalError("Undo entry not written");
486 }
487 }
488 undo.set(pageId);
489 undoAll.set(pageId);
490 Data buffer = getBuffer();
491 buffer.writeByte((byte) UNDO);
492 buffer.writeVarInt(pageId);
493 if (page.getBytes()[0] == 0) {
494 buffer.writeVarInt(1);
495 } else {
496 int pageSize = store.getPageSize();
497 if (COMPRESS_UNDO) {
498 int size = compress.compress(page.getBytes(), pageSize, compressBuffer, 0);
499 if (size < pageSize) {
500 buffer.writeVarInt(size);
501 buffer.checkCapacity(size);
502 buffer.write(compressBuffer, 0, size);
503 } else {
504 buffer.writeVarInt(0);
505 buffer.checkCapacity(pageSize);
506 buffer.write(page.getBytes(), 0, pageSize);
507 }
508 } else {
509 buffer.writeVarInt(0);
510 buffer.checkCapacity(pageSize);
511 buffer.write(page.getBytes(), 0, pageSize);
512 }
513 }
514 write(buffer);
515 }
516
517 private void freeLogPages(IntArray pages) {
518 if (trace.isDebugEnabled()) {
519 trace.debug("log frees " + pages.get(0) + ".." + pages.get(pages.size() - 1));
520 }
521 Data buffer = getBuffer();
522 buffer.writeByte((byte) FREE_LOG);
523 int size = pages.size();
524 buffer.writeVarInt(size);
525 for (int i = 0; i < size; i++) {
526 buffer.writeVarInt(pages.get(i));
527 }
528 write(buffer);
529 }
530
531 private void write(Data data) {
532 pageOut.write(data.getBytes(), 0, data.length());
533 data.reset();
534 }
535
536 /**
537 * Mark a transaction as committed.
538 *
539 * @param sessionId the session
540 */
541 void commit(int sessionId) {
542 if (trace.isDebugEnabled()) {
543 trace.debug("log commit s: " + sessionId);
544 }
545 if (store.getDatabase().getPageStore() == null) {
546 // database already closed
547 return;
548 }
549 Data buffer = getBuffer();
550 buffer.writeByte((byte) COMMIT);
551 buffer.writeVarInt(sessionId);
552 write(buffer);
553 if (store.getDatabase().getFlushOnEachCommit()) {
554 flush();
555 }
556 }
557
558 /**
559 * Prepare a transaction.
560 *
561 * @param session the session
562 * @param transaction the name of the transaction
563 */
564 void prepareCommit(Session session, String transaction) {
565 if (trace.isDebugEnabled()) {
566 trace.debug("log prepare commit s: " + session.getId() + ", " + transaction);
567 }
568 if (store.getDatabase().getPageStore() == null) {
569 // database already closed
570 return;
571 }
572 // store it on a separate log page
573 int pageSize = store.getPageSize();
574 pageOut.flush();
575 pageOut.fillPage();
576 Data buffer = getBuffer();
577 buffer.writeByte((byte) PREPARE_COMMIT);
578 buffer.writeVarInt(session.getId());
579 buffer.writeString(transaction);
580 if (buffer.length() >= PageStreamData.getCapacity(pageSize)) {
581 throw DbException.getInvalidValueException("transaction name (too long)", transaction);
582 }
583 write(buffer);
584 // store it on a separate log page
585 flushOut();
586 pageOut.fillPage();
587 if (store.getDatabase().getFlushOnEachCommit()) {
588 flush();
589 }
590 }
591
592 /**
593 * A record is added to a table, or removed from a table.
594 *
595 * @param session the session
596 * @param tableId the table id
597 * @param row the row to add
598 * @param add true if the row is added, false if it is removed
599 */
600 void logAddOrRemoveRow(Session session, int tableId, Row row, boolean add) {
601 if (trace.isDebugEnabled()) {
602 trace.debug("log " + (add ? "+" : "-") + " s: " + session.getId() + " table: " + tableId + " row: " + row);
603 }
604 session.addLogPos(logSectionId, logPos);
605 logPos++;
606 Data data = dataBuffer;
607 data.reset();
608 int columns = row.getColumnCount();
609 data.writeVarInt(columns);
610 data.checkCapacity(row.getByteCount(data));
611 if (session.isRedoLogBinaryEnabled()) {
612 for (int i = 0; i < columns; i++) {
613 data.writeValue(row.getValue(i));
614 }
615 } else {
616 for (int i = 0; i < columns; i++) {
617 Value v = row.getValue(i);
618 if (v.getType() == Value.BYTES) {
619 data.writeValue(ValueNull.INSTANCE);
620 } else {
621 data.writeValue(v);
622 }
623 }
624 }
625 Data buffer = getBuffer();
626 buffer.writeByte((byte) (add ? ADD : REMOVE));
627 buffer.writeVarInt(session.getId());
628 buffer.writeVarInt(tableId);
629 buffer.writeVarLong(row.getKey());
630 if (add) {
631 buffer.writeVarInt(data.length());
632 buffer.checkCapacity(data.length());
633 buffer.write(data.getBytes(), 0, data.length());
634 }
635 write(buffer);
636 }
637
638 /**
639 * A table is truncated.
640 *
641 * @param session the session
642 * @param tableId the table id
643 */
644 void logTruncate(Session session, int tableId) {
645 if (trace.isDebugEnabled()) {
646 trace.debug("log truncate s: " + session.getId() + " table: " + tableId);
647 }
648 session.addLogPos(logSectionId, logPos);
649 logPos++;
650 Data buffer = getBuffer();
651 buffer.writeByte((byte) TRUNCATE);
652 buffer.writeVarInt(session.getId());
653 buffer.writeVarInt(tableId);
654 write(buffer);
655 }
656
657 /**
658 * Flush the transaction log.
659 */
660 void flush() {
661 if (pageOut != null) {
662 flushOut();
663 }
664 }
665
666 /**
667 * Switch to a new log section.
668 */
669 void checkpoint() {
670 Data buffer = getBuffer();
671 buffer.writeByte((byte) CHECKPOINT);
672 write(buffer);
673 undo = new BitField();
674 logSectionId++;
675 logPos = 0;
676 pageOut.flush();
677 pageOut.fillPage();
678 int currentDataPage = pageOut.getCurrentDataPageId();
679 logSectionPageMap.put(logSectionId, currentDataPage);
680 }
681
682 int getLogSectionId() {
683 return logSectionId;
684 }
685
686 int getLogFirstSectionId() {
687 return firstSectionId;
688 }
689
690 long getLogPos() {
691 return logPos;
692 }
693
694 /**
695 * Remove all pages until the given log (excluding).
696 *
697 * @param firstUncommittedSection the first log section to keep
698 */
699 void removeUntil(int firstUncommittedSection) {
700 if (firstUncommittedSection == 0) {
701 return;
702 }
703 int firstDataPageToKeep = logSectionPageMap.get(firstUncommittedSection);
704 firstTrunkPage = removeUntil(firstTrunkPage, firstDataPageToKeep);
705 store.setLogFirstPage(logKey, firstTrunkPage, firstDataPageToKeep);
706 while (firstSectionId < firstUncommittedSection) {
707 if (firstSectionId > 0) {
708 // there is no entry for log 0
709 logSectionPageMap.remove(firstSectionId);
710 }
711 firstSectionId++;
712 }
713 }
714
715 /**
716 * Remove all pages until the given data page.
717 *
718 * @param trunkPage the first trunk page
719 * @param firstDataPageToKeep the first data page to keep
720 * @return the trunk page of the data page to keep
721 */
722 private int removeUntil(int trunkPage, int firstDataPageToKeep) {
723 trace.debug("log.removeUntil " + trunkPage + " " + firstDataPageToKeep);
724 int last = trunkPage;
725 while (true) {
726 Page p = store.getPage(trunkPage);
727 PageStreamTrunk t = (PageStreamTrunk) p;
728 if (t == null) {
729 throw DbException.throwInternalError(
730 "log.removeUntil not found: " + firstDataPageToKeep + " last " + last);
731 }
732 logKey = t.getLogKey();
733 last = t.getPos();
734 if (t.contains(firstDataPageToKeep)) {
735 return last;
736 }
737 trunkPage = t.getNextTrunk();
738 IntArray list = new IntArray();
739 list.add(t.getPos());
740 for (int i = 0;; i++) {
741 int next = t.getPageData(i);
742 if (next == -1) {
743 break;
744 }
745 list.add(next);
746 }
747 freeLogPages(list);
748 pageOut.free(t);
749 }
750 }
751
752 /**
753 * Close without further writing.
754 */
755 void close() {
756 trace.debug("log close");
757 if (pageOut != null) {
758 pageOut.close();
759 pageOut = null;
760 }
761 writeBuffer = null;
762 }
763
764 /**
765 * Check if the session committed after than the given position.
766 *
767 * @param sessionId the session id
768 * @param logId the log id
769 * @param pos the position in the log
770 * @return true if it is committed
771 */
772 private boolean isSessionCommitted(int sessionId, int logId, int pos) {
773 SessionState state = sessionStates.get(sessionId);
774 if (state == null) {
775 return false;
776 }
777 return state.isCommitted(logId, pos);
778 }
779
780 /**
781 * Set the last commit record for a session.
782 *
783 * @param sessionId the session id
784 * @param logId the log id
785 * @param pos the position in the log
786 */
787 private void setLastCommitForSession(int sessionId, int logId, int pos) {
788 SessionState state = getOrAddSessionState(sessionId);
789 state.lastCommitLog = logId;
790 state.lastCommitPos = pos;
791 state.inDoubtTransaction = null;
792 }
793
794 /**
795 * Get the session state for this session. A new object is created if there
796 * is no session state yet.
797 *
798 * @param sessionId the session id
799 * @return the session state object
800 */
801 private SessionState getOrAddSessionState(int sessionId) {
802 Integer key = sessionId;
803 SessionState state = sessionStates.get(key);
804 if (state == null) {
805 state = new SessionState();
806 sessionStates.put(key, state);
807 state.sessionId = sessionId;
808 }
809 return state;
810 }
811
812 long getSize() {
813 return pageOut == null ? 0 : pageOut.getSize();
814 }
815
816 ArrayList<InDoubtTransaction> getInDoubtTransactions() {
817 ArrayList<InDoubtTransaction> list = New.arrayList();
818 for (SessionState state : sessionStates.values()) {
819 InDoubtTransaction in = state.inDoubtTransaction;
820 if (in != null) {
821 list.add(in);
822 }
823 }
824 return list;
825 }
826
827 /**
828 * Set the state of an in-doubt transaction.
829 *
830 * @param sessionId the session
831 * @param pageId the page where the commit was prepared
832 * @param commit whether the transaction should be committed
833 */
834 void setInDoubtTransactionState(int sessionId, int pageId, boolean commit) {
835 PageStreamData d = (PageStreamData) store.getPage(pageId);
836 d.initWrite();
837 Data buff = store.createData();
838 buff.writeByte((byte) (commit ? COMMIT : ROLLBACK));
839 buff.writeVarInt(sessionId);
840 byte[] bytes = buff.getBytes();
841 d.write(bytes, 0, bytes.length);
842 bytes = new byte[d.getRemaining()];
843 d.write(bytes, 0, bytes.length);
844 d.write();
845 }
846
847 /**
848 * Called after the recovery has been completed.
849 */
850 void recoverEnd() {
851 sessionStates = New.hashMap();
852 }
853
854 private void flushOut() {
855 pageOut.flush();
856 }
857
858 private Data getBuffer() {
859 if (writeBuffer.length() == 0) {
860 return writeBuffer;
861 }
862 return store.createData();
863 }
864
865
866 /**
867 * Get the smallest possible page id used. This is the trunk page if only
868 * appending at the end of the file, or 0.
869 *
870 * @return the smallest possible page.
871 */
872 int getMinPageId() {
873 return pageOut == null ? 0 : pageOut.getMinPageId();
874 }
875
876 }
877