File D:\maven-qcr-plugin-testsunitaires\qcr4java\h2-1.3.170-sources\target\site\qc\lecode\store\LobStorage.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.BufferedInputStream;
10 import java.io.IOException;
11 import java.io.InputStream;
12 import java.io.Reader;
13 import java.sql.Connection;
14 import java.sql.PreparedStatement;
15 import java.sql.ResultSet;
16 import java.sql.SQLException;
17 import java.sql.Statement;
18 import java.util.ArrayList;
19 import java.util.Arrays;
20 import java.util.HashMap;
21 import org.h2.constant.ErrorCode;
22 import org.h2.constant.SysProperties;
23 import org.h2.engine.Constants;
24 import org.h2.message.DbException;
25 import org.h2.tools.CompressTool;
26 import org.h2.util.IOUtils;
27 import org.h2.util.MathUtils;
28 import org.h2.util.New;
29 import org.h2.util.StringUtils;
30 import org.h2.util.Utils;
31 import org.h2.value.Value;
32 import org.h2.value.ValueLob;
33 import org.h2.value.ValueLobDb;
34
35 /**
36 * This class stores LOB objects in the database.
37 */
38 public class LobStorage {
39
40 /**
41 * The table id for session variables (LOBs not assigned to a table).
42 */
43 public static final int TABLE_ID_SESSION_VARIABLE = -1;
44
45 /**
46 * The table id for temporary objects (not assigned to any object).
47 */
48 public static final int TABLE_TEMP = -2;
49
50 /**
51 * The name of the lob data table. If this table exists, then lob storage is
52 * used.
53 */
54 public static final String LOB_DATA_TABLE = "LOB_DATA";
55
56 private static final String LOB_SCHEMA = "INFORMATION_SCHEMA";
57
58 private static final String LOBS = LOB_SCHEMA + ".LOBS";
59 private static final String LOB_MAP = LOB_SCHEMA + ".LOB_MAP";
60 private static final String LOB_DATA = LOB_SCHEMA + "." + LOB_DATA_TABLE;
61
62 private static final int BLOCK_LENGTH = 20000;
63
64 /**
65 * The size of cache for lob block hashes. Each entry needs 2 longs (16
66 * bytes), therefore, the size 4096 means 64 KB.
67 */
68 private static final int HASH_CACHE_SIZE = 4 * 1024;
69 private Connection conn;
70 private final HashMap<String, PreparedStatement> prepared = New.hashMap();
71 private long nextBlock;
72 private final CompressTool compress = CompressTool.getInstance();
73 private long[] hashBlocks;
74
75 private final DataHandler handler;
76 private boolean init;
77
78 public LobStorage(DataHandler handler) {
79 this.handler = handler;
80 }
81
82 /**
83 * Initialize the lob storage.
84 */
85 public void init() {
86 if (init) {
87 return;
88 }
89 synchronized (handler) {
90 conn = handler.getLobConnection();
91 init = true;
92 if (conn == null) {
93 return;
94 }
95 try {
96 Statement stat = conn.createStatement();
97 // stat.execute("SET UNDO_LOG 0");
98 // stat.execute("SET REDO_LOG_BINARY 0");
99 boolean create = true;
100 PreparedStatement prep = conn.prepareStatement(
101 "SELECT ZERO() FROM INFORMATION_SCHEMA.COLUMNS WHERE " +
102 "TABLE_SCHEMA=? AND TABLE_NAME=? AND COLUMN_NAME=?");
103 prep.setString(1, "INFORMATION_SCHEMA");
104 prep.setString(2, "LOB_MAP");
105 prep.setString(3, "POS");
106 ResultSet rs;
107 rs = prep.executeQuery();
108 if (rs.next()) {
109 prep = conn.prepareStatement(
110 "SELECT ZERO() FROM INFORMATION_SCHEMA.TABLES WHERE " +
111 "TABLE_SCHEMA=? AND TABLE_NAME=?");
112 prep.setString(1, "INFORMATION_SCHEMA");
113 prep.setString(2, "LOB_DATA");
114 rs = prep.executeQuery();
115 if (rs.next()) {
116 create = false;
117 }
118 }
119 if (create) {
120 stat.execute("CREATE CACHED TABLE IF NOT EXISTS " + LOBS +
121 "(ID BIGINT PRIMARY KEY, BYTE_COUNT BIGINT, TABLE INT) HIDDEN");
122 stat.execute("CREATE INDEX IF NOT EXISTS " +
123 "INFORMATION_SCHEMA.INDEX_LOB_TABLE ON " + LOBS + "(TABLE)");
124 stat.execute("CREATE CACHED TABLE IF NOT EXISTS " + LOB_MAP +
125 "(LOB BIGINT, SEQ INT, POS BIGINT, HASH INT, BLOCK BIGINT, PRIMARY KEY(LOB, SEQ)) HIDDEN");
126 // TODO the column name OFFSET was used in version 1.3.156,
127 // so this can be remove in a later version
128 stat.execute("ALTER TABLE " + LOB_MAP + " RENAME TO " + LOB_MAP + " HIDDEN");
129 stat.execute("ALTER TABLE " + LOB_MAP + " ADD IF NOT EXISTS POS BIGINT BEFORE HASH");
130 stat.execute("ALTER TABLE " + LOB_MAP + " DROP COLUMN IF EXISTS \"OFFSET\"");
131 stat.execute("CREATE INDEX IF NOT EXISTS " +
132 "INFORMATION_SCHEMA.INDEX_LOB_MAP_DATA_LOB ON " + LOB_MAP + "(BLOCK, LOB)");
133 stat.execute("CREATE CACHED TABLE IF NOT EXISTS " + LOB_DATA +
134 "(BLOCK BIGINT PRIMARY KEY, COMPRESSED INT, DATA BINARY) HIDDEN");
135 }
136 rs = stat.executeQuery("SELECT MAX(BLOCK) FROM " + LOB_DATA);
137 rs.next();
138 nextBlock = rs.getLong(1) + 1;
139 stat.close();
140 } catch (SQLException e) {
141 throw DbException.convert(e);
142 }
143 }
144 }
145
146 private long getNextLobId() throws SQLException {
147 String sql = "SELECT MAX(LOB) FROM " + LOB_MAP;
148 PreparedStatement prep = prepare(sql);
149 ResultSet rs = prep.executeQuery();
150 rs.next();
151 long x = rs.getLong(1) + 1;
152 reuse(sql, prep);
153 sql = "SELECT MAX(ID) FROM " + LOBS;
154 prep = prepare(sql);
155 rs = prep.executeQuery();
156 rs.next();
157 x = Math.max(x, rs.getLong(1) + 1);
158 reuse(sql, prep);
159 return x;
160 }
161
162 /**
163 * Remove all LOBs for this table.
164 *
165 * @param tableId the table id
166 */
167 public void removeAllForTable(int tableId) {
168 if (SysProperties.LOB_IN_DATABASE) {
169 init();
170 try {
171 String sql = "SELECT ID FROM " + LOBS + " WHERE TABLE = ?";
172 PreparedStatement prep = prepare(sql);
173 prep.setInt(1, tableId);
174 ResultSet rs = prep.executeQuery();
175 while (rs.next()) {
176 removeLob(rs.getLong(1));
177 }
178 reuse(sql, prep);
179 } catch (SQLException e) {
180 throw DbException.convert(e);
181 }
182 if (tableId == TABLE_ID_SESSION_VARIABLE) {
183 removeAllForTable(TABLE_TEMP);
184 }
185 // remove both lobs in the database as well as in the file system
186 // (compatibility)
187 }
188 ValueLob.removeAllForTable(handler, tableId);
189 }
190
191 /**
192 * Create a LOB object that fits in memory.
193 *
194 * @param type the value type
195 * @param small the byte array
196 * @return the LOB
197 */
198 public static Value createSmallLob(int type, byte[] small) {
199 if (SysProperties.LOB_IN_DATABASE) {
200 int precision;
201 if (type == Value.CLOB) {
202 precision = StringUtils.utf8Decode(small).length();
203 } else {
204 precision = small.length;
205 }
206 return ValueLobDb.createSmallLob(type, small, precision);
207 }
208 return ValueLob.createSmallLob(type, small);
209 }
210
211 /**
212 * Read a block of data from the given LOB.
213 *
214 * @param lob the lob id
215 * @param seq the block sequence number
216 * @return the block (expanded if stored compressed)
217 */
218 byte[] readBlock(long lob, int seq) throws SQLException {
219 synchronized (handler) {
220 String sql = "SELECT COMPRESSED, DATA FROM " + LOB_MAP + " M " +
221 "INNER JOIN " + LOB_DATA + " D ON M.BLOCK = D.BLOCK " +
222 "WHERE M.LOB = ? AND M.SEQ = ?";
223 PreparedStatement prep = prepare(sql);
224 prep.setLong(1, lob);
225 prep.setInt(2, seq);
226 ResultSet rs = prep.executeQuery();
227 if (!rs.next()) {
228 throw DbException.get(ErrorCode.IO_EXCEPTION_1, "Missing lob entry: "+ lob + "/" + seq).getSQLException();
229 }
230 int compressed = rs.getInt(1);
231 byte[] buffer = rs.getBytes(2);
232 if (compressed != 0) {
233 buffer = compress.expand(buffer);
234 }
235 reuse(sql, prep);
236 return buffer;
237 }
238 }
239
240 /**
241 * Retrieve the sequence id and position that is smaller than the requested
242 * position. Those values can be used to quickly skip to a given position
243 * without having to read all data.
244 *
245 * @param lob the lob
246 * @param pos the required position
247 * @return null if the data is not available, or an array of two elements:
248 * the sequence, and the offset
249 */
250 long[] skipBuffer(long lob, long pos) throws SQLException {
251 synchronized (handler) {
252 String sql = "SELECT MAX(SEQ), MAX(POS) FROM " + LOB_MAP +
253 " WHERE LOB = ? AND POS < ?";
254 PreparedStatement prep = prepare(sql);
255 prep.setLong(1, lob);
256 prep.setLong(2, pos);
257 ResultSet rs = prep.executeQuery();
258 rs.next();
259 int seq = rs.getInt(1);
260 pos = rs.getLong(2);
261 boolean wasNull = rs.wasNull();
262 rs.close();
263 reuse(sql, prep);
264 // upgraded: offset not set
265 return wasNull ? null : new long[]{seq, pos};
266 }
267 }
268
269 /**
270 * An input stream that reads from a remote LOB.
271 */
272 public static class RemoteInputStream extends InputStream {
273
274 /**
275 * The data handler.
276 */
277 private final DataHandler handler;
278
279 /**
280 * The lob id.
281 */
282 private final long lob;
283
284 private final byte[] hmac;
285
286 /**
287 * The position.
288 */
289 private long pos;
290
291 /**
292 * The remaining bytes in the lob.
293 */
294 private long remainingBytes;
295
296 public RemoteInputStream(DataHandler handler, long lob, byte[] hmac, long byteCount) {
297 this.handler = handler;
298 this.lob = lob;
299 this.hmac = hmac;
300 remainingBytes = byteCount;
301 }
302
303 public int read() throws IOException {
304 byte[] buff = new byte[1];
305 int len = read(buff, 0, 1);
306 return len < 0 ? len : (buff[0] & 255);
307 }
308
309 public int read(byte[] buff) throws IOException {
310 return read(buff, 0, buff.length);
311 }
312
313 public int read(byte[] buff, int off, int length) throws IOException {
314 if (length == 0) {
315 return 0;
316 }
317 length = (int) Math.min(length, remainingBytes);
318 if (length == 0) {
319 return -1;
320 }
321 length = handler.readLob(lob, hmac, pos, buff, off, length);
322 remainingBytes -= length;
323 if (length == 0) {
324 return -1;
325 }
326 pos += length;
327 return length;
328 }
329
330 public long skip(long n) {
331 remainingBytes -= n;
332 pos += n;
333 return n;
334 }
335
336 }
337
338 /**
339 * An input stream that reads from a LOB.
340 */
341 public class LobInputStream extends InputStream {
342
343 /**
344 * The size of the lob.
345 */
346 private final long length;
347
348 /**
349 * The remaining bytes in the lob.
350 */
351 private long remainingBytes;
352
353 /**
354 * The temporary buffer.
355 */
356 private byte[] buffer;
357
358 /**
359 * The position within the buffer.
360 */
361 private int pos;
362
363 /**
364 * The lob id.
365 */
366 private final long lob;
367
368 /**
369 * The lob sequence id.
370 */
371 private int seq;
372
373 public LobInputStream(long lob, long byteCount) {
374 this.lob = lob;
375 remainingBytes = byteCount;
376 this.length = byteCount;
377 }
378
379 public int read() throws IOException {
380 fillBuffer();
381 if (remainingBytes <= 0) {
382 return -1;
383 }
384 remainingBytes--;
385 return buffer[pos++] & 255;
386 }
387
388 public long skip(long n) throws IOException {
389 long remaining = n;
390 remaining -= skipSmall(remaining);
391 if (remaining > BLOCK_LENGTH) {
392 long toPos = length - remainingBytes + remaining;
393 try {
394 long[] seqPos = skipBuffer(lob, toPos);
395 if (seqPos == null) {
396 remaining -= super.skip(remaining);
397 return n - remaining;
398 }
399 seq = (int) seqPos[0];
400 long p = seqPos[1];
401 remainingBytes = length - p;
402 remaining = toPos - p;
403 } catch (SQLException e) {
404 throw DbException.convertToIOException(e);
405 }
406 pos = 0;
407 buffer = null;
408 }
409 fillBuffer();
410 remaining -= skipSmall(remaining);
411 remaining -= super.skip(remaining);
412 return n - remaining;
413 }
414
415 private int skipSmall(long n) {
416 if (n > 0 && buffer != null && pos < buffer.length) {
417 int x = MathUtils.convertLongToInt(Math.min(n, buffer.length - pos));
418 pos += x;
419 remainingBytes -= x;
420 return x;
421 }
422 return 0;
423 }
424
425 public int read(byte[] buff) throws IOException {
426 return readFully(buff, 0, buff.length);
427 }
428
429 public int read(byte[] buff, int off, int length) throws IOException {
430 return readFully(buff, off, length);
431 }
432
433 private int readFully(byte[] buff, int off, int length) throws IOException {
434 if (length == 0) {
435 return 0;
436 }
437 int read = 0;
438 while (length > 0) {
439 fillBuffer();
440 if (remainingBytes <= 0) {
441 break;
442 }
443 int len = (int) Math.min(length, remainingBytes);
444 len = Math.min(len, buffer.length - pos);
445 System.arraycopy(buffer, pos, buff, off, len);
446 pos += len;
447 read += len;
448 remainingBytes -= len;
449 off += len;
450 length -= len;
451 }
452 return read == 0 ? -1 : read;
453 }
454
455 private void fillBuffer() throws IOException {
456 if (buffer != null && pos < buffer.length) {
457 return;
458 }
459 if (remainingBytes <= 0) {
460 return;
461 }
462 try {
463 buffer = readBlock(lob, seq++);
464 pos = 0;
465 } catch (SQLException e) {
466 throw DbException.convertToIOException(e);
467 }
468 }
469
470 }
471
472 private PreparedStatement prepare(String sql) throws SQLException {
473 if (SysProperties.CHECK2) {
474 if (!Thread.holdsLock(handler)) {
475 throw DbException.throwInternalError();
476 }
477 }
478 PreparedStatement prep = prepared.remove(sql);
479 if (prep == null) {
480 prep = conn.prepareStatement(sql);
481 }
482 return prep;
483 }
484
485 private void reuse(String sql, PreparedStatement prep) {
486 if (SysProperties.CHECK2) {
487 if (!Thread.holdsLock(handler)) {
488 throw DbException.throwInternalError();
489 }
490 }
491 prepared.put(sql, prep);
492 }
493
494 /**
495 * Delete a LOB from the database.
496 *
497 * @param lob the lob id
498 */
499 public void removeLob(long lob) {
500 try {
501 synchronized (handler) {
502 if (conn == null) {
503 return;
504 }
505 String sql = "SELECT BLOCK, HASH FROM " + LOB_MAP + " D WHERE D.LOB = ? " +
506 "AND NOT EXISTS(SELECT 1 FROM " + LOB_MAP + " O " +
507 "WHERE O.BLOCK = D.BLOCK AND O.LOB <> ?)";
508 PreparedStatement prep = prepare(sql);
509 prep.setLong(1, lob);
510 prep.setLong(2, lob);
511 ResultSet rs = prep.executeQuery();
512 ArrayList<Long> blocks = New.arrayList();
513 while (rs.next()) {
514 blocks.add(rs.getLong(1));
515 int hash = rs.getInt(2);
516 setHashCacheBlock(hash, -1);
517 }
518 reuse(sql, prep);
519
520 sql = "DELETE FROM " + LOB_MAP + " WHERE LOB = ?";
521 prep = prepare(sql);
522 prep.setLong(1, lob);
523 prep.execute();
524 reuse(sql, prep);
525
526 sql = "DELETE FROM " + LOB_DATA + " WHERE BLOCK = ?";
527 prep = prepare(sql);
528 for (long block : blocks) {
529 prep.setLong(1, block);
530 prep.execute();
531 }
532 reuse(sql, prep);
533
534 sql = "DELETE FROM " + LOBS + " WHERE ID = ?";
535 prep = prepare(sql);
536 prep.setLong(1, lob);
537 prep.execute();
538 reuse(sql, prep);
539 }
540 } catch (SQLException e) {
541 throw DbException.convert(e);
542 }
543 }
544
545 /**
546 * Get the input stream for the given lob.
547 *
548 * @param lobId the lob id
549 * @param hmac the message authentication code (for remote input streams)
550 * @param byteCount the number of bytes to read, or -1 if not known
551 * @return the stream
552 */
553 public InputStream getInputStream(long lobId, byte[] hmac, long byteCount) throws IOException {
554 init();
555 if (conn == null) {
556 if (byteCount < 0) {
557 byteCount = Long.MAX_VALUE;
558 }
559 return new BufferedInputStream(new RemoteInputStream(handler, lobId, hmac, byteCount));
560 }
561 if (byteCount == -1) {
562 synchronized (handler) {
563 try {
564 String sql = "SELECT BYTE_COUNT FROM " + LOBS + " WHERE ID = ?";
565 PreparedStatement prep = prepare(sql);
566 prep.setLong(1, lobId);
567 ResultSet rs = prep.executeQuery();
568 if (!rs.next()) {
569 throw DbException.get(ErrorCode.IO_EXCEPTION_1, "Missing lob: "+ lobId).getSQLException();
570 }
571 byteCount = rs.getLong(1);
572 reuse(sql, prep);
573 } catch (SQLException e) {
574 throw DbException.convertToIOException(e);
575 }
576 }
577 }
578 return new LobInputStream(lobId, byteCount);
579 }
580
581 private ValueLobDb addLob(InputStream in, long maxLength, int type) {
582 try {
583 byte[] buff = new byte[BLOCK_LENGTH];
584 if (maxLength < 0) {
585 maxLength = Long.MAX_VALUE;
586 }
587 long length = 0;
588 long lobId = -1;
589 int maxLengthInPlaceLob = handler.getMaxLengthInplaceLob();
590 String compressAlgorithm = handler.getLobCompressionAlgorithm(type);
591 try {
592 byte[] small = null;
593 for (int seq = 0; maxLength > 0; seq++) {
594 int len = (int) Math.min(BLOCK_LENGTH, maxLength);
595 len = IOUtils.readFully(in, buff, 0, len);
596 if (len <= 0) {
597 break;
598 }
599 maxLength -= len;
600 byte[] b;
601 if (len != buff.length) {
602 b = new byte[len];
603 System.arraycopy(buff, 0, b, 0, len);
604 } else {
605 b = buff;
606 }
607 if (seq == 0 && b.length < BLOCK_LENGTH && b.length <= maxLengthInPlaceLob) {
608 small = b;
609 break;
610 }
611 synchronized (handler) {
612 if (seq == 0) {
613 lobId = getNextLobId();
614 }
615 storeBlock(lobId, seq, length, b, compressAlgorithm);
616 }
617 length += len;
618 }
619 if (lobId == -1 && small == null) {
620 // zero length
621 small = new byte[0];
622 }
623 if (small != null) {
624 // CLOB: the precision will be fixed later
625 ValueLobDb v = ValueLobDb.createSmallLob(type, small, small.length);
626 return v;
627 }
628 return registerLob(type, lobId, TABLE_TEMP, length);
629 } catch (IOException e) {
630 if (lobId != -1) {
631 removeLob(lobId);
632 }
633 throw DbException.convertIOException(e, null);
634 }
635 } catch (SQLException e) {
636 throw DbException.convert(e);
637 }
638 }
639
640 private ValueLobDb registerLob(int type, long lobId, int tableId, long byteCount) {
641 synchronized (handler) {
642 try {
643 String sql = "INSERT INTO " + LOBS + "(ID, BYTE_COUNT, TABLE) VALUES(?, ?, ?)";
644 PreparedStatement prep = prepare(sql);
645 prep.setLong(1, lobId);
646 prep.setLong(2, byteCount);
647 prep.setInt(3, tableId);
648 prep.execute();
649 reuse(sql, prep);
650 ValueLobDb v = ValueLobDb.create(type, this, tableId, lobId, null, byteCount);
651 return v;
652 } catch (SQLException e) {
653 throw DbException.convert(e);
654 }
655 }
656 }
657
658 /**
659 * Copy a lob.
660 *
661 * @param type the type
662 * @param oldLobId the old lob id
663 * @param tableId the new table id
664 * @param length the length
665 * @return the new lob
666 */
667 public ValueLobDb copyLob(int type, long oldLobId, int tableId, long length) {
668 synchronized (handler) {
669 try {
670 init();
671 long lobId = getNextLobId();
672 String sql = "INSERT INTO " + LOB_MAP + "(LOB, SEQ, POS, HASH, BLOCK) " +
673 "SELECT ?, SEQ, POS, HASH, BLOCK FROM " + LOB_MAP + " WHERE LOB = ?";
674 PreparedStatement prep = prepare(sql);
675 prep.setLong(1, lobId);
676 prep.setLong(2, oldLobId);
677 prep.executeUpdate();
678 reuse(sql, prep);
679
680 sql = "INSERT INTO " + LOBS + "(ID, BYTE_COUNT, TABLE) " +
681 "SELECT ?, BYTE_COUNT, ? FROM " + LOBS + " WHERE ID = ?";
682 prep = prepare(sql);
683 prep.setLong(1, lobId);
684 prep.setLong(2, tableId);
685 prep.setLong(3, oldLobId);
686 prep.executeUpdate();
687 reuse(sql, prep);
688
689 ValueLobDb v = ValueLobDb.create(type, this, tableId, lobId, null, length);
690 return v;
691 } catch (SQLException e) {
692 throw DbException.convert(e);
693 }
694 }
695 }
696
697 private long getHashCacheBlock(int hash) {
698 if (HASH_CACHE_SIZE > 0) {
699 initHashCache();
700 int index = hash & (HASH_CACHE_SIZE - 1);
701 long oldHash = hashBlocks[index];
702 if (oldHash == hash) {
703 return hashBlocks[index + HASH_CACHE_SIZE];
704 }
705 }
706 return -1;
707 }
708
709 private void setHashCacheBlock(int hash, long block) {
710 if (HASH_CACHE_SIZE > 0) {
711 initHashCache();
712 int index = hash & (HASH_CACHE_SIZE - 1);
713 hashBlocks[index] = hash;
714 hashBlocks[index + HASH_CACHE_SIZE] = block;
715 }
716 }
717
718 private void initHashCache() {
719 if (hashBlocks == null) {
720 hashBlocks = new long[HASH_CACHE_SIZE * 2];
721 }
722 }
723
724 /**
725 * Store a block in the LOB storage.
726 *
727 * @param lobId the lob id
728 * @param seq the sequence number
729 * @param pos the position within the lob
730 * @param b the data
731 * @param compressAlgorithm the compression algorithm (may be null)
732 */
733 void storeBlock(long lobId, int seq, long pos, byte[] b, String compressAlgorithm) throws SQLException {
734 long block;
735 boolean blockExists = false;
736 if (compressAlgorithm != null) {
737 b = compress.compress(b, compressAlgorithm);
738 }
739 int hash = Arrays.hashCode(b);
740 synchronized (handler) {
741 block = getHashCacheBlock(hash);
742 if (block != -1) {
743 String sql = "SELECT COMPRESSED, DATA FROM " + LOB_DATA +
744 " WHERE BLOCK = ?";
745 PreparedStatement prep = prepare(sql);
746 prep.setLong(1, block);
747 ResultSet rs = prep.executeQuery();
748 if (rs.next()) {
749 boolean compressed = rs.getInt(1) != 0;
750 byte[] compare = rs.getBytes(2);
751 if (compressed == (compressAlgorithm != null) && Arrays.equals(b, compare)) {
752 blockExists = true;
753 }
754 }
755 reuse(sql, prep);
756 }
757 if (!blockExists) {
758 block = nextBlock++;
759 setHashCacheBlock(hash, block);
760 String sql = "INSERT INTO " + LOB_DATA + "(BLOCK, COMPRESSED, DATA) VALUES(?, ?, ?)";
761 PreparedStatement prep = prepare(sql);
762 prep.setLong(1, block);
763 prep.setInt(2, compressAlgorithm == null ? 0 : 1);
764 prep.setBytes(3, b);
765 prep.execute();
766 reuse(sql, prep);
767 }
768 String sql = "INSERT INTO " + LOB_MAP + "(LOB, SEQ, POS, HASH, BLOCK) VALUES(?, ?, ?, ?, ?)";
769 PreparedStatement prep = prepare(sql);
770 prep.setLong(1, lobId);
771 prep.setInt(2, seq);
772 prep.setLong(3, pos);
773 prep.setLong(4, hash);
774 prep.setLong(5, block);
775 prep.execute();
776 reuse(sql, prep);
777 }
778 }
779
780 /**
781 * An input stream that reads the data from a reader.
782 */
783 static class CountingReaderInputStream extends InputStream {
784
785 private final Reader reader;
786 private long length;
787 private long remaining;
788 private int pos;
789 private final char[] charBuffer = new char[Constants.IO_BUFFER_SIZE];
790 private byte[] buffer;
791
792 CountingReaderInputStream(Reader reader, long maxLength) {
793 this.reader = reader;
794 this.remaining = maxLength;
795 buffer = Utils.EMPTY_BYTES;
796 }
797
798 public int read(byte[] buff, int offset, int len) throws IOException {
799 if (buffer == null) {
800 return -1;
801 }
802 if (pos >= buffer.length) {
803 fillBuffer();
804 if (buffer == null) {
805 return -1;
806 }
807 }
808 len = Math.min(len, buffer.length - pos);
809 System.arraycopy(buffer, pos, buff, offset, len);
810 pos += len;
811 return len;
812 }
813
814 public int read() throws IOException {
815 if (buffer == null) {
816 return -1;
817 }
818 if (pos >= buffer.length) {
819 fillBuffer();
820 if (buffer == null) {
821 return -1;
822 }
823 }
824 return buffer[pos++];
825 }
826
827 private void fillBuffer() throws IOException {
828 int len = (int) Math.min(charBuffer.length, remaining);
829 if (len > 0) {
830 len = reader.read(charBuffer, 0, len);
831 } else {
832 len = -1;
833 }
834 if (len < 0) {
835 buffer = null;
836 } else {
837 buffer = StringUtils.utf8Encode(new String(charBuffer, 0, len));
838 length += len;
839 remaining -= len;
840 }
841 pos = 0;
842 }
843
844 public long getLength() {
845 return length;
846 }
847
848 public void close() throws IOException {
849 reader.close();
850 }
851
852 }
853
854 /**
855 * Create a BLOB object.
856 *
857 * @param in the input stream
858 * @param maxLength the maximum length (-1 if not known)
859 * @return the LOB
860 */
861 public Value createBlob(InputStream in, long maxLength) {
862 if (SysProperties.LOB_IN_DATABASE) {
863 init();
864 if (conn == null) {
865 // remote connections:
866 // need to use a temp file, because the input stream could come from
867 // the same database, which would create a weird situation (trying
868 // to read a block while write something)
869 return ValueLobDb.createTempBlob(in, maxLength, handler);
870 }
871 return addLob(in, maxLength, Value.BLOB);
872 }
873 return ValueLob.createBlob(in, maxLength, handler);
874 }
875
876 /**
877 * Create a CLOB object.
878 *
879 * @param reader the reader
880 * @param maxLength the maximum length (-1 if not known)
881 * @return the LOB
882 */
883 public Value createClob(Reader reader, long maxLength) {
884 if (SysProperties.LOB_IN_DATABASE) {
885 init();
886 if (conn == null) {
887 // remote connections:
888 // need to use a temp file, because the input stream could come from
889 // the same database, which would create a weird situation (trying
890 // to read a block while write something)
891 return ValueLobDb.createTempClob(reader, maxLength, handler);
892 }
893 long max = maxLength == -1 ? Long.MAX_VALUE : maxLength;
894 CountingReaderInputStream in = new CountingReaderInputStream(reader, max);
895 ValueLobDb lob = addLob(in, Long.MAX_VALUE, Value.CLOB);
896 lob.setPrecision(in.getLength());
897 return lob;
898 }
899 return ValueLob.createClob(reader, maxLength, handler);
900 }
901
902 /**
903 * Set the table reference of this lob.
904 *
905 * @param lobId the lob
906 * @param table the table
907 */
908 public void setTable(long lobId, int table) {
909 synchronized (handler) {
910 try {
911 init();
912 String sql = "UPDATE " + LOBS + " SET TABLE = ? WHERE ID = ?";
913 PreparedStatement prep = prepare(sql);
914 prep.setInt(1, table);
915 prep.setLong(2, lobId);
916 prep.executeUpdate();
917 reuse(sql, prep);
918 } catch (SQLException e) {
919 throw DbException.convert(e);
920 }
921 }
922 }
923
924 }
925